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