Ogg-Vorbis format, 52Mbytes

WAV format, 290Mbytes

Distributed systems and concurrent systems

• Distributed and concurrent systems both deal with "multiple things happening together"
• Distributed systems deal with multiple execution threads on more than one computer
• Concurrent systems deal with multiple execution threads on a single computer
• They share some common aspects
  • need to communicate
  • sharing of state
• They differ in the amount of shared information and in communication bandwidth
• These produce different programming concerns
• A common way of dealing with some of the problems caused by distributed systems on a local basis is to use concurrency techniques

Concurrency

• A concurrent program can be described as one that does more than one thing at a time
• Concurrency first appeared in hardware with e.g. dedicated I/O processors which offload processing from the main CPU
• Concurrency was mainly used in operating systems, with prime examples being Unix processes

Unix processes

• The O/S schedules and runs a number of processes
• Some sort of time-sharing or process switching mechanism is used to allocate time to each process
• One process can create another, with a parent-child relationship
• There is limited sharing of information between parent-child processes:
  • Open files
  • Shared memory
  • Named pipes and streams
  • Sockets
• Processes each have their own stack and private data areas

Scheduling policies

• A process scheduler may be pre-emptive or co-operative
• Pre-emptive: the scheduler is in charge of how long a process runs for. If a process exceeds its timeslice, it is stopped by the scheduler
• Co-operative: each process is in charge of how long it runs for. When a process feels like co-operating, it will surrender execution
• Pre-emptive is good, co-operative is usually bad
• Most O/S now support pre-emptive scheduling, but not all (Windows 3.1 is a major co-operative scheduler). NT has both models, to support "traditional" 16-bit Windows applications as well as newer 32-bit ones

Threads

• Threads are "lightweight processes". They share more information than heavyweight processes
• Typically threads share
  • the same system stack
  • the same heap memory
• Threads usually have their own procedure call/local variable stacks

Shared memory problems

• If two processes/threads share memory, then both can read and write to it
• An assignment such as i = i + 1 or i++ involves both a read and a write
• The scheduler does not treat such an operation as "atomic" and may interrupt it half way through to perform a context switch to another process.
• So if process/thread A reads a value, control switches to B which writes a value, then back to A which also writes a value, then B's change is lost
• Many techniques exist to avoid these problems, such as locks, semaphores, monitors, etc

Threads

• Java includes threads as part of the language model
• No extra libraries are required
• An application consists of at least one "user" thread which runs the main() method
• The main thread may create and run other threads as well
• A JVM may run several "kernel" and "user" threads as well
• An application terminates when all user threads terminate

Thread scheduling

• Many aspects of threads are looked after by the JVM
• Thread scheduling may use scheduling models supplied by the O/S
• Java threads may behave differently depending on whether the O/S supports pre-emptive or co-operative threads

Thread creation

• Subclass from Thread:

        class MyThread extends Thread {
            public void run() {
                // code for my thread
            }
        }

  Start this thread by

        MyThread thr = new MyThread();
        thr.start();

• Implement Runnable:

        class MyRun implements Runnable {
            public void run() {
                // code for my thread
            }
        }

  Start this thread by

        MyRun run = new MyRunnable();
        new Thread(run).start();

Example

import java.io.*;

public class SimpleThread extends Thread {

    protected int base;

    static public void main(String[] argv) {
	for (int n = 0; n < 3; n++) {
	    new SimpleThread(n).start();
	}
    }

    public SimpleThread(int base) {
	this.base = base;
    }

    public void run() {
	int count = 0;
	while (count < 5) {
	    System.out.println("Thread " + base + " loop " + count);
	    count++;
	    try {
		sleep((base + 1) * 1000);
	    } catch(InterruptedException e) {
		e.printStackTrace();
	    }
	}
	System.out.println("Thread " + base + " finished");
    }
}

Sequential server

while (msg = getMessage())
    handle msg

