Mercurial > hg > blitz_stable

comparison src/EDU/oswego/cs/dl/util/concurrent/QueuedSemaphore.java @ 0:3dc0c5604566

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Initial checkin of blitz 2.0 fcs - no installer yet.
author Dan Creswell <dan.creswell@gmail.com>
date Sat, 21 Mar 2009 11:00:06 +0000
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:3dc0c5604566
1 /*
2 File: QueuedSemaphore.java
3
4 Originally written by Doug Lea and released into the public domain.
5 This may be used for any purposes whatsoever without acknowledgment.
6 Thanks for the assistance and support of Sun Microsystems Labs,
7 and everyone contributing, testing, and using this code.
8
9 History:
10 Date Who What
11 11Jun1998 dl Create public version
12 5Aug1998 dl replaced int counters with longs
13 24Aug1999 dl release(n): screen arguments
14 */
15
16
17 package EDU.oswego.cs.dl.util.concurrent;
18
19 /**
20 * Abstract base class for semaphores relying on queued wait nodes.
21 * <p>[<a href="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html"> Introduction to this package. </a>]
22 **/
23
24
25 public abstract class QueuedSemaphore extends Semaphore {
26
27 protected final WaitQueue wq_;
28
29 QueuedSemaphore(WaitQueue q, long initialPermits) {
30 super(initialPermits);
31 wq_ = q;
32 }
33
34 public void acquire() throws InterruptedException {
35 if (Thread.interrupted()) throw new InterruptedException();
36 if (precheck()) return;
37 WaitQueue.WaitNode w = new WaitQueue.WaitNode();
38 w.doWait(this);
39 }
40
41 public boolean attempt(long msecs) throws InterruptedException {
42 if (Thread.interrupted()) throw new InterruptedException();
43 if (precheck()) return true;
44 if (msecs <= 0) return false;
45
46 WaitQueue.WaitNode w = new WaitQueue.WaitNode();
47 return w.doTimedWait(this, msecs);
48 }
49
50 protected synchronized boolean precheck() {
51 boolean pass = (permits_ > 0);
52 if (pass) --permits_;
53 return pass;
54 }
55
56 protected synchronized boolean recheck(WaitQueue.WaitNode w) {
57 boolean pass = (permits_ > 0);
58 if (pass) --permits_;
59 else wq_.insert(w);
60 return pass;
61 }
62
63
64 protected synchronized WaitQueue.WaitNode getSignallee() {
65 WaitQueue.WaitNode w = wq_.extract();
66 if (w == null) ++permits_; // if none, inc permits for new arrivals
67 return w;
68 }
69
70 public void release() {
71 for (;;) {
72 WaitQueue.WaitNode w = getSignallee();
73 if (w == null) return; // no one to signal
74 if (w.signal()) return; // notify if still waiting, else skip
75 }
76 }
77
78 /** Release N permits **/
79 public void release(long n) {
80 if (n < 0) throw new IllegalArgumentException("Negative argument");
81
82 for (long i = 0; i < n; ++i) release();
83 }
84
85 /**
86 * Base class for internal queue classes for semaphores, etc.
87 * Relies on subclasses to actually implement queue mechanics
88 **/
89
90 protected static abstract class WaitQueue {
91
92 protected abstract void insert(WaitNode w);// assumed not to block
93 protected abstract WaitNode extract(); // should return null if empty
94
95 protected static class WaitNode {
96 boolean waiting = true;
97 WaitNode next = null;
98
99 protected synchronized boolean signal() {
100 boolean signalled = waiting;
101 if (signalled) {
102 waiting = false;
103 notify();
104 }
105 return signalled;
106 }
107
108 protected synchronized boolean doTimedWait(QueuedSemaphore sem,
109 long msecs)
110 throws InterruptedException {
111 if (sem.recheck(this) || !waiting)
112 return true;
113 else if (msecs <= 0) {
114 waiting = false;
115 return false;
116 }
117 else {
118 long waitTime = msecs;
119 long start = System.currentTimeMillis();
120
121 try {
122 for (;;) {
123 wait(waitTime);
124 if (!waiting) // definitely signalled
125 return true;
126 else {
127 waitTime = msecs - (System.currentTimeMillis() - start);
128 if (waitTime <= 0) { // timed out
129 waiting = false;
130 return false;
131 }
132 }
133 }
134 }
135 catch(InterruptedException ex) {
136 if (waiting) { // no notification
137 waiting = false; // invalidate for the signaller
138 throw ex;
139 }
140 else { // thread was interrupted after it was notified
141 Thread.currentThread().interrupt();
142 return true;
143 }
144 }
145 }
146 }
147
148 protected synchronized void doWait(QueuedSemaphore sem)
149 throws InterruptedException {
150 if (!sem.recheck(this)) {
151 try {
152 while (waiting) wait();
153 }
154 catch(InterruptedException ex) {
155 if (waiting) { // no notification
156 waiting = false; // invalidate for the signaller
157 throw ex;
158 }
159 else { // thread was interrupted after it was notified
160 Thread.currentThread().interrupt();
161 return;
162 }
163 }
164 }
165 }
166 }
167
168 }
169
170
171 }