Thread Coordination & Communication

package unsynchronized;

/**
 *  NOTE: This implementation of the SteamBoiler class is implemented
 *  without regard to synchronization.  Because of this, it fails
 *  to maintain boiler pressure below the safety limit.
 */
public class SteamBoiler
{
  public static final int PRESSURE_LIMIT = 20;
  public static final int BURNER_COUNT = 10;
  
  public Burner[] start()
  {
    Burner[] burners = new Burner[BURNER_COUNT];
    for (int burner = 0; burner < BURNER_COUNT; burner++)
    {
      burners[burner] = new Burner(this);
      burners[burner].start();
    }
    
    return burners;
  }
  
  public int getPressure()
  {
    return m_pressure;
  }
  
  public void setPressure(int pressure)
  {
    m_pressure = pressure;
  }
  
  public static void main(String[] args)
  {
    SteamBoiler boiler = new SteamBoiler();
    
    Burner[] burners = boiler.start();
    
    for (int burner = 0; burner < BURNER_COUNT; burner++)
    {
      try
      {
        burners[burner].join();     // Wait for thread to finish
      }
      catch (InterruptedException e)
      {
        // This thread was interrupted.
      }
    }
    
    System.out.println("Pressure reads " + boiler.getPressure() +
        ", limit is " + PRESSURE_LIMIT);
  }
  
  /////// Data //////
  private int m_pressure = 0;
}

class Burner extends Thread
{
  public static final int PRESSURE_INCREMENT = 15;
  
  public Burner(SteamBoiler boiler)
  {
    m_boiler = boiler;
  }
  
  public void run()
  {
    burn();
  }
  
  private void burn()
  {
    if (m_boiler.getPressure() <
        SteamBoiler.PRESSURE_LIMIT - PRESSURE_INCREMENT)
    {
      // Wait to simulate delay
      try
      {
        sleep(100);
      }
      catch (InterruptedException e)
      {
        // Ignore
      }
      
      m_boiler.setPressure( m_boiler.getPressure() +
          PRESSURE_INCREMENT );
    }
  }
  
  ///// Private data //////
  private SteamBoiler m_boiler;
}

package unsynchronized;

import java.awt.BorderLayout;
import java.awt.Color;
import java.awt.Dimension;
import java.awt.FlowLayout;
import java.awt.Graphics;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import javax.swing.BorderFactory;
import javax.swing.JApplet;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;
import unsynchronized.SteamBoiler;

/**
 *  This class provides a visual representation of the SteamBoiler.
 */
public class SteamBoilerFrame extends JFrame implements Runnable
{
  /**
   * Main entry point
   */
  public static void main(String[] args)
  {
    SteamBoilerFrame frame = new SteamBoilerFrame();
    frame.setDefaultLookAndFeelDecorated(true);
    frame.setDefaultCloseOperation(EXIT_ON_CLOSE);
    frame.setSize(300, 200);
    frame.setVisible(true);
  }
  
  public SteamBoilerFrame()
  {
    setTitle("Steam Boiler");
    m_gauge = new PressureGauge( new SteamBoiler() );
    setContentPane( createBoilerLayout() );
    // Start this thread
    m_timer = new Thread(this);
    m_timer.start();    // Start the timer
  }
  
  public void stop()
  {
    m_timer = null; // Stop the timer
  }
  
