Chapter 10 -- Partial Bounded Queue

113 views

Skip to first unread message

Munesh Singh

unread,

Mar 9, 2018, 5:59:46 AM3/9/18

to Art of Multiprocessor Programming

I am trying to understand the code given in Chapter 10 related to bounded partial queue. I have the following queries:

1. Why we have created only a single node for head and not for tail? --> Fig 10.2 Line number 9

2. I could not understand enqueue operation in the line number 24

Node e = new Node(x);
tail.next=tail; tail=e;

Why tail.next is pointing to itself?

Hope someone can clarify on this.

Munesh Singh

unread,

Mar 11, 2018, 1:43:10 AM3/11/18

to Art of Multiprocessor Programming

Well, to add to my post already here, the code in the book on Partial Bounded Queue (Fig 10.3; Line 24) emits null. When it is modified to the following, the output is correctly printed:

Node e = new Node(x);

tail.next=e; tail=e;

Correct me if my wrong whether this modification by me is ok.

I am also putting the code here for completeness:

package JavaTest.lect11.a;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BoundedQueue<T> {
	ReentrantLock enqLock,deqLock;
	Condition notEmptyCondition,notFullCondition;
	AtomicInteger size;
	volatile Node head,tail;
	int capacity;
	
	public BoundedQueue(int _capacity) {
		capacity=_capacity;
		head=new Node(null);
		tail=head;
		size=new AtomicInteger(0);
		enqLock=new ReentrantLock();
		notFullCondition=enqLock.newCondition();
		deqLock=new ReentrantLock();
		notEmptyCondition=deqLock.newCondition();
	}
	
	public void enq(T x) {
		boolean mustWakeDequeuers=false;
		enqLock.lock();
		try {
			while(size.get() == capacity)
				try {
					notFullCondition.await();
				} catch(InterruptedException ex) {}
			Node e=new Node(x);
			tail.next=e; tail=e;
			if(size.getAndIncrement()==0) 
				mustWakeDequeuers=true;
			System.out.println("size: "+size.get());
			
		} finally {
			enqLock.unlock();
		}
		if(mustWakeDequeuers) {
			deqLock.lock();
			try {
				notEmptyCondition.signalAll();
			} finally {
				deqLock.unlock();
			}
		}
	}
	
	public T deq() {
		T result;
		boolean mustWakeEnqueuers=false;
		deqLock.lock();
		try {
			while(size.get()==0) 
				try {
					notEmptyCondition.await();
				} catch(InterruptedException ex) {}
			System.out.println(head.next.value);
			result=head.next.value;
			head=head.next;
			if(size.getAndDecrement()==capacity)
				mustWakeEnqueuers=true;
			System.out.println("size: "+size.get());
			
		} finally {
			deqLock.unlock();
		}
		if(mustWakeEnqueuers) {
			enqLock.lock();
			try {
				notFullCondition.signalAll();
			} finally {
				enqLock.unlock();
			}
		}
		return result;
	}
	
	protected class Node {
		public T value;
		public volatile Node next;
		public Node(T x) {
			value=x;
			next=null;
		}
	}
}

package JavaTest.lect11.a;

public class TestThread extends Thread {
	private BoundedQueue<String> bq;
	private int i;
	public TestThread(BoundedQueue<String> bq,int i)  {
		this.bq=bq;
		this.i=i;
	}
	
	public void run() {

		bq.enq("A"+i);
		bq.enq("B"+i);
		//bq.enq("C"+i);
		//bq.enq("D"+i);

	}

}

package JavaTest.lect11.a;

public class MainDequeue {
	public static void main(String[] args) {
		BoundedQueue<String> bq=new BoundedQueue<String>(4);

		Thread t1=new TestThread(bq,1);
		Thread t2=new TestThread(bq,2);


		t1.start();
		t2.start();

		System.out.println(bq.deq());
		System.out.println(bq.deq());
	}
}