Thread Synchronization

Created

Thread Syncrhonization

const int NUM_INCR=100000000, NTHREADS=2;
typedef struct { volatile int count; } Shared;

void *worker(void *arg) {
	Shared *obj = arg;
	for (int i = 0; i < NUM_INCR/NTHREADS; i++)
		obj->count++;
	return NULL;
}

int main(void) {
	Shared *obj = calloc(1, sizeof(Shared));
	pthread_t threads[NTHREADS];
	for (int i = 0; i < NTHREADS; i++)
		pthread_create(&threads[i], NULL, worker, obj);
	for (int i = 0; i < NTHREADS; i++)
		pthread_join(threads[i], NULL);
	printf("%d\n", obj->count);
	return 0;
}
  • Uses two threads, which increment a shared counter.
  • The counter is incremeneted 100,000,000 times from zero. However, after running the program we get 53015619.

Incrementing the counter obj->count++ is not atomic. What var++ is actually doing is

reg = var;
reg = reg + 1;
var = reg;

When threads are executing concurrently, it’s possible for the variable to change between the time its value is loaded and the time the updated value is stored

  • data race causes a lost update.

Concurrent access to shared data can screw up things as simple as integer counter. Imagine the mess it will make of your linked list, balanced tree, etc. This is because data structures have invariants that must be preserved:

  • Mutations (insertions and removals) often violate these invariants temporarily. In a sequential program, it's not a problem because the operation will complete (and restore the invariants) before anyone notice. However, in a concurrent program where multiple threads could acces the data structure, this can be a huge problem. Think of what happens if another thread accesses the data structure when it is int he middle of insertion/removal.
  • Synchronization protects this shared data from concurrent access.

Semaphores and Mutexes.

Critical Sections

A critical section is a region of code in which mutual exclusion must be guaranteed for correct behavior. Sections of code that we want them to do one at a time because this region of code that accesses and/or modifies shared data.

  • Mutual exclusion means that at most one concurrent task (thread) may be accessing shared data at any given time
  • Enforcing mutual exclusion in critical sections guarantees atomicity. This means this code in critical section can execute to completion without interruption.
  • For the shared counter example above, the update to shared counter variable is a critical section.

Semaphores and mutexes are two types of synchronization constructs available in pthreads. They both are used to guarantee mutual exclusion.

  • Semaphores can also be used to manage access to a finite resource.
  • Mutexes (a.k.a., “mutual exclusion locks”) are simpler.

Mutexes:

Commands pthread_mutex_t: data type for a pthreads mutex

pthread_mutex_init: initialize a mutex

pthread_mutex_lock: locks a mutex for exclusive access

  • If another thread has already locked the mutex, calling thread must wait
  • Call this at the beginning of the critical section

pthread_mutex_unlock: unlocks a mutex

  • If any threads are waiting to lock the mutex, one will be woken up and allowed to acquire it

pthread_mutex_destroy: destroys a mutex (once it is no longer needed)

Only one thread at a time can lock a mutex at a single point of time. If no one has locked it, then it can lock. Otherwise if the mutex is already locked, then it will have to wait before it can lock it.

To use a mutex to protect a shared data structure:

  • associate a pthread_mutex_t variable with each instance of the data strucutre.
  • Initialize with pthread_mutex_init when the data structure is initialized. Do this in the objects constructor.
  • Each critical section is protected with calls to pthread_mutex_lock and pthread_mutex_unlock.
  • Destroy mutex pthread_mutex_destroy when data structure is deallocated. Do this in the object's deconstructor.

Updated shared counter program with mutex:

typedef struct {
	volatile int count;      
	pthread_mutex_t lock;
} Shared;
void *worker(void *arg) {
	Shared *obj = arg;
	for (int i = 0; i < NUM_INCR/NTHREADS; i++) {
		pthread_mutex_lock(&obj->lock);
		obj->count++;
		pthread_mutex_unlock(&obj->lock);
	}	
	return NULL;
}
int main(void) {
	Shared *obj = calloc(1, sizeof(Shared));      
	pthread_mutex_init(&obj->lock, NULL);      
	pthread_t threads[NTHREADS];
	for (int i = 0; i < NTHREADS; i++)
		pthread_create(&threads[i], NULL, worker, obj);
	for (int i = 0; i < NTHREADS; i++)
		pthread_join(threads[i], NULL);
	printf("%d\n", obj->count);
	pthread_mutex_destroy(&obj->lock);      
	return 0;
}
  • This version is slower but the counter is correct.

Contention

Contention occurs when multiple threads try to access the same shared data structure at the same time.

Costs associated with synchronization:

  1. Cost of entering and leaving critical section (e.g., locking and unlocking a mutex)
  2. Reduced parallelism due to threads having to take turns (when contending for access to shared data). When the thread is waiting, it is wasted resources as it could be doing something useful instead. So, we want threads to do as much sequential work.
  3. Cost of OS kernel code to suspend and resume threads as they wait to enter critical sections.

These costs can be significant. Best performance occurs when threads synchronize relatively infrequently.

Guard

In C++, we can use guard objects to create critical sections

  • A guard object has a reference to a mutex
  • Its constructor locks the mutex
  • Its destructor unlocks the mutex

The advantage is that its simplier and it guarantees that the mutex will be released because the lifetime of the guard object is the extent of the critical section.

  • Avoids deadlocks due to mutex not being released (e.g., because of control flow, an exception, etc.)

Implementation

The Guard object

class Guard {
public:
	Guard(pthread_mutex_t &lock)
		: lock(lock) {
		pthread_mutex_lock(&lock);
	}
	
	~Guard() {
		pthread_mutex_unlock(&lock);
	}

private:
	Guard(const Guard &);
	Guard &operator=(const Guard &);
	pthread_mutex_t &lock;
};