  private JPanel createBoilerLayout()
  {
    JPanel panel = new JPanel( new BorderLayout() );
    panel.setBackground(Color.lightGray);
    
    JLabel label = new JLabel("Steam Boiler (Max Pressure = " +
                              SteamBoiler.PRESSURE_LIMIT + ")"
                              );
    JPanel north = new JPanel( new FlowLayout(FlowLayout.CENTER) );
    north.add(label);
    panel.add(north, BorderLayout.NORTH);
    
    panel.add(m_gauge, BorderLayout.CENTER);
    
    JPanel south = new JPanel( new FlowLayout(FlowLayout.CENTER) );
    m_startButton.addActionListener( new ActionListener()
      {
        public void actionPerformed(ActionEvent event)
        {
          m_startButton.setEnabled(false);
          m_gauge.getBoiler().start();   // Start the boiler
        }
      }
    );
    south.add(m_startButton);
    
    m_lastPressure = m_gauge.getPressure();
    m_pressureText = new JLabel(PRESSURE_TEXT + m_lastPressure);
    south.add(m_pressureText);
    
    m_resetButton.addActionListener( new ActionListener()
      {
        public void actionPerformed(ActionEvent event)
        {
          m_gauge.setBoiler( new SteamBoiler() );
          m_startButton.setEnabled(true);
        }
      }
    );
    south.add(m_resetButton);
    panel.add(south, BorderLayout.SOUTH);
    
    return panel;
  }
  
    /*
     *   This constitutes a timer thread to refresh the gauge
     *   on a regular basis.
     */
  public void run()
  {
    while (m_timer != null)
    {
      try
      {
        Thread.sleep(100);
      }
      catch (InterruptedException e)
      {
        // Ignore
      }
      
      // Repaint if the pressure has changed since last time
      int pressure = m_gauge.getPressure();
      if (pressure != m_lastPressure)
      {
        m_lastPressure = pressure;
        m_pressureText.setText(PRESSURE_TEXT + pressure);
        m_gauge.repaint();
      }
    }
  }
  
  ///////// Private Data //////
  private boolean         m_laidOut = false;
  private Thread          m_timer;
  private PressureGauge   m_gauge;
  private JButton         m_startButton = new JButton("Burn!");
  private JButton         m_resetButton = new JButton("Reset");
  private JLabel          m_pressureText;
  private int             m_lastPressure = 0;
  
  private static final String PRESSURE_TEXT = "Pressure = ";
}

class PressureGauge extends JPanel
{
  public PressureGauge(SteamBoiler boiler)
  {
    setBoiler(boiler);
  }
  
  public void setBoiler(SteamBoiler boiler)
  {
    m_boiler = boiler;
    repaint();
  }
  
  public SteamBoiler getBoiler()
  {
    return m_boiler;
  }
  
  public int getPressure()
  {
    return m_boiler.getPressure();
  }
  
  public void paintComponent(Graphics g)
  {
    super.paintComponent(g);
    Dimension dim = getSize();
    int x = (dim.width - OUTER_WIDTH)/2;
    int y = 0;
    
    // Draw outer white area
    g.setColor(Color.WHITE);
    g.fillRect(x, y, OUTER_WIDTH, dim.height);
    
    // Draw outline
    g.setColor(Color.black);
    int width = OUTER_WIDTH - INSET - INSET;
    int height = dim.height - INSET - INSET;
    g.drawRect(x + INSET - 1, y + INSET - 1, width + 1, height + 1);
    
    // Draw the "mercury"
    Color color = Color.blue;
    if (getPressure() > SteamBoiler.PRESSURE_LIMIT)
      color = Color.red;
    g.setColor(color);
    
    float factor = (float) getPressure() / (float) (SteamBoiler.PRESSURE_LIMIT * 10);
    height = (int) ((float)height * factor);
    g.fillRect(x + INSET, dim.height - height - INSET, width, height);
  }
  
  ///// Private data ///////
  private static final int OUTER_WIDTH = 20;
  private static final int INSET = 5;
  
  private SteamBoiler m_boiler;
}

package threads;

public class SharedResource
{
    // ...

    private String m_value = "";

    public synchronized void setValue(String value)  // 1
    {
        m_value = value;
    }

    public synchronized String getValue()	     // 2
    {
        return m_value;
    }

    private static int m_count = 0;

    public static synchronized void incrementCount()  // 3
    {
        m_count++;
    }

    private Double m_salary = new Double(32000.0);