Concurrent server

while (msg = getMessage())
    create new thread to handle msg

Thread life-cycle

Thread alive states

Example broken shared memory

• The following program (based on one from Core Java) shares a number of "bank accounts".
• Amounts are transferred from one to another on a random basis: the sum should remain constant
• Every now and then the sum is printed
• Each thread "yields" halfway through a transfer. A pre-emptive O/S could interrupt the method when it felt like, including at this point. Yielding at this point allows most opportunity for errors
• On a test run under Linux it was producing wrong results after 30 transactions. Without the yield() it was still running "correctly" after 100,000 transactions (but would probably break sometime)

/**
 * @version 1.20 1999-04-26
 * @author Cay Horstmann
 */

public class UnsynchBankTest
{  public static void main(String[] args)
   {  Bank b = new Bank(NACCOUNTS, INITIAL_BALANCE);
      int i;
      for (i = 0; i < NACCOUNTS; i++)
      {  TransferThread t = new TransferThread(b, i,
            INITIAL_BALANCE);
         t.setPriority(Thread.NORM_PRIORITY + i % 2);
         t.start();
      }
   }

   public static final int NACCOUNTS = 10;
   public static final int INITIAL_BALANCE = 10000;
}

class Bank
{  public Bank(int n, int initialBalance)
   {  accounts = new int[n];
      int i;
      for (i = 0; i < accounts.length; i++)
         accounts[i] = initialBalance;
      ntransacts = 0;
   }

   public void transfer(int from, int to, int amount)
   {  if (accounts[from] < amount) return;
      accounts[from] -= amount;

      // let the O/S run another thread 
      Thread.currentThread().yield();

      accounts[to] += amount;
      ntransacts++;
      if (ntransacts % NTEST == 0) test();
   }

   public void test()
   {  int sum = 0;

      for (int i = 0; i < accounts.length; i++)
         sum += accounts[i];

      System.out.println("Transactions:" + ntransacts
         + " Sum: " + sum);
   }

   public int size()
   {  return accounts.length;
   }

   public static final int NTEST = 10;
   private int[] accounts;
   private long ntransacts = 0;
}

class TransferThread extends Thread
{  public TransferThread(Bank b, int from, int max)
   {  bank = b;
      fromAccount = from;
      maxAmount = max;
   }

   public void run()
   {  try
      {  while (!interrupted())
         {  int toAccount = (int)(bank.size() * Math.random());
            int amount = (int)(maxAmount * Math.random());
            bank.transfer(fromAccount, toAccount, amount);
            sleep(1);
         }
      }
      catch(InterruptedException e) {}
   }

   private Bank bank;
   private int fromAccount;
   private int maxAmount;
}

Mutual exclusion

• Mutual exclusion can be accomplished in Java using the synchronized keyword
• This can be applied to methods

        synchronized void writeLogEntry() {...}

  or to blocks of code

        ...
        synchronized(anObject) {
            ...
        }
        ...

• A synchronized method/block can be viewed as a "critical section" that excludes access the object by any other synchronized methods during its execution

Use of synchronized

• The use of this word is pretty straightforward: whenever a resource needs to be accessed in an atomic manner, then make the access synchronized

Fixing bank account

• The unsynch bank account allowed "simultaneous" access to the elements of the bank account array
• Removing the yield() does not fix the problem
• Synchronizing the transfer() and test() methods does

  
        public synchronized void test() {...}
        public synchronized void transfer(int from, int to, int amount) {...}

