COMS 4995 Advanced Systems Programming

Synchronization

Semaphore Overview

Semaphore is an integer value manipulated by two methods:

1. Increment: increase the integer
   • a.k.a. verhoog, V(), up, sem_post()
2. Decrement: wait until value > 0, then decrease the integer value
   • a.k.a. probeer, P(), down, sem_wait()
   • Note blocking semantic: unlike increment (which can run whenever), decrement blocks until value is positive.

Initial value affects semaphore semantics as follows.

Binary Semaphore

Binary semaphore (a.k.a. lock) has initial value 1, protecting one resource:

1. Before acquiring resource, run sem_wait(). Value decremented to 0.
2. Use resource
3. Run sem_post() to release the resource. Value incremented to 1.

Resource is limited to 1 user at a time. Concurrent access while resource is locked sees value of 0, blocks, and is woken up when value is incremented back to 1.

Counting Semaphore

Counting semaphore has initial value N > 1, protecting N resources.

1. Before acquiring resource, run sem_wait(). Value is decremented by 1.
2. Use resource
3. Run sem_post() to release the resource. Value incremented by 1.

Since there are N resources, concurrent access is blocked only after N concurrent accesses (i.e. when the value hits 0). N + 1th concurrent access sees value of 0, blocks, is woken up when the value becomes positive again (i.e. some users posted the semaphore).

Ordering Semaphore

Ordering semaphore takes advantage of blocking semantic to implement “events”. e.g.:

sem = 0  // initial value is 0

P1: X -> sem_wait() -> Y

P2: A -> B -> sem_post()

P1 completes task X, then blocks until P2 completes tasks A & B before moving on to task Y, which presumably depends on the completion of tasks A & B. P1 has to wait until P2 increments the semaphore value.

POSIX API

Unnamed Semaphore

Initializing and destroying unnamed POSIX semaphores:

#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);
        // Returns: 0 if OK, –1 on error

int sem_destroy(sem_t *sem);
        // Returns: 0 if OK, –1 on error

sem_t *sem: Pointer to semaphore object.

int pshared: If semaphore is meant to be shared with child processes, pass in non-zero value. Otherwise (i.e., shared among threads), pass in 0.

unsigned int value: Initial value for the semaphore.

If unnammed semaphore is to be shared with child processes, where should semaphore be declared?

• Can the semaphore simply be declared globally?
• Recall shared anonymous memory mappings from mmap()

Named semaphore

Creating, opening, closing, and removing named POSIX semaphores:

#include <semaphore.h>

sem_t *sem_open(const char *name, int oflag, ...
                /* mode_t mode, unsigned int value  */ );
        // Returns: Pointer to semaphore if OK, SEM_FAILED on error

int sem_close(sem_t *sem);
        // Returns: 0 if OK, –1 on error

int sem_unlink(const char *name);
        // Returns: 0 if OK, –1 on error

Similar semantics to file API syscalls. Named semaphores are meant to be used by unrelated processes, using the file name as a “redezvous” point. On Linux, named semaphores are stored in the filesystem under /dev/shm. We’ll revisit named semaphores when we cover IPC mechanisms.

Incrementing/Decrementing Semaphores

#include <semaphore.h>
#include <time.h>

int sem_post(sem_t *sem);

int sem_wait(sem_t *sem);

int sem_trywait(sem_t *sem);

int sem_timedwait(sem_t *restrict sem,
                    const struct timespec *restrict tsptr);

// All functions return 0 if OK, –1 on error

Blocking semantics:

• sem_post() never blocks
• sem_wait() blocks until semaphore value is positive
  • Sets errno to EINTR if interrupted by a signal
• sem_trywait() does NOT block, fails immediately if semaphore value is 0
  • Sets errno to EAGAIN if the semaphore is unavailable
• sem_timedwait() is the same as sem_wait() but with a timeout value
  • Sets errno to ETIMEDOUT if sem_timedwait() returns due to the timeout

Producer-Consumer Problem

Overview

Consider a workflow where producers generate and enqueue workloads into some data structure, and consumers dequeue and process them. A ring buffer is commonly used to implement the data structure for this problem. A ring buffer is a fixed-size array where the last element of the array conceptually wraps around to the first element of the array. As such, one must track the indices of the first and last logical elements in the array as they continuously shift and wrap around. The following diagram demonstrates using a ring buffer for the producer-consumer problem:

The producer index points to the next empty slot in the ring buffer, and the consumer index points to the first filled slot containing the next workload.

The producers shouldn’t enqueue new workloads if the buffer is full and the consumers cannot dequeue workloads if the buffer is empty. This means that we’ll have to track the number of empty/filled slots, and the producers and consumers need to block accordingly.

Producer-Consumer with Semaphore

We can implement these semantics with semaphores as follows:

uint32_t buffer[N];

sem_t empty; // Number of empty slots
sem_t full;  // Number of filled slots
sem_t guard; // Mutex for ring buffer access

uint32_t pi = 0; // Producer index
uint32_t ci = 0; // Consumer index

int main() {
    // ...
    sem_init(&empty, 0, N);
    sem_init(&full,  0, 0);
    sem_init(&guard, 0, 1);
    // ...
}

void enqueue(uint32_t piece)
{
    sem_wait(&empty);
    sem_wait(&guard);

    buffer[pi] = piece;
    pi = (pi + 1) % N;
    
    sem_post(&guard);
    sem_post(&full);
}

uint32_t dequeue() 
{
    uint32_t piece;
    sem_wait(&full);
    sem_wait(&guard);

    piece = buffer[ci];
    ci = (ci + 1) % N;
    
    sem_post(&guard);
    sem_post(&empty);
    return piece;
}

See producer-consumer.c for a complete running example.

Producer-Consumer with Condition Variable

We can implement the producer-consumer semantics similarly using condition variables as follows:

pthread_cond_t  empty_cond;
pthread_cond_t  full_cond;
pthread_mutex_t mutex;

// ...

void enqueue(uint32_t piece) {
    pthread_mutex_lock(&mutex);

    // Wait until there's space in the ring buffer
    while (is_full())
        pthread_cond_wait(&empty_cond, &mutex);

    // ...

    pthread_cond_signal(&full_cond);
    pthread_mutex_unlock(&mutex);
}

---

Last updated: 2024-02-11