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;

/**
 * A semaphore that guarantees 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).
 *
 * <p>This class introduces a fairness guarantee that can be useful
 * in some contexts.
 * However, it needs to be realized that the order in which invocations are
 * processed can be different from the order in which an application perceives
 * those invocations to be processed. Effectively, when no permit is available
 * each thread is stored in a FIFO queue. As permits are released, they
 * are allocated to the thread at the head of the queue, and so the order
 * in which threads enter the queue, is the order in which they come out.  
 * This order need not have any relationship to the order in which requests 
 * were made, nor the order in which requests actually return to the caller as 
 * these depend on the underlying thread scheduling, which is not guaranteed 
 * to be predictable or fair.
 *
 * <p>As 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/01/29 07:08:21 $
 * @editor $Author: dholmes $
 *
 */
public class FairSemaphore extends Semaphore {

    /*
     * This differs from Semaphore only in that it uses a
     * FairReentrantLock.  Because the FairReentrantLock 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 FairReentrantLock()); 
    }

    /**
     * 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();
        lock.lock();
        try {
            count += permits;
            for (int i = 0; i < permits; ++i)
                available.signal();
        }
        finally {
            lock.unlock();
        }
    }
        

}


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