• Strictly, synchronising test() is redundant since it is only called from transfer(), but it does no harm

  
  /**
   * @version 1.20 1999-04-26
   * @author Cay Horstmann
   * @author Jan Newmarch
   */
  
  public class SynchBankTest
  {  public static void main(String[] args)
     {  Bank b = new Bank(NACCOUNTS, INITIAL_BALANCE);
        int i;
        for (i = 0; i < NACCOUNTS; i++)
        {  TransferThread t = new TransferThread(b, i,
              INITIAL_BALANCE);
           t.start();
        }
     }
  
     public static final int NACCOUNTS = 10;
     public static final int INITIAL_BALANCE = 10000;
  }
  
  class Bank
  {  public Bank(int n, int initialBalance)
     {  accounts = new int[n];
        int i;
        for (i = 0; i < accounts.length; i++)
           accounts[i] = initialBalance;
        ntransacts = 0;
     }
  
     public synchronized void transfer(int from, int to, int amount)
     {  if (accounts[from] < amount) return;
        accounts[from] -= amount;
        accounts[to] += amount;
        ntransacts++;
        if (ntransacts % NTEST == 0) test();
     }
  
      /*
       * This method doesn't need to be synchronized since it 
       * is only called from a method that is already synchronized.
       * But it doesn't harm to synchronize it, and may be safer
       * in case you ever call it fromn somewhere else
       */
     public synchronized void test()
     {  int sum = 0;
  
        for (int i = 0; i < accounts.length; i++)
           sum += accounts[i];
  
        System.out.println("Transactions:" + ntransacts
           + " Sum: " + sum);
     }
  
     public int size()
     {  return accounts.length;
     }
  
     public static final int NTEST = 10;
     private int[] accounts;
     private long ntransacts = 0;
  }
  
  class TransferThread extends Thread
  {  public TransferThread(Bank b, int from, int max)
     {  bank = b;
        fromAccount = from;
        maxAmount = max;
     }
  
     public void run()
     {  try
        {  while (!interrupted())
           {  int toAccount = (int)(bank.size() * Math.random());
              int amount = (int)(maxAmount * Math.random());
              bank.transfer(fromAccount, toAccount, amount);
              sleep(1);
           }
        }
        catch(InterruptedException e) {}
     }
  
     private Bank bank;
     private int fromAccount;
     private int maxAmount;
  }
  
  

• This corrects the problem, but does not allow the maximum concurrency: in order to transfer between two accounts, all transfers but one are blocked, including those that do not involve these accounts (see "Deadlock" later)

More synchronization

• Synchronization occurs on an object i.e. multiple threads executing code associated with a single object
• Synchronizing a method is equivalent to synchronizing all of the method code on the object this
• In the SynchBankTest, the synchronizing object is the Bank object b created by main()
• The method transfer() could have been written

  public void transfer(int from, int to, int amount)
   {    synchronized(this) {
            if (accounts[from] < amount) return;
            accounts[from] -= amount;
            accounts[to] += amount;
            ntransacts++;
            if (ntransacts % NTEST == 0) test();
        }
   }
  

Synchronization on a shared object

• The above example uses a single Bank object to synchronize on
• If threads use different objects, they can only synchronize on a shared object
• The following (artificial) example creates two separate Adder objects and then uses a third shared object synchObject to avoid contention

/** 
 * This is a hard way of doing arithmetic:
 * fire off a separate thread for each sum
 * and then synchronize them on a shared object
 * to avoid  contention
 */ 

public class Adder extends Thread {

    // static fields are shared by all instances
    // of the Adder class i.e. one field for
    // all instances
    private static Object synchObject;
    private static int sum = 0;

    // non-static fields belong to individual
    // instances of each Adder object i.e.
    // separate fields for each instance
    private int increment;

    public static void main(String[] args) {
	Adder a1 = new Adder(1);
	Adder a3 = new Adder(3);

	// this could have been set to a1 or a3
	synchObject = new Object();

	// the following should do
	//   sum += 1
	//   sum += 3
	// in an indeterminate order
	a1.start();
	a3.start();
	try {
	    // sleep for long enough for all threads
	    // to complete (hopefully)
	    sleep(1000);
	} catch(InterruptedException e) {
	    // ignore
	}
	// now all threads should have finished
	System.out.println("Sum is " + sum);
    }

