util/concurrent/FairSemaphore.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package java.util.concurrent;
import java.util.concurrent.locks.*;

/**
 * A semaphore guaranteeing that threads invoking any of the {@link
 * #acquire() acquire} methods are allocated permits in the order in
 * which their invocation of those methods was processed
 * (first-in-first-out; FIFO). Note that FIFO ordering necessarily
 * applies to specific internal points of execution within these
 * methods.  So, it is possible for one thread to invoke
 * <tt>acquire</tt> before another, but reach the ordering point after
 * the other, and similarly upon return from the method.
 *
 * <p>Because permits are allocated in-order, this class also provides
 * convenience methods to {@link #acquire(long) acquire} and {@link
 * #release(long) release} multiple permits at a time. 
 *
 * @since 1.5
 * @spec JSR-166
 * @revised $Date: 2003/07/08 00:46:33 $
 * @editor $Author: dl $
 * @author Doug Lea
 *
 */
public class FairSemaphore extends Semaphore {

    /*
     * This differs from Semaphore only in that it uses a Fair
     * ReentrantLock.  Because the Fair ReentrantLock guarantees FIFO
     * queuing, we don't need to do anything fancy to prevent
     * overtaking etc.  for the multiple-permit methods. The only
     * minor downside is that multi-permit acquires will sometimes
     * repeatedly wake up finding that they must re-wait. A custom
     * solution could minimize this, at the expense of being slower
     * and more complex for the typical case.
     */

    /**
     * Construct a <tt>FiFoSemaphore</tt> with the given number of
     * permits.
     * @param permits the initial number of permits available
     */
    public FairSemaphore(long permits) { 
        super(permits, new ReentrantLock(true)); 
    }

    /**
     * Acquires the given number of permits from this semaphore, 
     * blocking until all are available, 
     * or the thread is {@link Thread#interrupt interrupted}.
     *
     * <p>Acquires the given number of permits, if they are available,
     * and returns immediately,
     * reducing the number of available permits by the given amount.
     *
     * <p>If insufficient permits are available then the current thread becomes
     * disabled for thread scheduling purposes and lies dormant until
     * one of two things happens:
     * <ul>
     * <li>Some other thread invokes one of the {@link #release() release} 
     * methods for this semaphore, the current thread is next to be assigned
     * permits and the number of available permits satisfies this request; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread.
     * </ul>
     *
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting
     * for a permit,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared. 
     * Any available permits are assigned to the next waiting thread as if
     * they had been made available by a call to {@link #release()}.
     *
     * @param permits the number of permits to acquire
     *
     * @throws InterruptedException if the current thread is interrupted
     * @throws IllegalArgumentException if permits less than zero.
     *
     * @see Thread#interrupt
     */
    public void acquire(long permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();

        lock.lockInterruptibly();
        try {
            while (count - permits < 0) 
                available.await();
            count -= permits;
        }
        catch (InterruptedException ie) {
            available.signal();
            throw ie;
        }
        finally {
            lock.unlock();
        }
        
    }

    /**
     * Acquires the given number of permits from this semaphore, 
     * blocking until all are available.
     *
     * <p>Acquires the given number of permits, if they are available,
     * and returns immediately,
     * reducing the number of available permits by the given amount.
     *
     * <p>If insufficient permits are available then the current thread becomes
     * disabled for thread scheduling purposes and lies dormant until
     * some other thread invokes one of the {@link #release() release} 
     * methods for this semaphore, the current thread is next to be assigned
     * permits and the number of available permits satisfies this request.
     *
     * <p>If the current thread
     * is {@link Thread#interrupt interrupted} while waiting
     * for permits then it will continue to wait and its position in the
     * queue is not affected. When the
     * thread does return from this method its interrupt status will be set.
     *
     * @param permits the number of permits to acquire
     * @throws IllegalArgumentException if permits less than zero.
     *
     */
    public void acquireUninterruptibly(long permits) {
        if (permits < 0) throw new IllegalArgumentException();

        lock.lock();
        try {
            while (count - permits < 0) 
                available.awaitUninterruptibly();
            count -= permits;
        }
        finally {
            lock.unlock();
        }
    }