    public void giveRaise(double percent)
    {
        synchronized (m_salary) 		       // 4
        {
            double salary = m_salary.doubleValue();
            salary += salary * percent /100;
            m_salary = new Double(salary);
        }
    }
}

private void burn()
{
    synchronized (m_boiler)
    {
        if (m_boiler.getPressure() < 
                   SteamBoiler.PRESSURE_LIMIT - PRESSURE_INCREMENT)
        {
            // Wait to simulate delay
            try
            {
                sleep(100);
            }
            catch (InterruptedException e)
            {
                // Ignore
            }
                
            m_boiler.setPressure( m_boiler.getPressure() + 
                                  PRESSURE_INCREMENT );
        }
    }
}

package threads;

/**
 *  NOTE: This implementation of the SteamBoiler class is implemented
 *  with regard to synchronization.  
 */
public class SteamBoiler
{
  public static final int PRESSURE_LIMIT = 20;
  public static final int BURNER_COUNT = 10;
  
  public Burner[] start()
  {
    Burner[] burners = new Burner[BURNER_COUNT];
    for (int burner = 0; burner < BURNER_COUNT; burner++)
    {
      burners[burner] = new Burner(this);
      burners[burner].start();
    }
    
    return burners;
  }
  
  public int getPressure()
  {
    return m_pressure;
  }
  
  public void setPressure(int pressure)
  {
    m_pressure = pressure;
  }
  
  public static void main(String[] args)
  {
    SteamBoiler boiler = new SteamBoiler();
    
    Burner[] burners = boiler.start();
    
    for (int burner = 0; burner < BURNER_COUNT; burner++)
    {
      try
      {
        burners[burner].join();     // Wait for thread to finish
      }
      catch (InterruptedException e)
      {
        // This thread was interrupted.
      }
    }
    
    System.out.println("Pressure reads " + boiler.getPressure() +
      ", limit is " + PRESSURE_LIMIT);
  }
  
  /////// Data //////
  private int m_pressure = 0;
}

class Burner extends Thread
{
  public static final int PRESSURE_INCREMENT = 15;
  
  public Burner(SteamBoiler boiler)
  {
    m_boiler = boiler;
  }
  
  public void run()
  {
    burn();
  }
  
  private void burn()
  {
    synchronized (m_boiler)
    {
      if (m_boiler.getPressure() <
        SteamBoiler.PRESSURE_LIMIT - PRESSURE_INCREMENT)
      {
        // Wait to simulate delay
        try
        {
          sleep(100);
        }
        catch (InterruptedException e)
        {
          // Ignore
        }
        
        m_boiler.setPressure( m_boiler.getPressure() +
          PRESSURE_INCREMENT );
      }
    }
  }
  
  ///// Private data //////
  private SteamBoiler m_boiler;
}

package threads;

/**
*  This class will create a deadlock situation.
*/
public class Deadlock
{
    public static void main(String[] args)
    {
        Account a1 = new Account("Account 1", 10000);
        Account a2 = new Account("Account 2", 30000);
        Thread threadA = 
            new TransferFundsThread("Thread A", a1, a2, 5);
        Thread threadB = 
            new TransferFundsThread("Thread B", a2, a1, 20);
        threadA.start();
        threadB.start();
    }
}

class Account
{
    public Account(String name, int balance)
    {
        m_name = name;
        m_balance = balance;
    }
    
    public int getBalance()
    {
        return m_balance;
    }
    
    public void setBalance(int newBalance)
    {
        m_balance = newBalance;
    }
    
    public String getName()
    {
        return m_name;
    }
    
    private String m_name;
    private int    m_balance;
}

class TransferFundsThread extends Thread
{
    public TransferFundsThread(String name, 
                               Account from, 
                               Account to, 
                               int amount)
    {
        super(name);
        m_from   = from;
        m_to     = to;
        m_amount = amount;
    }
    
    public void run()
    {
        transferFunds();
    }
    
