File: QueuedSemaphore.java

package info (click to toggle)
concurrent-dfsg 1.3.4-4
  • links: PTS, VCS
  • area: main
  • in suites: buster, jessie, jessie-kfreebsd, squeeze, stretch, wheezy
  • size: 976 kB
  • ctags: 2,018
  • sloc: java: 10,704; xml: 49; makefile: 12
file content (171 lines) | stat: -rw-r--r-- 4,864 bytes parent folder | download | duplicates (3)
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;
            }
          }
        }
      }
    }

  }


}