    /**
     * Acquires the given number of permits from this semaphore, only if 
     * all are available at the  time of invocation.
     * <p>Acquires the given number of permits, if they are available, and 
     * returns immediately, with the value <tt>true</tt>,
     * reducing the number of available permits by the given amount.
     *
     * <p>If insufficient permits are available then this method will return
     * immediately with the value <tt>false</tt> and the number of available
     * permits is unchanged.
     *
     * @param permits the number of permits to acquire
     *
     * @return <tt>true</tt> if the permits were acquired and <tt>false</tt>
     * otherwise.
     * @throws IllegalArgumentException if permits less than zero.
     */
    public boolean tryAcquire(long permits) {
        if (permits < 0) throw new IllegalArgumentException();

        lock.lock();
        try {
            if (count - permits >= 0) {
                count -= permits;
                return true;
            }
            return false;
        }
        finally {
            lock.unlock();
        }
    }

    /**
     * Acquires the given number of permits from this semaphore, if all 
     * become available within the given waiting time and the
     * current thread has not been {@link Thread#interrupt interrupted}.
     * <p>Acquires the given number of permits, if they are available and 
     * returns immediately, with the value <tt>true</tt>,
     * reducing the number of available permits by the given amount.
     * <p>If insufficient permits are available then
     * the current thread becomes disabled for thread scheduling
     * purposes and lies dormant until one of three things happens:
     * <ul>
     * <li>Some other thread invokes one of the {@link #release() release} 
     * methods for this semaphore, the current thread is next to be assigned
     * permits and the number of available permits satisfies this request; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread; or
     * <li>The specified waiting time elapses.
     * </ul>
     * <p>If the permits are acquired then the value <tt>true</tt> is returned.
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting to acquire
     * the permits,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     * Any available permits are assigned to the next waiting thread as if
     * they had been made available by a call to {@link #release()}.
     *
     * <p>If the specified waiting time elapses then the value <tt>false</tt>
     * is returned.
     * The given waiting time is a best-effort lower bound. If the time is
     * less than or equal to zero, the method will not wait at all.
     * Any available permits are assigned to the next waiting thread as if
     * they had been made available by a call to {@link #release()}.
     *
     * @param permits the number of permits to acquire
     * @param timeout the maximum time to wait for the permits
     * @param granularity the time unit of the <tt>timeout</tt> argument.
     * @return <tt>true</tt> if all permits were acquired and <tt>false</tt>
     * if the waiting time elapsed before all permits were acquired.
     *
     * @throws InterruptedException if the current thread is interrupted
     * @throws IllegalArgumentException if permits less than zero.
     *
     * @see Thread#interrupt
     *
     */
    public boolean tryAcquire(long permits, long timeout, TimeUnit granularity)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        lock.lockInterruptibly();
        long nanos = granularity.toNanos(timeout);
        try {
            for (;;) {
                if (count - permits >= 0) {
                    count -= permits;
                    return true;
                }
                if (nanos <= 0)
                    return false;
                nanos = available.awaitNanos(nanos);
            }
        }
        catch (InterruptedException ie) {
            available.signal();
            throw ie;
        }
        finally {
            lock.unlock();
        }

    }


    /**
     * Releases the given number of permits, returning them to the semaphore.
     * <p>Releases the given number of permits, increasing the number of 
     * available permits by that amount.
     * If any threads are blocking trying to acquire permits, then the
     * one that has been waiting the longest
     * is selected and given the permits that were just released.
     * If the number of available permits satisfies that thread's request
     * then that thread is re-enabled for thread scheduling purposes; otherwise
     * the thread continues to wait. If there are still permits available
     * after the first thread's request has been satisfied, then those permits
     * are assigned to the next waiting thread. If it is satisfied then it is
     * re-enabled for thread scheduling purposes. This continues until there
     * are insufficient permits to satisfy the next waiting thread, or there
     * are no more waiting threads.
     *
     * <p>There is no requirement that a thread that releases a permit must
     * have acquired that permit by calling {@link Semaphore#acquire acquire}.
     * Correct usage of a semaphore is established by programming convention
     * in the application.
     *
     * @param permits the number of permits to release
     * @throws IllegalArgumentException if permits less than zero.
     */
    public void release(long permits) {
        if (permits < 0) throw new IllegalArgumentException();
        // Even if using fair locks, releases should try to barge in.
        if (!lock.tryLock())
            lock.lock();
        try {
            count += permits;
            for (int i = 0; i < permits; ++i)
                available.signal();
        }
        finally {
            lock.unlock();
        }
    }
}
Revision:	1.5
Committed:	Wed Jul 9 23:23:17 2003 UTC (20 years, 10 months ago) by dl
Branch:	MAIN
Changes since 1.4:	+4 -2 lines
Log Message:	Misc performance tunings
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5			*/
6
7			package java.util.concurrent;
8	dl	1.4	import java.util.concurrent.locks.*;
9	dl	1.1
10			/**
11	dl	1.2	* A semaphore guaranteeing that threads invoking any of the {@link
12			* #acquire() acquire} methods are allocated permits in the order in
13			* which their invocation of those methods was processed
14			* (first-in-first-out; FIFO). Note that FIFO ordering necessarily
15			* applies to specific internal points of execution within these
16			* methods. So, it is possible for one thread to invoke
17			* <tt>acquire</tt> before another, but reach the ordering point after
18			* the other, and similarly upon return from the method.
19	dl	1.1	*
20	dl	1.2	* <p>Because permits are allocated in-order, this class also provides
21			* convenience methods to {@link #acquire(long) acquire} and {@link
22			* #release(long) release} multiple permits at a time.
23	dl	1.1	*
24			* @since 1.5
25			* @spec JSR-166
26	dl	1.5	* @revised $Date: 2003/07/08 00:46:33 $
27	dl	1.2	* @editor $Author: dl $
28	dl	1.3	* @author Doug Lea
29	dl	1.1	*
30			*/
31			public class FairSemaphore extends Semaphore {
32
33			/*
34	dl	1.4	* This differs from Semaphore only in that it uses a Fair
35			* ReentrantLock. Because the Fair ReentrantLock guarantees FIFO
36			* queuing, we don't need to do anything fancy to prevent
37	dl	1.1	* overtaking etc. for the multiple-permit methods. The only
38			* minor downside is that multi-permit acquires will sometimes
39			* repeatedly wake up finding that they must re-wait. A custom
40			* solution could minimize this, at the expense of being slower
41			* and more complex for the typical case.
42			*/
43
44			/**
45			* Construct a <tt>FiFoSemaphore</tt> with the given number of
46			* permits.
47			* @param permits the initial number of permits available
48			*/
49			public FairSemaphore(long permits) {
50	dl	1.4	super(permits, new ReentrantLock(true));
51	dl	1.1	}
52
53			/**
54			* Acquires the given number of permits from this semaphore,
55			* blocking until all are available,
56			* or the thread is {@link Thread#interrupt interrupted}.
57			*
58			* <p>Acquires the given number of permits, if they are available,
59			* and returns immediately,
60			* reducing the number of available permits by the given amount.
61			*
62			* <p>If insufficient permits are available then the current thread becomes
63			* disabled for thread scheduling purposes and lies dormant until
64			* one of two things happens:
65			* <ul>
66			* <li>Some other thread invokes one of the {@link #release() release}
67			* methods for this semaphore, the current thread is next to be assigned
68			* permits and the number of available permits satisfies this request; or
69			* <li>Some other thread {@link Thread#interrupt interrupts} the current
70			* thread.
71			* </ul>
72			*
73			* <p>If the current thread:
74			* <ul>
75			* <li>has its interrupted status set on entry to this method; or
76			* <li>is {@link Thread#interrupt interrupted} while waiting
77			* for a permit,
78			* </ul>
79			* then {@link InterruptedException} is thrown and the current thread's
80			* interrupted status is cleared.
81			* Any available permits are assigned to the next waiting thread as if
82			* they had been made available by a call to {@link #release()}.
83			*
84			* @param permits the number of permits to acquire
85			*
86			* @throws InterruptedException if the current thread is interrupted
87			* @throws IllegalArgumentException if permits less than zero.
88			*
89			* @see Thread#interrupt
90			*/
91			public void acquire(long permits) throws InterruptedException {
92			if (permits < 0) throw new IllegalArgumentException();
93
94			lock.lockInterruptibly();
95			try {
96			while (count - permits < 0)
97			available.await();
98			count -= permits;
99			}
100			catch (InterruptedException ie) {
101			available.signal();
102			throw ie;
103			}
104			finally {
105			lock.unlock();
106			}
107
108			}
109
110			/**
111			* Acquires the given number of permits from this semaphore,
112			* blocking until all are available.
113			*
114			* <p>Acquires the given number of permits, if they are available,
115			* and returns immediately,
116			* reducing the number of available permits by the given amount.
117			*
118			* <p>If insufficient permits are available then the current thread becomes
119			* disabled for thread scheduling purposes and lies dormant until
120			* some other thread invokes one of the {@link #release() release}
121			* methods for this semaphore, the current thread is next to be assigned
122			* permits and the number of available permits satisfies this request.
123			*
124			* <p>If the current thread
125			* is {@link Thread#interrupt interrupted} while waiting
126			* for permits then it will continue to wait and its position in the
127			* queue is not affected. When the
128			* thread does return from this method its interrupt status will be set.
129			*
130			* @param permits the number of permits to acquire
131			* @throws IllegalArgumentException if permits less than zero.
132			*
133			*/
134			public void acquireUninterruptibly(long permits) {
135			if (permits < 0) throw new IllegalArgumentException();
136
137			lock.lock();
138			try {
139			while (count - permits < 0)
140			available.awaitUninterruptibly();
141			count -= permits;
142			}
143			finally {
144			lock.unlock();
145			}
146			}
147
148			/**
149			* Acquires the given number of permits from this semaphore, only if
150			* all are available at the time of invocation.
151			* <p>Acquires the given number of permits, if they are available, and
152			* returns immediately, with the value <tt>true</tt>,
153			* reducing the number of available permits by the given amount.
154			*
155			* <p>If insufficient permits are available then this method will return
156			* immediately with the value <tt>false</tt> and the number of available
157			* permits is unchanged.
158			*
159			* @param permits the number of permits to acquire
160			*
161			* @return <tt>true</tt> if the permits were acquired and <tt>false</tt>
162			* otherwise.
163			* @throws IllegalArgumentException if permits less than zero.
164			*/
165			public boolean tryAcquire(long permits) {
166			if (permits < 0) throw new IllegalArgumentException();
167
168			lock.lock();
169			try {
170			if (count - permits >= 0) {
171			count -= permits;
172			return true;
173			}
174			return false;
175			}
176			finally {
177			lock.unlock();
178			}
179			}
180
181			/**
182			* Acquires the given number of permits from this semaphore, if all
183			* become available within the given waiting time and the
184			* current thread has not been {@link Thread#interrupt interrupted}.
185			* <p>Acquires the given number of permits, if they are available and
186			* returns immediately, with the value <tt>true</tt>,
187			* reducing the number of available permits by the given amount.
188			* <p>If insufficient permits are available then
189			* the current thread becomes disabled for thread scheduling
190			* purposes and lies dormant until one of three things happens:
191			* <ul>
192			* <li>Some other thread invokes one of the {@link #release() release}
193			* methods for this semaphore, the current thread is next to be assigned
194			* permits and the number of available permits satisfies this request; or
195			* <li>Some other thread {@link Thread#interrupt interrupts} the current
196			* thread; or
197			* <li>The specified waiting time elapses.
198			* </ul>
199			* <p>If the permits are acquired then the value <tt>true</tt> is returned.
200			* <p>If the current thread:
201			* <ul>
202			* <li>has its interrupted status set on entry to this method; or
203			* <li>is {@link Thread#interrupt interrupted} while waiting to acquire
204			* the permits,
205			* </ul>
206			* then {@link InterruptedException} is thrown and the current thread's
207			* interrupted status is cleared.
208			* Any available permits are assigned to the next waiting thread as if
209			* they had been made available by a call to {@link #release()}.
210			*
211			* <p>If the specified waiting time elapses then the value <tt>false</tt>
212			* is returned.
213			* The given waiting time is a best-effort lower bound. If the time is
214			* less than or equal to zero, the method will not wait at all.
215			* Any available permits are assigned to the next waiting thread as if
216			* they had been made available by a call to {@link #release()}.
217			*
218			* @param permits the number of permits to acquire
219			* @param timeout the maximum time to wait for the permits
220			* @param granularity the time unit of the <tt>timeout</tt> argument.
221			* @return <tt>true</tt> if all permits were acquired and <tt>false</tt>
222			* if the waiting time elapsed before all permits were acquired.
223			*
224			* @throws InterruptedException if the current thread is interrupted
225			* @throws IllegalArgumentException if permits less than zero.
226			*
227			* @see Thread#interrupt
228			*
229			*/
230			public boolean tryAcquire(long permits, long timeout, TimeUnit granularity)
231			throws InterruptedException {
232			if (permits < 0) throw new IllegalArgumentException();
233			lock.lockInterruptibly();
234			long nanos = granularity.toNanos(timeout);
235			try {
236			for (;;) {
237			if (count - permits >= 0) {
238			count -= permits;
239			return true;
240			}
241			if (nanos <= 0)
242			return false;
243			nanos = available.awaitNanos(nanos);
244			}
245			}
246			catch (InterruptedException ie) {
247			available.signal();
248			throw ie;
249			}
250			finally {
251			lock.unlock();
252			}
253
254			}
255
256
257
258			/**
259			* Releases the given number of permits, returning them to the semaphore.
260			* <p>Releases the given number of permits, increasing the number of
261			* available permits by that amount.
262			* If any threads are blocking trying to acquire permits, then the
263			* one that has been waiting the longest
264			* is selected and given the permits that were just released.
265			* If the number of available permits satisfies that thread's request
266			* then that thread is re-enabled for thread scheduling purposes; otherwise
267			* the thread continues to wait. If there are still permits available
268			* after the first thread's request has been satisfied, then those permits
269			* are assigned to the next waiting thread. If it is satisfied then it is
270			* re-enabled for thread scheduling purposes. This continues until there
271			* are insufficient permits to satisfy the next waiting thread, or there
272			* are no more waiting threads.
273			*
274			* <p>There is no requirement that a thread that releases a permit must
275			* have acquired that permit by calling {@link Semaphore#acquire acquire}.
276			* Correct usage of a semaphore is established by programming convention
277			* in the application.
278			*
279			* @param permits the number of permits to release
280			* @throws IllegalArgumentException if permits less than zero.
281			*/
282			public void release(long permits) {
283			if (permits < 0) throw new IllegalArgumentException();
284	dl	1.5	// Even if using fair locks, releases should try to barge in.
285			if (!lock.tryLock())
286			lock.lock();
287	dl	1.1	try {
288			count += permits;
289			for (int i = 0; i < permits; ++i)
290			available.signal();
291			}
292			finally {
293			lock.unlock();
294			}
295			}
296			}