    /**
     *  In order to transfer funds safely, we need to be able 
     *  to hold a lock on both the m_from and the m_to accounts, 
     *  and not release the locks until the transfer is completed.
     *  NOTE: This version does not consider the possibility that 
     *  there may be insufficient funds available.  This has been 
     *  omitted to simplify the example, but would have to be done 
     *  in the real world.
     */
    private void transferFunds()
    {
        synchronized (m_from)
        {
            System.out.println(getName() + 
                               ": Getting balance from " + 
                               m_from.getName());
            int balance1 = m_from.getBalance();
            System.out.println(getName() + 
                               ": Balance from " + 
                               m_from.getName() + 
                               " = " + balance1);
                    
            try
            {
                sleep(100);         // insert delay
            }
            catch (InterruptedException e)
            {
                // ignore
            }
            
            synchronized (m_to)
            {
                System.out.println(getName() + 
                                   ": Getting balance from " + 
                                   m_to.getName());
                int balance2 = m_to.getBalance();
                System.out.println(getName() + 
                                   ": Balance from " + 
                                   m_to.getName() + 
                                   " = " + balance2);
                                   
                if (balance1 >= m_amount)
                {
                    System.out.println(getName() + 
                                       ": Withdrawing funds from " +
                                       m_from.getName());
                    m_from.setBalance(balance1 - m_amount);
                    System.out.println(getName() + 
                                       ": Depositing funds into " +
                                       m_to.getName());
                    m_to.setBalance(balance2 + m_amount);
                    
                    System.out.println(getName() + 
                                       ": New balances are:\n" +
                                       "  " + m_from.getName() + 
                                       " = " + 
                                       m_from.getBalance() + "\n" +
                                       "  " + m_to.getName() + " = " +
                                       m_to.getBalance() );
                }
            }
        }
    }
    
    /// Private data ///
    private Account m_from;
    private Account m_to;
    private int     m_amount;
}

package threads;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


/**
 *  This is a Producer thread that supplies strings
 *  to a Consumer thread.  It reads a line from System.in
 *  and passes it to the Consumer through a SharedBuffer,
 *  until the end of stream occurs.
 */
public class Producer extends Thread
{
  public static void main(String[] args)
  {
    // Create a shared buffer
    SharedBuffer buffer = new SharedBuffer();
    // Create and start a consumer thread
    Consumer consumer = new Consumer(buffer);
    consumer.start();
    // Create and start a Producer thread
    Producer producer = new Producer(buffer);
    producer.start();
  }
  
  Producer(SharedBuffer buffer)
  {
    super("Producer");
    m_buffer = buffer;
  }
  
  public void run()
  {
    // Connect up to System.in
    BufferedReader reader = new BufferedReader(
      new InputStreamReader(System.in)
      );
    
    try
    {
      // Loop, asking for input from the user until EOF
      while (true)
      {
        String line = reader.readLine();
        if (line == null)
          break;      // End of input
        m_buffer.putData(line);
      }
    }
    catch (IOException e)
    {
      e.printStackTrace();
    }
    
    System.out.println("Done with input: Signalling to Consumer");
    m_buffer.putData(null);
  }
  
  /// Private data ///
  private SharedBuffer m_buffer;
}

/**
 *  This is a Consumer thread that waits for input from
 *  the Producer, supplied through a SharedBuffer.  It
 *  accepts the string passed, reverses it, and prints out
 *  the result.  When it encounters a null string, the thread
 *  exits.
 */
class Consumer extends Thread
{
  Consumer(SharedBuffer buffer)
  {
    super("Consumer");
    m_buffer = buffer;
  }
  
  public void run()
  {
    while (true)
    {
      String line = m_buffer.getData();
      if (line == null)
        break;
      // Reverse the line and print it out
      line = reverse(line);
      System.out.println(line);
    }
    
    System.out.println("Consumer thread exiting...");
  }
  
  private String reverse(String string)
  {
    StringBuffer buffer = new StringBuffer(string);
    return buffer.reverse().toString();
  }
  