    public Adder(int increment) {
	this.increment = increment;
    }

    public void run() {
	synchronized(synchObject) {
	    sum += increment;
	}
    }
}

Fine control of threads

There are a number of ways in which threads can control execution

• Thread.sleep(milliseconds)
• Thread.yield() allows the scheduler to select another runnable thread
• wait() block until woken up by a call to notify
• notify() and notifyAll() wakeup threads waiting to regain control of the monitor's lock

Wait/notify

• wait()/notifyAll() occur in pairs, associated with a synchronizing object
• One thread in a synchronized block waits.
• This gives up control of the synchronizing object to another thread
• When a thread wants to wakeup any waiting thread, it notifies the thread
• The method notifyAll() is preferred to notify()
• Note that wait() and notifyAll() can usually only be called within a synchronized block or method

  synchronized(obj) {
        ...
        obj.wait();
        ...
   }

Reader/Writer example

import java.io.*;

/**
 * This class copies strings from one thread to another.
 * The class ReaderWriter has a field sharedStr that is
 * set by Writer and printed by Reader. The Writer sets a 
 * value, wakes up the Reader. The Writer then waits till 
 * the Reader is finished and then writes another value.
 * The Reader waits till it is woken and then prints the
 * value. It then wakes up the Writer and waits again.
 * The loops terminate when the Writer writes an empty string ""
 */
public class ReaderWriter {
    public static String sharedStr;
    public static Object synchObject;
 
    public static void main(String[] args) {
	synchObject = new Object();
	new Reader().start();
	new Writer().start();
    }
}

class Writer extends Thread {
    private String[] lines = {"line 1",
			      "line 2",
			      "line 3",
			      "line 4",
			      ""        // terminating string
    };

    public void run() {
	for (int n = 0; n  < lines.length; n++) {
	    // synch on the shared object
	    synchronized(ReaderWriter.synchObject) {
		System.out.println("Writer assigning to string");
		ReaderWriter.sharedStr = lines[n];

		// another thread synch'ed on 
		// the synchObject can wake up
		System.out.println("Waking up reader");
		ReaderWriter.synchObject.notifyAll();

		// and we wait until we are
		// woken before continuing
		try {
		    System.out.println("Writer waiting");
		    ReaderWriter.synchObject.wait();
		} catch(InterruptedException e) {
		    // ignore
		}
	    }
	}
    }
}

class Reader extends Thread {

    public void run() {
	while (true) {
	    synchronized(ReaderWriter.synchObject) {
		// wait until woken before trying
		// to print anything
		try {
		    System.out.println("Reader waiting");
		    ReaderWriter.synchObject.wait();
		} catch(InterruptedException e) {
		    // ignore
		}

		if (ReaderWriter.sharedStr.equals("")) {
		    // end of loop. 
		    // wakeup anyone else before breaking
		    // out of loop
		    ReaderWriter.synchObject.notifyAll();
		    break;
		}

		System.out.println(ReaderWriter.sharedStr);
		
		// and then tell another thread to wakeup
		ReaderWriter.synchObject.notifyAll();
	    }
	}
    }
}

Deadlock

Example:

Two processes wish to communicate with each other, in a 2-way manner. Two pipes can be used for this:

Example:

When the "give way to the right" rule was in force, this was a common situation:

Deadlock detection

A resource allocation graph is a graph showing processes, resources and the connections between them. Processes are modelled by circles, resources by squares. If a process controls a resource then there is an arrow from resource to process. If a process requests a resource an arrow is shown the other way.

For example, P1 is using R1 and has requested R2, while P2 is using R2.

Deadlock can now occur if P2 requests R1 - setting up a cycle.

The total set of conditions for deadlock to occur are:

1. Mutual exclusion. Each resource is either currently assigned to one process or is available.
2. Hold and wait. Processes holding resources can request new ones.
3. No preemption. Resources granted cannot be taken away, but must be released by the process holding them.
4. Circular wait. There must be a cycle in the resource allocation graph.

