Concurrency and Threads¶

Threads vs Processes¶

Traditional Operating Systems have Processes
- Each process has its own address space
- Single thread of control
- When a process makes a system call, the process blocks until the call completes.
In many situations it is desirable to have multiple threads of control for:
- Explicit parallelism - to take advantage of multiple processors
- To allow computation to continue while waiting for system calls to return
  
  example: one thread reads a file into a buffer, another thread compresses the buffer, and a final thread writes out a compressed file.
- To reduce the cost of context switching
- Implicit parallelism - to keep a single processor busy

What are Processes, Threads?¶

Processes have:
- Program counter
- Stack
- Register set
- An exclusive virtual address space
- Sandboxing from other process except to the extent that the process participates in IPC
Threads:
- Threads can be thought of (and are often referred to as) lightweight processes
- Provide multiple threads of control
- Have multiple program counters
- Have multiple stacks
- One parent process, the virtual address space is shared across processes
- Each thread runs sequentially
- In a given process with N threads, 0-i threads may be blocked, and 0-k threads are runnable or running.

Common Threading Use Cases¶

Client / Server - ex: file servers:
- One thread listens on a network socket for clients
- When a client connects, a new thread is spawned to handle the request. This permits several clients to connect to the file server at one time because each request is handled by a separate thread of execution.
- To send the file data, two threads can be used the first can read from the file on disk and the second can write the read buffers to the socket.
Example Client-Server - a TCP echo server

	public class TcpServer
	{
		private Socket _socket;
		private Thread _serverThread;

		public TcpServer() {
			_socket = new Socket (AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
			_serverThread = new Thread (Server);
			_serverThread.Start ();
		}

		private void Server() {
			_socket.Bind (new IPEndPoint (IPAddress.Any, 8080));
			_socket.Listen (100);
			while (true) {
				var serverClientSocket = _socket.Accept ();
				new Thread (Server) { IsBackground = true }.Start (serverClientSocket);
			}
		}

		private void ServerThread(Object arg) {
			try {
				var socket = (Socket)arg;
				socket.Send (Encoding.UTF8.GetBytes ("Echo"));
			} catch(SocketException se) {
				Console.WriteLine (se.Message);
			}
		}
	}

	public class TcpClient {
		public void ConnectToServer() {
			var socket = new Socket (AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
			socket.Connect (new IPEndPoint (IPAddress.Parse("127.0.0.1"), 8080));
			var buffer = new byte[1024];
			var receivedBytes = socket.Receive (buffer);
			if (receivedBytes > 0) {
				Array.Resize (ref buffer, receivedBytes);
				Console.WriteLine (System.Text.Encoding.UTF8.GetString (buffer));
			}
		}
	}

Parallel computation:
- An algorithm is designed to solve some small part or subproblem of a larger problem
- To the extent that the subproblems are not inter-dependent, they can be executed in parallel
- Multiple threads can work against a common task queue.

Warning

The focus of this course is on distributed (not parallel) systems. Nevertheless, you may find that you want to take advantage of parallel computing in your work. We encourage you to read Christopher and Thiruvathukal, http://hpjpc.googlecode.com, which contains many examples of parallel algorithms in Java. You may also find Läufer, Lewis, and Thiruvathukal’s Scala workshop tutorial helpful. See http://scalaworkshop.cs.luc.edu.

Example Parallel Computation - factoring an integer

	public class ParallelComputationWorkItem {
		public readonly long LowerBound;
		public readonly long UpperBound;
		public readonly long Number;
		public readonly ICollection<long> Divisors;
		public ParallelComputationWorkItem(long lower, long upper, long number, ICollection<long> divisors) {
			LowerBound = lower;
			UpperBound = upper;
			Number = number;
			Divisors = divisors;
		}
	}

	public class ParallelComputation {

		public void PrintFactors(long number) {
			var threads = new List<Thread>();
			var divisors = new List<long> ();
			var cpuCount = Environment.ProcessorCount;
			long lower = 1;
			for (var i = 0; i < cpuCount; i++) {
				var thread = new Thread (Worker);
				var upper = lower + (number / cpuCount);
				thread.Start(new ParallelComputationWorkItem(lower, upper, number, divisors));
				threads.Add (thread);
				lower = upper;
			}
			foreach (var thread in threads) {
				thread.Join ();
			}
			Console.WriteLine ("Divisors - ");
			foreach (var divisor in divisors) {
				Console.WriteLine (divisor);
			}
		}

		private void Worker(object args) {
			var workItem = (ParallelComputationWorkItem)args;
			for (var i = workItem.LowerBound; i < workItem.UpperBound; i++) {
				if (workItem.Number % i == 0) {
					lock (workItem.Divisors) {
						workItem.Divisors.Add (i);
					}
				}
			}
		}
	}

Pipeline processing:
- An algorithm must be executed in several stages that depend upon each other.
- For example if there are three stages, then three threads can be launched for each of the stages. As the first thread completes some part of the total work, it can pass it to a queue for the second stage to be processed by the second thread. At this time, the first thread and second thread can work on their own stages in parallel. The same continues to the third thread for the third stage of computation.
Example Pipeline Processing - file compression

	public class PipelineComputation {
		private readonly Queue<byte[]> _readData;
		private readonly Queue<byte[]> _compressionData;
		private volatile bool _reading = true;
		private volatile bool _compressing = true;

		public PipelineComputation () {
			_readData = new Queue<byte[]>();
			_compressionData = new Queue<byte[]>();
		}

		public void PerformCompression() {
			var readerThread = new Thread (FileReader);
			var compressThread = new Thread (Compression);
			var writerThread = new Thread (FileWriter);
			readerThread.Start ();
			compressThread.Start ();
			writerThread.Start ();
			readerThread.Join ();
			compressThread.Join ();
			writerThread.Join ();
		}

		private void FileReader() {
			using (var stream = new FileStream("file.txt", FileMode.Open, FileAccess.Read)) {
				int len;
				var buffer = new byte[1024];
				while ((len = stream.Read(buffer, 0, buffer.Length)) > 0) {
					if (len != buffer.Length) {
						Array.Resize (ref buffer, len);
					}
					lock (_readData) {
						while (_readData.Count > 10) {
							Monitor.Wait (_readData);
						}
						_readData.Enqueue(buffer);
						Monitor.Pulse (_readData);
					}
				}
			}
			_reading = false;
		}

		private void Compression() {
			var workLeft = false;
			while (_reading || workLeft) {
				workLeft = false;
				byte[] dataToCompress = null;
				lock (_readData) {
					while (_reading && _readData.Count == 0) {
						Monitor.Wait (_readData, 100);
					}
					workLeft = _readData.Count > 1;
					if (_readData.Count > 0) {
						dataToCompress = _readData.Dequeue ();
					}
				}
				if (dataToCompress != null) {
					var compressed = Compress(dataToCompress);
					lock (_compressionData) {
						while (_compressionData.Count > 10) {
							Monitor.Wait (_compressionData, 100);
						}
						_compressionData.Enqueue (compressed);
						Monitor.Pulse (_compressionData);
					}
				}
			}
			_compressing = false;
		}

		private static byte[] Compress(byte[] data) {
			var memStream = new MemoryStream ();
			using(var compressionStream = new GZipStream(memStream, CompressionMode.Compress)) {
				compressionStream.Write(data, 0, data.Length);
			}
			return memStream.ToArray ();
		}

		private void FileWriter() {
			using (var stream = new FileStream("file.gz", FileMode.OpenOrCreate, FileAccess.Write)) {
				var workLeft = false;
				while (_compressing || workLeft) {
					workLeft = false;
					byte[] compressedData = null;
					lock (_compressionData) {
						while (_compressionData.Count == 0 && _compressing) {
							Monitor.Wait (_compressionData, 100);
						}
						workLeft = _compressionData.Count > 1;
						if (_compressionData.Count > 0) {
							compressedData = _compressionData.Dequeue ();
						}
					}
					if (compressedData != null) {
						stream.Write (compressedData, 0, compressedData.Length);
					}
				}
			}
		}
	}

Example Pipeline Processing - a more concise and language friendly file compression

	public class ConcisePipelineComputation
	{
		public ConcisePipelineComputation () {
		}

		public void PerformCompression() {
			var fileBlocks = new ThreadedList<byte[]>(FileReader());
			var compressedBlocks = new ThreadedList<byte[]> (Compression(fileBlocks));
			FileWriter (compressedBlocks);
		}

		private IEnumerable<byte[]> FileReader() {
			using (var stream = new FileStream("file.txt", FileMode.Open, FileAccess.Read)) {
				int len;
				var buffer = new byte[1024];
				while ((len = stream.Read(buffer, 0, buffer.Length)) > 0) {
					if (len != buffer.Length) {
						Array.Resize (ref buffer, len);
					}
					yield return buffer;
				}
			}
		}
		
		private IEnumerable<byte[]> Compression(IEnumerable<byte[]> readBuffer) {
			foreach (var buffer in readBuffer) {
				yield return Compress (buffer);
			}
		}

		private static byte[] Compress(byte[] data) {
			var memStream = new MemoryStream ();
			using(var compressionStream = new GZipStream(memStream, CompressionMode.Compress)) {
				compressionStream.Write(data, 0, data.Length);
			}
			return memStream.ToArray ();
		}

		private void FileWriter(IEnumerable<byte[]> compressedBuffer) {
			using (var stream = new FileStream("file.gz", FileMode.OpenOrCreate, FileAccess.Write)) {
				foreach (var buffer in compressedBuffer) {
					stream.Write (buffer, 0, buffer.Length);
				}
			}
		}
	}

Helper class - ThreadedList

	public class ThreadedList<T> : IEnumerable<T>
	{
		private readonly IEnumerable<T> _list;

		public ThreadedList (IEnumerable<T> list){
			_list = list;
		}

		public IEnumerator<T> GetEnumerator ()
		{
			return new ThreadedEnumerator<T>(_list.GetEnumerator ());
		}

		System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator ()
		{
			return GetEnumerator ();
		}

		private class ThreadedEnumerator<S> : IEnumerator<S> {

			private readonly IEnumerator<S> _enumerator;
			private readonly Queue<S> _queue;
			private const int _maxQueueSize = 10;
			private readonly Thread _thread;
			private volatile bool _keepGoing = true;
			private volatile bool _finishedEnumerating = false;
			private S _current;

			public ThreadedEnumerator(IEnumerator<S> enumerator) {
				_enumerator = enumerator;
				_thread = new Thread(Enumerate);
				_thread.Start();
			}

			private void Enumerate() {
				while (_keepGoing) {
					if (_enumerator.MoveNext ()) {
						var current = _enumerator.Current;
						lock (_queue) {
							while (_queue.Count > _maxQueueSize && _keepGoing) {
								Monitor.Wait (_queue, 100);
							}
							if (_keepGoing) {
								_queue.Enqueue (current);
								Monitor.Pulse (_queue);
							}
						}
					} else {
						break;
					}
				}
				_finishedEnumerating = true;
			}

			public bool MoveNext ()
			{
				lock (_queue) {
					while (!_finishedEnumerating && _queue.Count == 0) {
						Monitor.Wait (_queue, 100);
					}
					if (_queue.Count > 0) {
						_current = _queue.Dequeue ();
						Monitor.Pulse (_queue);
						return true;
					} else {
						_current = default(S);
						return false;
					}
				}
			}

			public void Reset () {
				lock (_queue) {
					lock (_enumerator) {
						_enumerator.Reset ();
						_queue.Clear ();
					}
				}
			}

			object System.Collections.IEnumerator.Current {
				get { return _current; }
			}

			public void Dispose () {
				_keepGoing = false;
				_thread.Join ();
			}

			public S Current {
				get { return _current; }
			}
		}
	}

Mutual Exclusion¶

Mutual exclusion is a general problem that applies to both processes and threads.
Processes
- Occurs with processes that share resources such as shared memory, files, and other resources that must be updated atomically
- When not otherwise shared, the address space of a process is protected against reads/writes by other processes
Threads
- Because threads share more resources such as having a shared process heap, there are more resources that need to be potentially protected
- Because the address space is shared among threads in one process, cooperation and coordination is required for threads that read from and write to shared data structures
When mutual exclusion is achieved, atomic operations on shared data structures are guaranteed to be atomic and not interrupted by other threads.

Tools for Achieving Mutual Exclusion¶

Mutex
- Has two operations: Lock() and Unlock()
- Has two states: Locked or Unlocked
- A lock can be acquired before entering a critical region and unlock can be called when leaving the critical region
- If all critical regions are covered by a mutex, then mutual exclusion has been achieved and operations can be said to be atomic
Semaphore
- Has two operations: Up() and Down()
- Has N states: a counter that has a value from 0 - N
- Up() increases the value by 1
- Down() decreases the value by 1
- When the semaphore has a value > 0, then a thread of execution can enter the critical region
- When the semaphore has a value = 0, then a thread is blocked
- The purpose of a semaphore is used to:
  
  Limit the number of threads that enter a critical region
  
  Limit the number of items in a queue between two threads working in a pipeline processing pattern.
Monitor
- Has four operations: Lock(), Unlock(), Pulse(), Wait()
- Allows for more complicated and user-coded conditions for entering critical regions
- The locking semantics are more complicated for the simplest cases, but can express more complicated mutual exclusion cases in simpler ways than can semaphores or mutexes
Additional details may be found in the Operating Systems course
- Mutual Exclusion - http://osdi.cs.courseclouds.com/html/mutualexclusion.html
- Deadlock - http://osdi.cs.courseclouds.com/html/deadlock.html

Mutex Example/Java¶

This code example shows how to implement a classic mutex, a.k.a. a Lock, in Java.

These examples come from http://hpjpc.googlecode.com by Christopher and Thiruvathukal.

public class Lock {
    protected boolean locked;

    public Lock() {
        locked = false;
    }

    public synchronized void lock() throws InterruptedException {
        while (locked)
            wait();
        locked = true;
    }

    public synchronized void unlock() {
        locked = false;
        notify();
    }
}

Semaphore Example¶

This shows how to implement a counting semaphore in the Java programming language.

public class Semaphore {
    /**
     * The current count, which must be non-negative.
     */
    protected int count;

    /**
     * Create a counting Semaphore with a specified initial count.
     *
     * @param initCount The initial value of count.
     * @throws com.toolsofcomputing.thread.NegativeSemaphoreException if initCount &lt; 0.
     */
    public Semaphore(int initCount) throws NegativeSemaphoreException {
        if (initCount < 0)
            throw new NegativeSemaphoreException();
        count = initCount;
    }

    /**
     * Create a counting Semaphore with an initial count of zero.
     */
    public Semaphore() {
        count = 0;
    }

    /**
     * Subtract one from the count. Since count must be non-negative, wait until
     * count is positive before decrementing it.
     *
     * @throws InterruptedException if thread is interrupted while waiting.
     */
    public synchronized void down() throws InterruptedException {
        while (count == 0)
            wait();
        count--;
    }

    /**
     * Add one to the count. Wake up a thread waiting to "down" the semaphore, if
     * any.
     */
    public synchronized void up() {
        count++;
        notify();
    }
}

Barrier¶

This shows how to implement a barrier, which is a synchronization mechanism for awaiting a specified number of threads before processing can continue. Once all threads have arrived, processing can continue.

public class SimpleBarrier {
    /**
     * Number of threads that still must gather.
     */
    protected int count;
    /**
     * Total number of threads that must gather.
     */
    protected int initCount;

    /**
     * Creates a Barrier at which n threads may repeatedly gather.
     *
     * @param n total number of threads that must gather.
     */

    public SimpleBarrier(int n) {
        if (n <= 0)
            throw new IllegalArgumentException(
                    "Barrier initialization specified non-positive value " + n);
        initCount = count = n;
    }

    /**
     * Is called by a thread to wait for the rest of the n threads to gather
     * before the set of threads may continue executing.
     *
     * @throws InterruptedException If interrupted while waiting.
     */
    public synchronized void gather() throws InterruptedException {
        if (--count > 0)
            wait();
        else {
            count = initCount;
            notifyAll();
        }
    }
}

Deadlock - a classic problem¶

A classic problem in computer science and one that is often studied in operating systems to show the hazards of working with shared, synchronized state, is the dining philosophers problem. We won’t describe the entire problem here but you can read http://en.wikipedia.org/wiki/Dining_philosophers_problem.

our “solution” has the following design:

Fork: A class to represent the forks shared by adjacent philosophers at the table.
Diner0: A class used to represent a philosopher. The philosopher does three things a philosopher normally does: think(), sleep(), and eat().
Diners0: A class used to represent all of the diners seated at the table with their shared forks. This is where the concurrency takes place.

Fork¶

public class Fork {
    public char id;

    private Lock lock = new Lock();

    public Fork(int value) {
        this.id = new Integer(value).toString().charAt(0);
    }

    public void pickup() throws InterruptedException {
        lock.lock();
    }

    public void putdown() throws InterruptedException {
        lock.unlock();
    }
}

Diner0¶

public class Diner0 extends Thread {
    private char state = 't';
    private Fork L, R;

    public Diner0(Fork left, Fork right) {
        super();
        L = left;
        R = right;
    }

    protected void think() throws InterruptedException {
        sleep((long) (Math.random() * 7.0));
    }

    protected void eat() throws InterruptedException {
        sleep((long) (Math.random() * 7.0));
    }

    public char getDinerState() {
        return state;
    }

    public void run() {
        int i;

        try {
            for (i = 0; i < 1000; i++) {
                state = 't';
                think();
                state = L.id;
                sleep(1);
                L.pickup();
                state = R.id;
                sleep(1);
                R.pickup();
                state = 'e';
                eat();
                L.putdown();
                R.putdown();
            }
            state = 'd';
        } catch (InterruptedException e) {
        }
    }
}

Diners0¶

public class Diners0 {
    private static Fork[] fork = new Fork[5];
    private static Diner0[] diner = new Diner0[5];

    public static void main(String[] args) {
        int i, j = 0;
        boolean goOn;

        for (i = 0; i < 5; i++) {
            fork[i] = new Fork(i);
        }

        for (i = 0; i < 5; i++) {
            diner[i] = new Diner0(fork[i], fork[(i + 1) % 5]);
        }

        for (i = 0; i < 5; i++) {
            diner[i].start();
        }

        int newPrio = Thread.currentThread().getPriority() + 1;

        Thread.currentThread().setPriority(newPrio);

        goOn = true;

        while (goOn) {
            for (i = 0; i < 5; i++) {
                System.out.print(diner[i].getDinerState());
            }

            if (++j % 5 == 0)
                System.out.println();
            else
                System.out.print(' ');

            goOn = false;

            for (i = 0; i < 5; i++) {
                goOn |= diner[i].getDinerState() != 'd';
            }

            try {
                Thread.sleep(51);
            } catch (InterruptedException e) {
                return;
            }
        }
    }
}

Diners1 - eliminating deadlock with resource enumeration¶

        for (i = 0; i < 4; i++) {
            diner[i] = new Diner0(fork[i], fork[(i + 1) % 5]);
        }
        diner[4] = new Diner0(fork[0], fork[4]);

Execution - With Deadlock¶

If you have Java and Gradle installed on your computer, you can try these out!

Make sure you have the HPJPC source code:

hg clone https://bitbucket.org/loyolachicagocs_books/hpjpc-source-java

The following Gradle task in build.gradle shows how to run Diners0’s main() method:

task Diners0(dependsOn: 'classes', type: JavaExec) {
    main = 'info.jhpc.textbook.chapter03.Diners0'
    classpath = sourceSets.main.runtimeClasspath
    args = []
}

To build:

gradle build

To run:

gradle Diners0

If you run this, you will notice that deadlock ensues fairly quick. The diners get into a state where they are waiting on each others’ forks in a cycle:

$ gradle Diners0
:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:Diners0
tet4t 023et 12ett 0et40 e134e
et340 12ett 12ett 123et e1340
0234e e23e4 1tt40 t23et ett40
t23et 1ett0 12et0 t23et 1tt40
1t34e 12et0 1et40 12e30 t234e
12e30 1et40 tetet et3t4 1t3e4
1e240 1tte4 12tt0 t2ete t2tt0
11e3t et3t0 t234e e1340 11t40
1t340 0e24e tttet tt34e 12e3t
1t24e 0t3e4 tet4e 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340
12340 12340 12340 12340 12340

Deadlock-Free Version¶

Diners1 has a similar Gradle task:

task Diners1(dependsOn: 'classes', type: JavaExec) {
    main = 'info.jhpc.textbook.chapter03.Diners1'
    classpath = sourceSets.main.runtimeClasspath
    args = []
}

Run:

gradle Diners1

Output:

$ gradle Diners1
:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:Diners1
ttttt 1t240 t2et4 1et4t tt2e4
1e2et 1et4t e13e0 tt3e4 ettt0
t2te0 0e24e 1ettt e1e3t t1te4
0tt4e 1etet e13tt tt24e 1t3et
tettt 0ttet ete3t tt33e 0et4e
1ete0 01t3t 0tt3e 1e240 te2te
e1et0 1e2et 02e3e t1t3t 1t3tt
02ete 1et4t e13et et33t 02tte
1ett0 et3t0 ete30 t2e3e et3e0
0et4e ettt0 0e2e4 01t4e 1e2et
...
12e30 tet3e 1etet 0ttt0 0etet
1et4t e2tt4 tt3e4 0t3et 12et0
1ett0 e1tet 12e30 1tttt etet0
tettt 1e2t0 0t3e4 tettt ttttt
023e4 ttttt 023e4 1e2d0 e13d0
02ed4 e2edt 1etd0 et3d0 1tedt
02ede 0etde 1etd0 t2tdt t2ede
01tde et2d0 112dt tedde tedd4
tedde 02dd0 1edd0 etdd0 1tddt
1eddt 1eddt 01dde 0tdd0 t2dde
t2ddt eddd4 tddde tddd4 tdddt
0ddde eddd0 tdddt 0ddde eddd0
tddd4 eddd0 0ddde 0ddde eddd0
eddd0 0ddde 0ddde tddd4 0dddt
eddd0 tddd4 1dddd 0dddd 1dddd
tdddd tdddd ddddd
BUILD SUCCESSFUL

Total time: 18.426 secs

The diners, as desired, end up finishing their meal after some time.

We assume they have moved over to the bar or found a nice place to have dessert.

Common Data Structures in Concurrent Programming¶

Bound Buffer
- Makes use of a mutex and semaphore internally
- Defines a maximum number of items that exist in the bound buffer’s queue.
- Has two operations: Enqueue() and Dequeue()
- Enqueue() - enqueues items in the data structure. If the enqueue operation would cause the bound buffer to exceed the maximum, the Enqueue() call will block until another thread dequeues at least one item.
- Dequeue() - dequeues an item from the data structure. If there are zero items in the queue, Dequeue() will block until another thread enqueues an item in the data structure
- Bound buffers are used to make sure that when one thread is producing work for a second thread, that if one thread is faster or slower than the other, that they appropriately wait to some extent for each other.
Example Bound Buffer

	public class BoundBuffer<T> {
		private readonly Semaphore _full;
		private readonly Semaphore _empty;
		private readonly Semaphore _lock;
		private readonly Queue<T> _queue;

		public BoundBuffer (int maxCount) {
			_empty = new Semaphore (maxCount, maxCount);
			_full = new Semaphore (0, maxCount);
			_lock = new Semaphore (1, 1);
			_queue = new Queue<T> ();
		}

		public void Enqueue(T item) {
			_empty.WaitOne ();
			_lock.WaitOne ();
			_queue.Enqueue (item);
			_lock.Release (1);
			_full.Release (1);
		}

		public T Dequeue() {
			_full.WaitOne ();
			_lock.WaitOne ();
			var item = _queue.Dequeue ();
			_lock.Release(1);
			_empty.Release(1);
			return item;
		}
	}

Design Considerations¶

Threading requires the support of the operating system - a threading library / package is needed
- In Windows, this is a part of the Windows SDK and .NET Framework
- In Linux and Mac OSX, PThreads provides threading
Thread usage and creation
- Threads can be started and stopped on demand or a thread pool can be used
- Starting threads dynamically:
  
  Has some cost associated with asking the OS to create and schedule the thread
  
  It can be architecturally challenging to maintain an appropriate number of threads across software components
  
  This is overall the most simple approach
- Thread Pools
  
  The number of threads can be defined at compile time or when the program is first launched
  
  Instead of creating a new thread, the program acquires a thread and passes a function pointer to the thread to execute
  
  When the given task is completed, the thread is returned to the pool.
  
  This approach does not have the overhead of creating / destroying threads as threads are reused.
  
  This approach often requires library support or some additional code.
The total number of threads
- Having several hundred threads on a system with an order of magnitude fewer cores can cause you to run into trouble.
- If a majority of those threads are runnable, then the program will spend most of its time context switching between those threads rather than actually getting work done.
- If such a system is dynamically starting and stopping threads, then the program will most likely spend most of its time creating and destroying threads.

Kernel Threads vs User Mode Threads¶

There are two types of threads:
- Kernel Threads
  -Supported by modern operating systems -Scheduled by the operating system
- User Threads
  -Supported by almost everything -Scheduled by the process
Context switching:
- Kernel threads have a higher overhead because the scheduler must be invoked and there might be a time lag before a runnable thread is actually executed.
- Kernel threads often perform very well because the operating system has more information about the resource state of the computer and can make better global scheduling decisions than can a program
- User-mode threads can context switch with fewer overall operations, but scheduling them is guess-work.
- User mode threads can be created more rapidly because new stacks and scheduler entries do not need to be created by the operating system
Where are user-mode threads used?
- In systems without kernel mode threads
- When the number of threads a system needs is in the hundreds or thousands (user-mode threads scale better in these scenarios)
Where are kernel-mode threads used?
- When the number of threads is not very high (less than 10 per core)
- When blocking calls are involved (user-mode thread libraries usually have separate I/O libraries)

Concurrent File Copy Example¶

FileCopy0: The sequential version
FileCopy1: The concurrent version

Sequential File Copy¶

public class FileCopy0 {
    public static final int BLOCK_SIZE = 4096;

    public static void copy(String src, String dst) throws IOException {
        FileReader fr = new FileReader(src);
        FileWriter fw = new FileWriter(dst);
        char[] buffer = new char[BLOCK_SIZE];
        int bytesRead;
        while (true) {
            bytesRead = fr.read(buffer);
            System.out.println(bytesRead + " bytes read");
            if (bytesRead < 0)
                break;
            fw.write(buffer, 0, bytesRead);
            System.out.println(bytesRead + " bytes written");

        }
        fw.close();
        fr.close();
    }

    public static void main(String[] args) {
        String srcFile = args[0];
        String dstFile = args[1];
        try {
            copy(srcFile, dstFile);
        } catch (Exception e) {
            System.out.println("Copy failed.");
        }
    }
}

Concurrent File Copy Organization¶

Quick overview of the various classes:

Pool: Maintains a list of buffers that can be used for allocating/freeing blocks of data without triggering new (or dispose) repeatedly.
Buffer: Used as a shared object for reading and writing blocks of data (via the FileCopyReader1 and FileCopyWriter1 classes)
BufferQueue: Used to queue up blocks as they are read or written. This allows for varying speeds of reader and writer, subject to the number of blocks available in the Pool.
FileCopyReader1: Used to run the reader thread.
FileCopyWriter1: Used to run the writer thread.
FileCopy1: Used to act as a drop in replacement for FileCopy0. Sets up the reader and writer threads and then joins with both when the reading/writing are completed.

public class FileCopy1 {

    public static int getIntProp(Properties p, String key, int defaultValue) {

        try {
            return Integer.parseInt(p.getProperty(key));
        } catch (Exception e) {
            return defaultValue;
        }
    }

    public static void main(String[] args) {
        String srcFile = args[0];
        String dstFile = args[1];
        Properties p = new Properties();
        try {
            FileInputStream propFile = new FileInputStream("FileCopy1.rc");
            p.load(propFile);
        } catch (Exception e) {
            System.err.println("FileCopy1: Can't load Properties");
        }

        int buffers = getIntProp(p, "buffers", 20);
        int bufferSize = getIntProp(p, "bufferSize", 4096);

        System.out.println("source = " + args[0]);
        System.out.println("destination = " + args[1]);
        System.out.println("buffers = " + buffers);
        System.out.println("bufferSize = " + bufferSize);

        Pool pool = new Pool(buffers, bufferSize);
        BufferQueue copyBuffers = new BufferQueue();

        FileCopyReader1 src;

        try {
            src = new FileCopyReader1(srcFile, pool, copyBuffers);
        } catch (Exception e) {
            System.err.println("Cannot open " + srcFile);
            return;
        }

        FileCopyWriter1 dst;

        try {
            dst = new FileCopyWriter1(dstFile, pool, copyBuffers);
        } catch (Exception e) {
            System.err.println("Cannot open " + dstFile);
            return;
        }

        src.start();
        dst.start();

        try {
            src.join();
        } catch (Exception e) {
        }

        try {
            dst.join();
        } catch (Exception e) {
        }
    }
}

public class FileCopyReader1 extends Thread {

    private FileReader fr;
    private Pool pool;
    private BufferQueue copyBuffers;

    public FileCopyReader1(String filename, Pool pool, BufferQueue copyBuffers)
            throws IOException {
        this.pool = pool;
        this.copyBuffers = copyBuffers;
        fr = new FileReader(filename);
    }

    public void run() {
        Buffer buffer;
        int bytesRead = 0;

        do {
            try {
                buffer = pool.use();
                bytesRead = fr.read(buffer.getBuffer());
            } catch (Exception e) {
                buffer = new Buffer(0);
                bytesRead = 0;
            }
            if (bytesRead < 0) {
                buffer.setSize(0);
            } else {
                buffer.setSize(bytesRead);
            }
            copyBuffers.enqueueBuffer(buffer);
        } while (bytesRead > 0);

        try {
            fr.close();
        } catch (Exception e) {
            return;
        }
    }
}

public class FileCopyWriter1 extends Thread {

    private FileWriter fw;
    private Pool pool;
    private BufferQueue copyBuffers;

    public FileCopyWriter1(String filename, Pool pool, BufferQueue copyBuffers)
            throws IOException {

        this.pool = pool;
        this.copyBuffers = copyBuffers;
        fw = new FileWriter(filename);
    }

    public void run() {
        Buffer buffer;

        while (true) {
            try {
                buffer = copyBuffers.dequeueBuffer();
            } catch (Exception e) {
                return;
            }
            if (buffer.getSize() > 0) {
                try {
                    char[] bufferData = buffer.getBuffer();
                    int size = bufferData.length;
                    fw.write(bufferData, 0, size);
                } catch (Exception e) {
                    break;
                }
                pool.release(buffer);
            } else
                break;
        }

        try {
            fw.close();
        } catch (Exception e) {
            return;
        }
    }
}

public class Buffer {
    private char[] buffer;
    private int size;

    public Buffer(int bufferSize) {
        buffer = new char[bufferSize];
        size = bufferSize;
    }

    public char[] getBuffer() {
        return buffer;
    }

    public int getSize() {
        return size;
    }

    public void setSize(int newSize) {
        if (newSize > size) {
            char[] newBuffer = new char[newSize];
            System.arraycopy(buffer, 0, newBuffer, 0, size);
            buffer = newBuffer;
        }

        size = newSize;
    }
}

public class BufferQueue {
    public Vector<Buffer> buffers = new Vector<Buffer>();

    public synchronized void enqueueBuffer(Buffer b) {
        if (buffers.size() == 0)
            notify();
        buffers.addElement(b);
    }

    public synchronized Buffer dequeueBuffer() throws InterruptedException {
        while (buffers.size() == 0)
            wait();

        Buffer firstBuffer = buffers.elementAt(0);
        buffers.removeElementAt(0);
        return firstBuffer;
    }
}

public class Pool {

    Vector<Buffer> freeBufferList = new Vector<Buffer>();
    OutputStream debug = System.out;
    int buffers, bufferSize;

    public Pool(int buffers, int bufferSize) {
        this.buffers = buffers;
        this.bufferSize = bufferSize;
        freeBufferList.ensureCapacity(buffers);

        for (int i = 0; i < buffers; i++)
            freeBufferList.addElement(new Buffer(bufferSize));
    }

    public synchronized Buffer use() throws InterruptedException {
        while (freeBufferList.size() == 0)
            wait();
        Buffer nextBuffer = freeBufferList.lastElement();
        freeBufferList.removeElement(nextBuffer);
        return nextBuffer;
    }

    public synchronized void release(Buffer oldBuffer) {
        if (freeBufferList.size() == 0)
            notify();

        if (freeBufferList.contains(oldBuffer))
            return;

        if (oldBuffer.getSize() < bufferSize)
            oldBuffer.setSize(bufferSize);

        freeBufferList.addElement(oldBuffer);
    }
}

Execution¶

You can run FileCopy0 and FileCopy1 by using the corresponding Gradle tasks.

task FileCopy0(dependsOn: 'classes', type: JavaExec) {
    main = 'info.jhpc.textbook.chapter03.FileCopy0'
    classpath = sourceSets.main.runtimeClasspath
    args = [fc0_src, fc0_dest]
}

As shown, there are two properties you can set: fc0_src and fc0_dest:

gradle FileCopy0 -Pfc0_src=inputFile -Pfc0_dest=outputFile

You can also run FileCopy1 (the same parameter names are used):

gradle FileCopy1 -Pfc0_src=inputFile -Pfc0_dest=outputFile