Overview

	Chapter 18 - First-In-First-Out (FIFO) Queue
	Java Thread Programming
	Paul Hyde


Simple Implementation in Java: SimpleObjectFIFO

The class SimpleObjectFIFO shows how this model can be implemented in Java. This version holds references to objects as its item type. Synchronization is used to ensure that two or more threads can safely interact with the FIFO queue (see Chapter 7, Concurrent Access to Objects and Variables ). The wait-notify mechanism is used to signal between threads when the state of the FIFO queue changes (see Chapter 8, Inter-thread Communication ). Whenever an item is added or removed, notifyAll() is used to signal any and all waiting threads that the state of the FIFO queue has changed.

When one thread tries to add an item to a full queue, it blocks until the queue is no longer full. It does this by entering a wait state (which implicitly releases the lock) to allow another thread to come along and remove an item. When the other thread removes an item, it signals the waiting thread by invoking notifyAll() . The waiting thread then wakes up, reacquires the lock, returns from wait() , and completes the request to add the item.


Similarly, when one thread tries to remove an item from an empty queue, it blocks until the queue is no longer empty. It enters a wait state to allow another thread to come along and add an item. Listing 18.1 shows the code for SimpleObjectFIFO .


Listing 18.1 SimpleObjectFIFO.javaA Simple FIFO Queue That Uses an Array Internally


1: public class SimpleObjectFIFO extends Object {

2: private Object[] queue;

3: private int capacity;

4: private int size ;

5: private int head;

6: private int tail;

8: public SimpleObjectFIFO(int cap) {

9: capacity = (cap > 0) ? cap : 1; // at least 1

10: queue = new Object[capacity];

11: head = 0;

12: tail = 0;

13: size = 0;

14: }

15:

16: public synchronized int getSize() {

17: return size;

18: }

19:

20: public synchronized boolean isFull() {

21: return (size == capacity);

22: }

23:

24: public synchronized void add(Object obj) throws InterruptedException {

25: while (isFull()) {

26: wait();

27: }

28:

29: queue[head] = obj;

30: head = (head + 1) % capacity;

31: size++;

32: c

33: notifyAll(); // let any waiting threads know about change

34: }

35:

36: public synchronized Object remove() throws InterruptedException {

37: while (size == 0) {

38: wait();

39: }

40:

41: Object obj = queue[tail];

42: queue[tail] = null; // dont block GC by keeping reference

43: tail = (tail + 1) % capacity;

44: size;

45:

46: notifyAll(); // let any waiting threads know about change

47:

48: return obj;

49: }

50:

51: public synchronized void printState() {

52: StringBuffer sb = new StringBuffer();

53:

54: sb.append(SimpleObjectFIFO:\n);

55: sb.append( capacity= + capacity + \n);

56:

57: sb.append( size= + size);

58: if (isFull()) {

59: sb.append(- FULL);

60: } else if (size == 0) {

61: sb.append(- EMPTY);

62: }

63: sb.append(\n);

64:

65: sb.append( head= + head + \n);

66: sb.append( tail= + tail + \n);

67:

68: for (int i = 0; i < queue.length; i++) {

69: sb.append( queue[ + i + ]= + queue[i] + \n);

70: }

71: c

72: System.out.print(sb);

73: }

74: }

SimpleObjectFIFO has several member variables. None of them are marked as volatile because all accesses and modifications to them occur within synchronized methods . The private member variable queue is an array of Object s used to hold the references to objects added (line 2). The length of this array is held in capacity (line 3). The current number of items in the FIFO queue is held in size (line 4). The index into queue for the next addition is held in head (line 5). The index into queue for the next item to be removed is held in tail (line 6).


The constructor (lines 814) takes as its only argument the capacity of the new FIFO queue. If the indicated capacity is less than 1 , a minimum of 1 is silently used (line 9). An array is allocated (line 10), and the head and tail pointers and size are all set to (lines 1013).


The getSize() method is synchronized to ensure that the most recent value for size is returned (lines 1618). The isFull() method is synchronized to ensure that nothing changes while the full determination is being made (lines 2022). The queue is considered to be full if the current size is equal to the capacity .

The add() method c (lines 2434) also must be synchronized for multithread safety. Because add() could potentially have to wait if the FIFO queue is currently full, it declares that it might throw an InterruptedException (line 24). This exception is thrown if the thread is interrupted before it has a chance to add the item to the queue. The while loop (lines 2527) is used to keep the thread from proceeding if the FIFO queue is currently full. It must be a while loop because if more than one thread is trying to simultaneously add items, the following statements would not always work:


if (isFull()) { // dangerous to use ˜if instead of ˜while

wait();

}

If two threads are trying to add an item, and a third thread removes one item, only one of the two trying to add may proceed. The notifyAll() method is used instead of notify() so that all waiting threads return from wait() to check out the changed conditions (this is a deliberate choice to support the notification of threads that dont want to add or remove items, but simply want to know about the changes to the FIFO queue). If only one slot is available, only one of the threads waiting to add an item should proceed. This is why a while loop is necessary to ensure thatalthough the thread was notified and returned from wait() it wont proceed unless the condition it was waiting on is still being met.

When the thread trying to add an item gets to line 29, at least one space is available in the FIFO queue for its item. The reference to the item is assigned to the slot where head is currently pointing (line 29). Next, head is incremented to point to the next cell . If head is already pointing to the last array position, it will be wrapped around to point to the first array position (line 30). Because an item has been added, size is incremented to reflect this change (line 31). Just before returning from add() , any and all waiting threads are notified of the change (line 33).