Deadlock and the bank account

• Making the transfer() method synchronized locks all accounts for a transfer between two of them
• This reduces concurrency
• Why not just lock the two accounts?
• This introduces a potential deadlock

  
        public void transfer(int from, int to, int amount) {
            if (accounts[from] < amount) return;
            synchronized(accounts[from]) {
                synchronized(accounts[to]) {
                    accounts[from] -= amount;
                    accounts[to] += amount;
                    ntransacts++;
                    if (ntransacts % NTEST == 0) test();
                }
            }
        }

• A simultaneous transfer from e.g. one to two, two to three and three to one will introduce a cycle

Deadlock prevention

To prevent deadlocks from occurring, one of the four conditions must be disallowed.

1. Mutual exclusion. Make some resources unsharable, such as printers, tape drives.
2. Hold and wait. Process must request all needed resources at one time. This results in wasted resources.

   Process must release all current resources before requesting more. This is not feasible.

3. No Preemption. Make it possible for the O/S to make a process give up a resource. This may result in lost data.
4. Circular wait. Make it impossible for cycles to occur in the graph by restricting the types of graphs that can be made. For example, give each resource a priority number. If a resource is held with a priority, then a new resource can only be requested of lower priority.

Bank account avoiding circular wait

• Prioritise the accounts by using their numeric order
• Always grab the higher order before the lower one

  
        public void transfer(int from, int to, int amount) {
            if (accounts[from] < amount) return;
            int high, low;
            if (from > to) {
                high = from;
                low = to;
            } else {
                high = to;
                low = from;
            }
            synchronized(accounts[high]) {
                synchronized(accounts[low]) {
                    accounts[from] -= amount;
                    accounts[to] += amount;
                    ntransacts++;
                    if (ntransacts % NTEST == 0) test();
                }
            }
        }

Deadlock avoidance

Deadlock prevention is to ensure that deadlocks never occur. Deadlock avoidance is attempting to ensure that resources are never allocated in a way that might cause deadlock.

There are situations that may give rise to cycles, but in practise will not lead to deadlock. For example, P1 and P2 both want R1 and R2, but in this manner:

• P1 needs R1 first, then R2, but with a slight overlap.
• P2 needs R2 first, then R1 but with no overlap.

P1 can be given R1 and P2 can be given R2. P1 will require R2 before it can give up R1, but that is ok - P2 will give up R2 before it needs R1.

The Banker's algorithm can be used to ensure this, but it is too complicated for general use.

Listing a directory (and subdirectories)

Listing all the files in a directory is a recursive operation:

• write the name of the current directory
• find all the files in the current directory
• for each file, if it is an ordinary file then write its name, else call the list files function with the new directory

Recursive single-thread solution

import java.io.*;

public class ListFiles {

    public static void main(String[] args) {
	ListFiles lister = new ListFiles(new File(args[0]));
    }

    public ListFiles(File dir) {
	listDirectory(dir);
    }

    public void listDirectory(File dir) {
	System.out.println(dir.getAbsolutePath());

	File[] files = dir.listFiles();
	if (files == null) {
	    return;
	}

	for (int n = 0; n < files.length; n++) {
	    File file = files[n];
	    if (! file.isDirectory()) {
		System.out.println(file.getAbsolutePath());
	    } else {
		listDirectory(file);
	    }
	}
    }
}

Multi-threaded solution

Each directory could be handled by a separate thread. When a list of files is found for each directory, it's name is printed if it is an ordinary file, but if it is a directory then a new thread is created to handle it.

import java.io.*;

public class ListFilesThreads extends Thread {

    private static Object synch;
    private static ThreadGroup group;

    private File dir;

    public static void main(String[] args) {
	ListFilesThreads lister = new ListFilesThreads(new File(args[0]));

	lister.start();
    }

