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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261

package com.holub.asynch;


//import java.util.LinkedList; NEB: I use my own, annotated LinkedList.
import java.util.NoSuchElementException;

import edu.cmu.cs.crystal.annotations.UseAnalyses;
import edu.cmu.cs.plural.annot.ClassStates;
import edu.cmu.cs.plural.annot.FalseIndicates;
import edu.cmu.cs.plural.annot.Full;
import edu.cmu.cs.plural.annot.Perm;
import edu.cmu.cs.plural.annot.PluralAnalysis;
import edu.cmu.cs.plural.annot.Pure;
import edu.cmu.cs.plural.annot.Pures;
import edu.cmu.cs.plural.annot.Refine;
import edu.cmu.cs.plural.annot.Share;
import edu.cmu.cs.plural.annot.State;
import edu.cmu.cs.plural.annot.States;
import edu.cmu.cs.plural.annot.TrueIndicates;
import edu.cmu.cs.plural.annot.Use;

/***********************************************************************
 *  NEB: Note! This class is used with permission of Allen I. Holub.
 *  We are extremely grateful for his permission!
 *  
 *  This is a thread-safe queue that blocks automatically if you
 *	try to dequeue from an empty queue. It's based on a linked list,
 *  so will never fill. (You'll never block on a queue-full condition
 *	because there isn't one.)
 *
 *	<p>
 *	This class uses the <code>LinkedList</code> class, introduced into
 *	the JDK at version 1.2. It will not work with earlier releases.
 *
 * <br><br>
 * <table border=1 cellspacing=0 cellpadding=5><tr><td><font size=-1><i>
 * <center>(c) 1999, Allen I. Holub.</center>
 * <p>
 * This code may not be distributed by yourself except in binary form,
 * incorporated into a java .class file. You may use this code freely
 * for personal purposes, but you may not incorporate it into any
 * commercial product without express permission of Allen I. Holub in
 * writing.
 * </td></tr></table>
 *
 * @author Allen I. Holub
 */
@Refine({
	@States(dim="STRUCTURE", value={"STRUCTURESTATE"}),
	@States(dim="PROTOCOL", value= {"CLOSED", "OPEN"})
})
@ClassStates({
	@State(name="STRUCTURE", 
			// NEB: Of course waiting_threads cannot be null, but we needed to map it
			// to the STRUCTURE dimension, and there was no other way to do this.
		   inv="share(elements) * reject_enqueue_requests == true => full(this,PROTOCOL) in OPEN"),
	@State(name="OPEN", inv="closed == false"), 
	@State(name="CLOSED", inv="closed == true")
})
//@PassingTest
@UseAnalyses(PluralAnalysis.NIMBY)
public class Blocking_queue
{
	private LinkedList elements 				= new LinkedList();
	private boolean	   closed					= false;
	private boolean	   reject_enqueue_requests	= false;
	private int		   waiting_threads			= 0;

	// NEB: Original class had no constructor
	@Perm(ensures="unique(this!fr) in OPEN,STRUCTURESTATE")
	public Blocking_queue() {
	}
	
	/*******************************************************************
	 * The Closed exception is thrown if you try to used an explicitly
	 * closed queue. See {@link #close}.
	 */

	public class Closed extends RuntimeException
	{	
        // Unique SerialVersionUID (should NEVER change)
        static final long serialVersionUID = 0L;
        
        @Perm(ensures="unique(this!fr)")
        private Closed()
		{	super("Tried to access closed Blocking_queue");
		}
	}

	/*******************************************************************
	 *	Enqueue an object
	 **/
	@Share(value="STRUCTURE")
	@Full(requires="OPEN", ensures="OPEN", value="PROTOCOL")
	public synchronized final void enqueue( Object new_element )
														throws Closed
	{	if( closed || reject_enqueue_requests )
			throw new Closed();
		elements.addLast( new_element );
		notify(); 										//#notify
	}

	/*******************************************************************
	 * Enqueue an item, and thereafter reject any requests to enqueue
	 * additional items. The queue is closed automatically when the
	 * final item is dequeued.
	 */
	@Perm(requires="full(this,PROTOCOL) in OPEN * share(this,STRUCTURE)"
	//,ensures="pure(this,PROTOCOL) in CLOSING" NEB: We could do this...
	// It   
	)
	public synchronized final void enqueue_final_item( Object new_element )	//#final.start
														throws Closed
	{	
		enqueue( new_element );
		reject_enqueue_requests = true;
	}															//#final.end

