• Some Observations:
  • Multi-threaded programs often make use of collections of various kinds
  • For performance reasons, collections are not, in general thread-safe
• Fortunately:
  • Various syncrhonized collections have been implemented
• Unfortunately:
  • They sometimes suffer from scalability problems because they can only be read from/written to by one thread at a time

• The Idea:
  • Provide coordination in other ways
• The Approaches:
  • Copy-on-Write (COW)
  • Compare-and-Swap (CAS)
  • Fine-Grained-Lock (FGL)

• The Approach:
  • The values in the collection are stored in an immutable array
  • A new copy of the collection is created every time it is modified
  • Snaphot iterators retain a reference to the backing array that was current at the start of iteration (which will never change)
• Examples:
  • CopyOnWriteArrayList
  • CopyOnWriteArraySet

• The Approach:
  • Before updating, a local copy of the variable is made
  • Calculations are performed using the local copy
  • The local copy and the variable are compared - if they are the same the update is performed, otherwise it either gives up or tries again
  • Provide weakly consistent iterators that tolerate concurrent modification, traverse elements as they existed when the iterator was constructed, and may (but are not guaranteed to) reflect modifications to the collection after the construction of the iterator
• Tradeoffs:
  • Eliminates many liveness defects
  • Can cause livelock
• Examples:
  • ConcurrentLinkedQueue
  • ConcurrentLinkedDeque
  • ConcurrentSkipListMap

• The Approach:
  • Use multiple (fine-grained) locks (e.g., lock striping) to make individual methods atomic without limiting access to all methods
  • Sometimes exploit details of the Java Memory Model to minimize the time that locks are held (or avoid acquiring locks)
• Tradeoffs:
  • Multiple threads can access the collection at the same time
  • It is harder to implement methods that operate on the collection as a whole (e.g., size(), isEmpty()) since all of the locks would have to be acquired
• Examples:
  • ConcurrentHashMap
  • LinkedBlockingQueue

• Objectives:
  • Full concurrency of retrievals
  • High expected concurrency of updates
  • Retrieval operations and update operations may overlap (i.e., an update for a key happens-before any non-null retrieval for that key)
• Retrievals:
  • Usually do not block
  • Usually return the value inserted by the most recent completed insert operation and may return a value added by a concurrent insertion
• Weakly Consistent Iterators:
  • Return each element once at most once
  • Never throw ConcurrentModificationException
  • May or may not reflect insertions or removals that occurred since the iterator was constructed

• Additional Atomic Methods (as per ConcurrentMap ):
  • putIfAbsent(k, v)
  • replace(k, v) and replace(k, v, n)
  • remove(k, v)
• Computational Methods (as per ConcurrentMap ):
  • Examples - compute(), forEach()
  • Approach - use functional programming techniques

• Objective (as per the BlockingQueue interface):
  • First-in-first-out (possibly space-bounded) collection
  • Wait for the queue to become non-empty when retrieveing
  • Wait for the queue to have space when storing
• Approach:
  • Separates the creation of work from its execution (e.g., using a "to do" list)
• Method Summary:

  • 	Throws Exception	Returns Special	Blocks Until	Times Out
    Insert	add(e)	offer(e)	put(e)	offer(e, time, timeunit)
    Remove	remove()	poll()	take()	poll(time, timeunit)
    Examine	element()	peek()

• The Setting:
  • In an application that dispatches emergency vehicles, a single Dispatcher is used by a DailyDispatchHandler and a RealTimeDispatchHandler, each of which executes in its own thread
• The Previous Implementation:
  • The thread executing the dispatch() method enters the waiting state if no vehicles are available
  • The thread that executes makeVehicleAvailable() notifies the waiting threads
• This Implementation:
  • The Dispatcher uses a LinkedBlockingQueue

javaexamples/threads/v6b/Dispatcher.java

import java.util.concurrent.*;


/**
 * Used to dispatch vehicles in a fleet (e.g., emergency
 * response vehicles).
 *
 * This version uses a LinkedBlockingQueue
 *
 * @author  Prof. David Bernstein, James Madison University
 * @version 6.0b
 */
public class Dispatcher
{
    protected int                             numberOfVehicles;
    protected LinkedBlockingQueue<Integer>    availableVehicles;
    

    /**
     * Construct a new Dispatcher
     *
     * @param n   The number of vehicles in the fleet
     */
    public Dispatcher(int n)
    {
       int     i;

       numberOfVehicles = n;
       availableVehicles = new LinkedBlockingQueue<Integer>();

       for (i=0; i < n; i++)
       {
          makeVehicleAvailable(i);
       }
    }

    /**
     * Dispatch an available vehicle
     *
     * @param  task  The task to be handled
     * @return       true if a vehicle was dispatched
     */
    public boolean dispatch(String task)
    {
       int      vehicle;
       Integer  v;

       v = availableVehicles.take();  // Wait if necessary
       vehicle = v.intValue();
       sendMessage(vehicle, task);

       return true;
    }

    /**
     * Makes a vehicle available for future dispatching
     *
     * @param vehicle  The number of the vehicle
     */
    public void makeVehicleAvailable(int vehicle)
    {
       availableVehicles.put(new Integer(vehicle));
    }

    /**
     * Sends a message to a vehicle
     *
     * @param vehicle  The number of the vehicle
     * @param message  The message to send
     */
    private void sendMessage(int vehicle, String message)
    {

       // This method would normally transmit the message 
       // to the vehicle.  For simplicity, it now writes it 
       // to the screen instead.

       System.out.println(vehicle+"\t"+message+"\n");
       System.out.flush();       
    }
}
        

• ArrayBlockingQueue :
  • Bounded (because it is backed by a fixed-size array)
  • Supports a fairness policy that orders waiting producer and concumer threads in FIFO order (which decreases throughput but avoids starvation)
• LinkedBlockingDeque (pronounced like "deck"):
  • Provides the capabilities of a doubly-ended queue (i.e., insertion/removal at/from both ends)
• PriorityBlockingQueue
  • Elements are ordered either based on their natural ordering or a Comparator ; the head is the least element