/**
 * Fifo is a thread-safe and efficient implementation of a First in-First
 * out queue.
 *
 * <h4>Description</h4>
 *
 * This is an implementation of a generic unbounded first-in/first-out
 * queue. There are two locks for the entire list which seperate the
 * synchronisation to the put and take methods.
 *
 * The first lock is associated with the head of the list and the second
 * protects insertions at the end of the list. After each
 * take, the previous node will become the header.
 *
 * <h4>Notes</h4>
 *
 * This example illustrates the use of a Lock Splitting technique
 * to control entry into a linked list of objects. It treads the
 * middle ground by avoiding fully synchronising the entry classes
 * and fully synchronising all the linked node objects being controlled,
 * thereby avoiding both concurrency and liveness issues.
 *
 * It is, however, an illustration of this technique rather than
 * a practical implementation which would usually be done using a
 * Vector.
 *
 * <h4>References</h4>
 *
 * This example has been adapted from the <i>Concurrent Programming in Java</i>
 * by Doug Lea (ISBN 0-201-69581-2) where it was called TwoLockQueue
 *
 * Jones. Cliff. "Accomodating Interference in the Formal Design of
 *       Concurrent Object-Based Programs", <I>Formal Methods in System
 *       Design</i>, March 1996
 *
 * @author      Source Code Documentation Templates
 *
 * @version     %I%, %G%
 *
 * @see         java.util.Vector
 */

package UK.co.demon.xinit.algorithms;

public class Fifo extends Thread
{
  private FifoNode head_;
  private FifoNode last_;
  private int count_;

  /**
   * Used as a lock for the last object on the list
   */
  private Object lockOnLast_;

  public  Fifo()
  {
    head_ = last_ = new FifoNode(null, null);
    lockOnLast_ = new Object();
    count_ = 0;
  }

 /**
  * Put an item onto the queue using the lockOnLast_ object
  * as a lock rather than <CODE>this)</CODE>
  *
  * @param	x 	The object you are removing from the list
  *
  * @return	void
  *
  * @see	#take
  */

  public void put(Object x)
  {
    FifoNode newNode = new FifoNode(x, null);

    synchronized (lockOnLast_)
    {
      last_.next = newNode;
      last_ = newNode;
      count_ += 1;
    }
  }

 /**
  * Take an item from the fifo using the lock on <CODE>this</CODE>
  *
  * @param	x 	The object you are removing from the fifo
  *
  * @return	Object  The object you have poped
  *
  * @see	#put
  */

  public synchronized Object take()
  {
    Object x = null;

    FifoNode first = head_.next;

    if (first != null)
    {
      x = first.value;
      head_ = first;
      count_ -= 1;
    }
    return x;
  }

  public int getCount()
  {
    return count_;
  }

  public static void main(String[] argv)
  {
    new FifoTestController(10);
  }
}

/**
 * FifoNode class represents individual nodes on a linked list
 *
 * @see UK.co.demon.xinit.algorithms.Fifo
 */

final class FifoNode
{
  Object value;
  FifoNode next;

  FifoNode(Object x, FifoNode n)
  {
    value = x;
    next = n;
  }
}

/*
 * This is a multi-threaded test class for the Fifo class.
 */

class FifoTestController extends Thread
{
  public static Fifo onTest = null;

  FifoTestController()
  {
    if (this.onTest == null)
    {
      System.err.println("Intialising Fifo Object\n");
      this.onTest = new Fifo();
    }
  }

  FifoTestController(int threads)
  {
    this(threads, threads);
  }

  FifoTestController(int Readers, int Writers)
  {
    for( int i = 0; i < Writers; i++)
    {
      new FifoTest("Writer " + i).start();
      try { Thread.sleep(100); } catch (InterruptedException e) {};
      System.out.println("Started New Writer " + i);
    }

    for( int i = 0; i < Readers; i++)
    {
      new FifoTest("Reader " + i).start();
      try { Thread.sleep(100); } catch (InterruptedException e) {};
      System.out.println("Started New Reader " + i);
    }
  }
}

class FifoTest extends FifoTestController
{
  String currentObject;
  public final static int WRITER = 1;
  public final static int READER = 2;
  private int type = WRITER;

  FifoTest(String name)
  {
    currentObject = "Instance of " + name + " " + this.hashCode();

    if(currentObject.indexOf("Reader") != -1)
    {
      type = READER;

      System.err.println("Initialised Reader Thread");
    }
    else
    {
      System.err.println("Initialised Reader Thread");
    }
  }

  public void run()
  {
    for(;;)
    {
      if(type == READER)
      {
         Object took = onTest.take();

         System.err.println("Took [" + (String)took + "]");
      }
      else
      {
         onTest.put(this.currentObject);
         System.err.println("Put [" + currentObject + "]");
      }

      System.out.println("Objects [ " + onTest.getCount() + "]");

      try { sleep(1000); } catch (InterruptedException e) {};
      yield();
    }
  }
}

Author: Graeme Burnett

Draft

Last Updated :