The remove() method (lines 3649) also must be synchronized for multithread safety. Because remove() could potentially have to wait if the FIFO queue is currently empty, it declares that it might throw an InterruptedException (line 36). This exception is thrown if the thread is interrupted before it has a chance to remove an item from the queue. The while loop (lines 3739) is used to keep the thread from proceeding if the FIFO queue is currently empty. The FIFO queue is considered empty as long as its size is . A while loop is also necessary here (for generally the same reasons that it was necessary inside add() ), in case two threads are simultaneously blocked waiting to remove an item.

When the thread trying to remove an item gets to line 41, at least one item is available for removal in the FIFO queue. The reference to this item is stored in obj (line 41) so that it can be returned at the end of the method. Next, the slot in the FIFO queue has its value set to null (line 42) to ensure that the queue doesnt keep any unnecessary references to items that have been removed. If this were not done, its possible that an item would not be available for garbage collection solely because of an outdated reference held in the FIFO queue. After this, tail is incremented to point to the next cell. If tail is already pointing to the last array position, it will be wrapped around to point to the first array position (line 43). Because an item has been removed, size is reduced to reflect this change (line 44). Just before returning from remove() , any and all waiting threads are notified of the change (line 46). Finally, the reference to the item just removed is returned to the caller (line 48).


The printState() method (lines 5173) cis used to print out the values of all the member variables and the current contents of the queue. It is synchronized to ensure that nothing changes during printing. This method isnt really part of a FIFO queue implementation, but I include it here to show you the internal values of the FIFO queue at various points in time.


The class SimpleObjectFIFOTest , in Listing 18.2, isc used to show how SimpleObjectFIFO works.


Listing 18.2 SimpleObjectFIFOTest.javaCode to Demonstrate SimpleObjectFIFO


1: public class SimpleObjectFIFOTest extends Object {

2: public static void main(String[] args) {

3: try {

4: SimpleObjectFIFO fifo = new SimpleObjectFIFO(5);

5: fifo.printState();

7: fifo.add(S01);

8: fifo.printState();

10: fifo.add(S02);

11: fifo.printState();

12:

13: fifo.add(S03);

14: fifo.printState();

15:

16: Object obj = fifo.remove();

17: System.out.println(just removed obj= + obj);

18: fifo.printState();

19:

20: fifo.add(S04);

21: fifo.printState();

22:

23: fifo.add(S05);

24: fifo.printState();

25:

26: fifo.add(S06);

27: fifo.printState();

28: } catch (InterruptedException x) {

29: x.printStackTrace();

30: }

31: }

32: }


SimpleObjectFIFOTest simply performs the steps shown in Figures 18.3 and 18.4 and invokes printState() after each change. When cit is run, it produces the output shown in Listing 18.3.


Listing 18.3 Output from SimpleObjectFIFOTest


1: SimpleObjectFIFO:

2: capacity=5

3: size=0 - EMPTY

4: head=0

5: tail=0

6: queue[0]=null

7: queue[1]=null

8: queue[2]=null

9: queue[3]=null

10: queue[4]=null

11: SimpleObjectFIFO:

12: capacity=5

13: size=1

14: head=1

15: tail=0

16: queue[0]=S01

17: queue[1]=null

18: queue[2]=null

19: queue[3]=null

20: queue[4]=null

21: SimpleObjectFIFO:

22: capacity=5

23: size=2

24: head=2

25: tail=0

26: queue[0]=S01

27: queue[1]=S02

28: queue[2]=null

29: queue[3]=null

30: queue[4]=null

31: SimpleObjectFIFO:

32: capacity=5

33: size=3

34: head=3c

35: tail=0

36: queue[0]=S01

37: queue[1]=S02

38: queue[2]=S03

39: queue[3]=null

40: queue[4]=null

41: just removed obj=S01

42: SimpleObjectFIFO:

43: capacity=5

44: size=2

45: head=3

46: tail=1

47: queue[0]=null

48: queue[1]=S02

49: queue[2]=S03

50: queue[3]=null

51: queue[4]=null

52: SimpleObjectFIFO:

53: capacity=5

54: size=3

55: head=4

56: tail=1

57: queue[0]=null

58: queue[1]=S02

59: queue[2]=S03

60: queue[3]=S04

61: queue[4]=null

62: SimpleObjectFIFO:

63: capacity=5

64: size=4

65: head=0

66: tail=1

67: queue[0]=null

68: queue[1]=S02

69: queue[2]=S03

70: queue[3]=S04

71: queue[4]=S05

72: SimpleObjectFIFO:

73: capacity=5

74: size=5 - FULL

75: head=1

76: tail=1

77: queue[0]=S06

78: queue[1]=S02

79: queue[2]=S03

80: queue[3]=S04

81: queue[4]=S05

You will cnotice that capacity remains constant throughout but that size moves up and down as items are added and removed from the FIFO queue. In particular, take note of what happens when S01 is removed (line 41). The FIFO queues size decreases from 3 to 2 , and the tail pointer moves from cell to cell 1 . Also, take special note that the reference that was held to S01 is set to null (line 47). You should compare the output here with Figures 18.3 and 18.4 and notice that they correspond closely with one another.


Toc

Chapter 18 - First-In-First-Out (FIFO) Queue