Author: jason.greene(a)jboss.com Date: 2007-09-25 16:55:20 -0400 (Tue, 25 Sep 2007) New Revision: 4506 Modified: core/trunk/src/main/java/org/jboss/cache/CacheImpl.java core/trunk/src/main/java/org/jboss/cache/util/ThreadGate.java Log: Fix JBCACHE-1173 - ThreadGate does not implement timeouts properly Modified: core/trunk/src/main/java/org/jboss/cache/CacheImpl.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/CacheImpl.java 2007-09-25 16:07:25 UTC (rev 4505) +++ core/trunk/src/main/java/org/jboss/cache/CacheImpl.java 2007-09-25 20:55:20 UTC (rev 4506) @@ -2104,7 +2104,8 @@ if (channel.flushSupported()) { - flushBlockGate.await(configuration.getStateRetrievalTimeout()); + if (! flushBlockGate.await(configuration.getStateRetrievalTimeout())) + throw new TimeoutException("State retrieval timed out waiting for flush unblock."); } rsps = rspFilter == null ? disp.callRemoteMethods(validMembers, method_call, modeToUse, timeout, buddyManager != null && buddyManager.isEnabled()) Modified: core/trunk/src/main/java/org/jboss/cache/util/ThreadGate.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/util/ThreadGate.java 2007-09-25 16:07:25 UTC (rev 4505) +++ core/trunk/src/main/java/org/jboss/cache/util/ThreadGate.java 2007-09-25 20:55:20 UTC (rev 4506) @@ -1,46 +1,119 @@ +/* +* JBoss, Home of Professional Open Source +* Copyright 2005, JBoss Inc., and individual contributors as indicated +* by the @authors tag. See the copyright.txt in the distribution for a +* full listing of individual contributors. +* +* This is free software; you can redistribute it and/or modify it +* under the terms of the GNU Lesser General Public License as +* published by the Free Software Foundation; either version 2.1 of +* the License, or (at your option) any later version. +* +* This software is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +* Lesser General Public License for more details. +* +* You should have received a copy of the GNU Lesser General Public +* License along with this software; if not, write to the Free +* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA +* 02110-1301 USA, or see the FSF site: http://www.fsf.org. +*/ package org.jboss.cache.util; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + /** - * Copyright (c) 2005 Brian Goetz and Tim Peierls - * Released under the Creative Commons Attribution License - * (http://creativecommons.org/licenses/by/2.5) - * Official home: http://www.jcip.net - * - * ThreadGate <p/> Recloseable gate using wait and notifyAll - * - * @author Brian Goetz and Tim Peierls + * A reclosable gate with timeout support. + * + * @author Jason T. Greene */ +public class ThreadGate +{ + private final ReentrantLock lock = new ReentrantLock(); + private final Condition condition = lock.newCondition(); -public class ThreadGate { - // CONDITION-PREDICATE: opened-since(n) (isOpen || generation>n) - private boolean isOpen; + private boolean open; + private int sequence; - private int generation; + /** + * Open the gate. + */ + public void open() + { + lock.lock(); + try + { + open = true; + sequence++; + condition.signalAll(); + } + finally + { + lock.unlock(); + } + } - public synchronized void close() - { - isOpen = false; - } + /** + * Close the gate. + */ + public void close() + { + lock.lock(); + try + { + open = false; + } + finally + { + lock.unlock(); + } + } - public synchronized void open() - { - ++generation; - isOpen = true; - notifyAll(); - } + /** + * Waits for the gate to open. + * + * @throws InterruptedException if this thread is interrupted + */ + public void await() throws InterruptedException + { + lock.lock(); + try + { + int snapshot = sequence; + while (!open && snapshot == sequence) + condition.await(); + } + finally + { + lock.unlock(); + } + } - // BLOCKS-UNTIL: opened-since(generation on entry) - public synchronized void await() throws InterruptedException - { - int arrivalGeneration = generation; - while(!isOpen && arrivalGeneration == generation) - wait(); - } - - // BLOCKS-UNTIL: opened-since(generation on entry) - public synchronized void await(long timeout) throws InterruptedException - { - int arrivalGeneration = generation; - while(!isOpen && arrivalGeneration == generation) - wait(timeout); - } + /** + * Waits for the gate to open or the specified time to elapse. + * + * @param time the maximum time in milliseconds to wait. + * @return false if gate timeout occurred + * @throws InterruptedException if this thread is interrupted + */ + public boolean await(long time) throws InterruptedException + { + lock.lock(); + try + { + int snapshot = sequence; + boolean success = true; + while (!open && snapshot == sequence && success) + success = condition.await(time, TimeUnit.MILLISECONDS); + + return success; + } + finally + { + lock.unlock(); + } + } } \ No newline at end of file