Using the guard object to define the critical section:

// Assume m_lock is a mutex
{  
	Guard g(m_lock);  
	// beginning of critical section
	// ...code of critical section...
	// end of critical section
}
  • The braces define the scope (and lifetime) of the guard object.

This approach is highly recommended.

Semaphores

A semaphore is a more general synchronization construct, invented by Edsger Dijkstra in the early 1960s.

When created, semaphore is initialized with a nonnegative integer count value.

Two operations:

  • P (“proberen”): waits until the semaphore has a non-zero value, then decrements the count by one
  • V (“verhogen”): increments the count by one, waking up a thread waiting to perform a P operation if appropriate

A mutex can be modeled as a semaphore whose initial value is 1

Semaphores in pthread:

  • Include the <semaphore.h> header file
  • Semaphore data type is sem_t
  • Functions:
    • sem_init: initialize a semaphore with specified initial count
    • sem_destroy: destroy a semaphore when no longer needed
    • sem_wait: wait and decrement (P)
    • sem_post: increment and wake up waiting thread (V)

Applications

Semaphores are useful for managing access to a limited resource

Example: limiting maximum number of threads in a server application

  • Initialize semaphore with desired maximum number of threads
  • Use P operation before creating a client thread
  • Use V operation when client thread finishes

Say we only want 100 threads to run. Everytime a new thread is run the counter decrements from 100 to zero. Then any new thread can't start up.

Bounded queue:

  • Initially empty, can have up to a fixed maximum number of elements. When enqueuing an item, thread waits until queue is not full. When dequeuing an item, thread waits until queue is not empty.
  • To implement this: utilize two semaphores and one mutex.
    • slots semaphore: tracks how many slots are available
    • items semaphore: tracks how many elements are present
    • Mutex is used for critical sections accessing queue data struct. Coordinate access to the queue data structure itself.
  • Data structures:

Bounded queue of generic (void *) pointers Bounded queue data type:

typedef struct {
  void **data;
  unsigned max_items, head, tail;
  sem_t slots, items;
  pthread_mutex_t lock;

} BoundedQueue;

Bounded queue operations:

BoundedQueue *bqueue_create(unsigned max_items);
void bqueue_destroy(BoundedQueue *bq);
void bqueue_enqueue(BoundedQueue *bq, void *item);
void *bqueue_dequeue(BoundedQueue *bq);

Creating bounded queue: The slots semaphore initialized with max number of items, and items semaphore initialized to 0

BoundedQueue *bqueue_create(unsigned max_items) {
	BoundedQueue *bq = malloc(sizeof(BoundedQueue));
	bq->data = malloc(max_items * sizeof(void *));
	bq->max_items = max_items;
	bq->head = bq->tail = 0;
	sem_init(&bq->slots, 0, max_items);
	sem_init(&bq->items, 0, 0);
	pthread_mutex_init(&bq->lock, NULL);
	return bq; 
}
  • The 0 in the sem_init function indiciates if it is shared.

Enqueuing an item: Slots decreases (must wait until nonzero before new item can be added), items increases

Queue implemented as a “circular” array of pointers: head refers to where next item will be added, tail refers to where next item will be removed

void bqueue_enqueue(BoundedQueue *bq, void *item) {
	// wait for empty slot
	sem_wait(&bq->slots);

	// put item into the queue
	pthread_mutex_lock(&bq->lock);
	bq->data[bq->head] = item;
	bq->head = (bq->head + 1) % bq->max_items;
	pthread_mutex_unlock(&bq->lock);

	// item is available
	sem_post(&bq->items);
}
  • Common way to access a shared data structure

Dequeuing an Item Items decreases (must wait until nonzero before item can be removed), slots increases. Very similar to enqueuing

    void *bqueue_dequeue(BoundedQueue *bq) {
      sem_wait(&bq->items); /* wait for item */
      pthread_mutex_lock(&bq->lock);
      void *item = bq->data[bq->tail];
      bq->tail = (bq->tail + 1) % bq->max_items;
      pthread_mutex_unlock(&bq->lock);
      sem_post(&bq->slots); /* empty slot is available */
      return item;
}

Usefullness Synchronized queues are extremely useful in multithreaded programs!
In particular they are useful for producer/consumer relationships between threads

  • Producer enqueues items
  • Consumer dequeues items
  • Bounded queue: ensures that producer doesn’t get too far ahead of consumer

More generally, a queue can be used to send a message to another thread

Prethreading

Creating threads incurs some overhead

Prethreading: program creates a fixed number of threads ahead of time, assigns work to them as it becomes available

Queues are an ideal mechanism to allow the “master” thread to send work to the worker threads

A queue can also be used for messages sent from the workers back to the master thread

Conway's game of line

Simulation:

  • Grid-based cellular automaton, cells are alive (1) or dead (0) I Live cells with 2 or 3 live neighbors survive
  • Dead cells with 3 live neighbors become alive
  • Otherwise, cell dies (or stays dead)
  • Over many generations, complex patterns can emerge

Conway’s game of life is not quite an embarrassingly parallel computation - Computation of generation n must finish before computation of generation n + 1 can start

Could start a new batch of worker threads each generation

  • But we’ll repeatedly pay the thread startup and teardown costs

Prethreading approach:

  • Create fixed set of worker threads
  • “Command queue” allows master thread to send tasks to the workers
  • “Done queue” allows workers to notify master thread when tasks are finished

Analysis: Compared to the sequential method, using four threads gets a 2x speedup.

  • Relatively large chunks of work were assigned. Costs of synchronization amortized over relatively large amounts of sequential computation done by worker threads
  • Queues are an effective mechanism for communication between threads

Linked from Computer System Fundamentals