    public ListFilesThreads(File dir) {
	this.dir = dir;
    }

    public void run() {
	listDirectory(dir);
    }

    public void listDirectory(File dir) {
	System.out.println(dir.getAbsolutePath());
	File[] files = dir.listFiles();
	if (files == null) {
	    return;
	}
	for (int n = 0; n < files.length; n++) {
	    File file = files[n];
	    if (! file.isDirectory()) {
		System.out.println(file.getAbsolutePath());
	    } else {
		new ListFilesThreads(file).start();
	    }
	}
    }
}

Summing the size of files in directories

Listing files can be made more complex by finding their size and summing them.

Recursive sum

import java.io.*;

public class SumFiles {

    private long sum = 0;

    public static void main(String[] args) {
	SumFiles lister = new SumFiles(new File(args[0]));
    }

    public SumFiles(File dir) {
	sumDirectory(dir);
	System.out.println("Sum is " + sum);
    }

    public void sumDirectory(File dir) {
	File[] files = dir.listFiles();

	if (files == null) {
	    return;
	}

	for (int n = 0; n < files.length; n++) {
	    File file = files[n];

	    if (! file.isDirectory()) {
		//System.out.println(file.length() + " " + file.toString());
		System.out.println(file.length() + " " + sum);
		sum += file.length();
	    } else {
		sumDirectory(file);
	    }
	}
    }
}

Multi-threaded sum

This one is a lot more complicated!

• The sum variable must be shared between all threads, so it needs to be static
• Access to sum must be synchronised
• The synchronisation must be on a common shared object, say the first SumFilesThread object created
• How do you know when the summation is complete?
  Wait until all threads have completed
• How do you know when all threads have completed?
  • Use Thread.activeCount(). But this is an estimate only, so is not reliable
  • Keep a count of all threads that are alive
• How do you keep count of all threads?
  • Have a shared static variable threadCount
  • Make all changes to this synchronized
  • When a new thread is created, increment the count (you can't wait until run() is called, because there may be a gap between creation and running)
  • When run() finishes, decrement the count

import java.io.*;

public class SumFilesThreads extends Thread {

    private static long sum = 0;
    private static long threadCount = 1;
    private static Object synch;

    private File dir;

    public static void main(String[] args) {
	SumFilesThreads lister = new SumFilesThreads(new File(args[0]));

	// synchronize on the first object created
	synch = lister;
	lister.start();
	synchronized(synch) {
	    while (true) {
		try {
		    // every time a thread dies, it should
		    // notify all others, so we will be woken
		    // up periodically
		    synch.wait();
		} catch(InterruptedException e) {
		    System.out.println("Interrupted " + e);
		}
		if (threadCount == 0) {
		    System.out.println("Sum " + sum);
		    System.exit(0);
		}
	    }
	}
    }

    public SumFilesThreads(File dir) {
	this.dir = dir;
    }

    public void run() {
	listDirectory(dir);
    }

    public void listDirectory(File dir) {
	if (! dir.isDirectory()) return;
	
	// System.out.println(dir.getAbsolutePath());
	File[] files = dir.listFiles();
	for (int n = 0; n < files.length; n++) {
	    File file = files[n];
	    if (! file.isDirectory()) {
		// make access to sum atomic
		synchronized(synch) {
		    sum += file.length();
		}
	    } else {
		synchronized(synch) {
		    // make access to threadCount atomic
		    threadCount++;
		}
		new SumFilesThreads(file).start();
	    }
	}

	synchronized(synch) {
	    // this thread is finished, wake up a 
	    // thread waiting on the synch object
	    threadCount--;
	    synch.notifyAll();
	}
    }
}

Conclusion

• Java makes it relatively easy to do thread level programming
• It supplies a set of simple methods in the Thread class
• Thread programming is not always simple: It is easy to make obscure and hard to trace errors using threads