  //// Private data ////
  private SharedBuffer m_buffer;
}

/**
 *  This is a SharedBuffer class to which the Producer puts
 *  data, and from which the Consumer retrieves it.
 *  It contains the necessary thread coordination code to
 *  ensure that the Producer and Consumer work together properly.
 */
class SharedBuffer
{
  static final int RESOURCE_LIMIT = 10;
  
  boolean isFull()
  {
    return m_data.size() > RESOURCE_LIMIT;
  }
  
  boolean isEmpty()
  {
    return m_data.size() <= 0;
  }
  
  synchronized void putData(String s)
  {
    while ( isFull() )  // Wait until there's room
    {
      try
      {
        wait();
      }
      catch (InterruptedException e)
      {
        // Ignore
      }
    }
    
    m_data.add(s);   // To the end of the FIFO list
    
    notify();       // Notify any waiter
  }
  
  synchronized String getData()
  {
    while ( isEmpty() ) // Wait until there's data available
    {
      try
      {
        wait();
      }
      catch (InterruptedException e)
      {
        // Ignore
      }
    }
    
    String data = (String)(m_data.get(0)); // Get first element
    if (data != null)
      m_data.remove(data);     // and remove it from the List
    
    notify();       // Notify any waiter
    
    return data;
  }
  
  //// Private data ////
  
  // A list of strings.
  private List<String> m_data = new ArrayList<String>();
  private boolean m_done = false;
}

package threads;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
import java.io.PrintWriter;

/**
 *  This is a PipedProducer thread that supplies strings
 *  to a PipedConsumer thread.  It reads a line from System.in
 *  and passes it to the PipedConsumer through a SharedBuffer,
 *  until the end of stream occurs.
 */
public class PipedProducer extends Thread
{
  public static void main(String[] args)
  {
    try
    {
      // Create the input and output pipes and hook them up
      PipedWriter writer = new PipedWriter();
      PipedReader reader = new PipedReader(writer);
      // Create and start a consumer thread
      PipedConsumer consumer = new PipedConsumer(reader);
      consumer.start();
      // Create and start a PipedProducer thread
      PipedProducer producer = new PipedProducer(writer);
      producer.start();
    }
    catch (IOException e)
    {
      e.printStackTrace();
    }
  }
  
  PipedProducer(PipedWriter writer)
  {
    super("Producer");
    m_writer = new PrintWriter(writer);
  }
  
  public void run()
  {
    // Connect up to System.in
    BufferedReader reader = new BufferedReader(
      new InputStreamReader(System.in)
      );
    
    try
    {
      // Loop, asking for input from the user until EOF
      while (true)
      {
        String line = reader.readLine();
        if (line == null)
          break;      // End of input
        m_writer.println(line);
      }
    }
    catch (IOException e)
    {
      e.printStackTrace();
    }
    
    System.out.println("Done with input: Closing pipe");
    m_writer.close();
  }
  
  /// Private data ///
  private PrintWriter m_writer;
}

/**
 *  This is a PipedConsumer thread that waits for input from
 *  the PipedProducer, supplied through a SharedBuffer.  It
 *  accepts the string passed, reverses it, and prints out
 *  the result.  When it encounters a null string, the thread
 *  exits.
 */
class PipedConsumer extends Thread
{
  PipedConsumer(PipedReader reader)
  {
    super("Consumer");
    m_reader = new BufferedReader(reader);
  }
  
  public void run()
  {
    try
    {
      while (true)
      {
        String line = m_reader.readLine();
        if (line == null)
          break;
        // Reverse the line and print it out
        line = reverse(line);
        System.out.println(line);
      }
    }
    catch (IOException e)
    {
      e.printStackTrace();
    }
    
    System.out.println("Consumer thread exiting...");
  }
  
  private String reverse(String string)
  {
    StringBuffer buffer = new StringBuffer(string);
    return buffer.reverse().toString();
  }
  
  //// Private data ////
  private BufferedReader m_reader;
}