	/*******************************************************************
	 *	Dequeues an element; blocks if the queue is empty
	 *	(until something is enqueued). Be careful of nested-monitor
	 *  lockout if you call this function. You must ensure that
	 *	there's a way to get something into the queue that does
	 *  not involve calling a synchronized method of whatever
	 *  class is blocked, waiting to dequeue something. A timeout is
	 *	not supported because of a potential race condition (see text).
	 *  You can {@link Thread#interrupt interrupt} the dequeueing thread
	 *  to break it out of a blocked dequeue operation, however.
	 *
	 *  @see #enqueue
	 *  @see #drain
	 *  @see #nonblocking_dequeue
	 *	@return the dequeued object or null if the wait timed out and
	 *			nothing was dequeued.
	 */
	@Perm(requires="share(this!fr,STRUCTURE) * pure(this!fr,PROTOCOL) in OPEN",
		  ensures="share(this!fr,STRUCTURE) * pure(this!fr,PROTOCOL)")
	public synchronized final Object dequeue( ) 
									throws InterruptedException, Closed
	{	
		atomic: { // NEB: Added atomic block
			if( closed )
				throw new Closed();
			try
			{	
				// If the queue is empty, wait. I've put the spin lock inside
				// an if so that the waiting_threads count doesn't jitter
				// while inside the spin lock. A thread is not considered to
				// be done waiting until it's actually acquired an element
	
				if( elements.size() <= 0 )
				{	++waiting_threads;
					//if( elements.size() <= 0 ) NEB: Used for testing
					while( elements.size() <= 0 )
					{	wait();						//#wait
						if( closed )
						{	--waiting_threads;
							throw new Closed();
						}
					}
					--waiting_threads;
				}
	
				Object head = elements.removeFirst();
				if( elements.size() == 0 && reject_enqueue_requests ) {
					reject_enqueue_requests = false; // NEB: New code
					close(); // just removed final item, close the queue.
					return head; // NEB: New code
				}
	
				return head;
			}
			catch( NoSuchElementException e )	// Shouldn't happen
			{	throw new Error(
						"Internal error (com.holub.asynch.Blocking_queue)");
			}
		}
	}
	
	/*******************************************************************
	 *	The is_empty() method is inherently unreliable in a
	 *  multithreaded situation. In code like the following,
	 *	it's possible for a thread to sneak in after the test but before
	 *	the dequeue operation and steal the element you thought you
	 *	were dequeueing.
	 *	<PRE>
	 *	Blocking_queue queue = new Blocking_queue();
	 *	//...
	 *	if( !some_queue.is_empty() )
	 *		some_queue.dequeue();
	 *	</PRE>
	 *	To do the forgoing reliably, you must synchronize on the
	 *	queue as follows:
	 *	<PRE>
	 *	Blocking_queue queue = new Blocking_queue();
	 *	//...
	 *	synchronized( queue )
	 *	{   if( !some_queue.is_empty() )
	 *			some_queue.dequeue();
	 *	}
	 *	</PRE>
	 *	The same effect can be achieved if the test/dequeue operation
	 *	is done inside a synchronized method, and the only way to
	 *	add or remove queue elements is from other synchronized
	 *	methods.
	 */
	 @Pures({
		 @Pure("STRUCTURE"),
		 @Pure(requires="OPEN",value="PROTOCOL")
	 })
	 public final boolean is_empty()
	 {	
		 atomic: { // NEB: Added atomic block
		 	return elements.size() <= 0;
	 	 }
	 }

	/*******************************************************************
	 * Return the number of threads waiting for a message on the
	 * current queue. See {@link #is_empty} for warnings about
	 * synchronization.
	 */
     @Pure("STRUCTURE")
	 public final int waiting_threads()
	 {	return waiting_threads;
	 }
	 	
     
     // NEB: I added the is_closed() method... because there wasn't one. :-(
     @Pure(use = Use.FIELDS,value="PROTOCOL")
     @TrueIndicates("CLOSED")
     @FalseIndicates("OPEN")
     public final synchronized boolean is_closed() {
    	 atomic: { // NEB: Added atomic block
    	 	return closed;
     	 }
     }

	/*******************************************************************
	 * Close the blocking queue. All threads that are blocked
	 * [waiting in dequeue() for items to be enqueued] are released.
	 * The {@link dequeue()} call will throw a {@link Blocking_queue.Closed}
	 * runtime
	 * exception instead of returning normally in this case.
	 * Once a queue is closed, any attempt to enqueue() an item will
	 * also result in a Blocking_queue.Closed exception toss.
	 *
	 * The queue is emptied when it's closed, so if the only references
	 * to a given object are those stored on the queue, the object will
	 * become garbage collectable.
	 */
     @Full(use = Use.FIELDS,value="PROTOCOL",requires="OPEN",ensures="CLOSED")
	 public synchronized void close()
	 {	
    	 atomic : { // NEB: Added atomic block
    	 	closed 	 = true;
    	 	elements = null;
    	 	notifyAll();
     	}
	 }
}