Lab6 - Mutex Lock

Race Condition

Process

#include <stdbool.h>
#include <stdio.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define ARRAY_SIZE 5
struct Item {
  int buffers[ARRAY_SIZE];
  int in;
  int out;
  int counter;
};
int main() {
  // Create shared array
  struct Item *shared_items =
      mmap(NULL, sizeof(struct Item), PROT_READ | PROT_WRITE,
           MAP_SHARED | MAP_ANONYMOUS, -1, 0);

  if (fork() == 0) {
    // Child process
    int value = 10;
    for(int i=0;i<100000;i++) {
      while (shared_items->counter == ARRAY_SIZE)
        ;
      printf("produce: %d\n", value);
      shared_items->buffers[shared_items->in] = value;
      shared_items->in = (shared_items->in + 1) % ARRAY_SIZE;
      shared_items->counter = shared_items->counter + 1;
      value += 10;
      // sleep(3);
    }

  } else {
    // Parent process
    // sleep(1);
    for(int i=0;i<100000;i++) {
      while (shared_items->counter == 0)
        ;
      int value = shared_items->buffers[shared_items->out];
      printf("consume: %d\n", value);
      shared_items->out = (shared_items->out + 1) % ARRAY_SIZE;
      shared_items->counter = shared_items->counter - 1;
      // sleep(1);
    }

    wait(NULL); // Wait for child
    munmap(shared_items, sizeof(struct Item));
  }

  return 0;
}

OUTPUT

consume: 999900
produce: 999950
consume: 999910
produce: 999960
consume: 999920
produce: 999970
consume: 999930
produce: 999980
consume: 999940
produce: 999990
consume: 999950
produce: 1000000
consume: 999960
consume: 999970
consume: 999980
consume: 999990
^C
c@c:~/cos3105$

Thread

#include <stdio.h>
#include <pthread.h>

int counter = 0;
// NO MUTEX!

void* increment_function(void* arg) {
    for (int i = 0; i < 100000; i++) {
        // NO LOCK!
        counter++;                           // Race condition!
        printf("Thread incremented counter to: %d\n", counter);
        // NO UNLOCK!
    }
    return NULL;
}

int main() {
    pthread_t thread1, thread2;

    pthread_create(&thread1, NULL, increment_function, NULL);
    pthread_create(&thread2, NULL, increment_function, NULL);

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    printf("\nFinal counter value: %d\n", counter);

    return 0;
}

OUTPUT

Thread incremented counter to: 199909
Thread incremented counter to: 199910
Thread incremented counter to: 199911
Thread incremented counter to: 199912
Thread incremented counter to: 199913
Thread incremented counter to: 199914
Thread incremented counter to: 199915
Thread incremented counter to: 199916
Thread incremented counter to: 199917
Thread incremented counter to: 199918
Thread incremented counter to: 199919
Thread incremented counter to: 199920
Thread incremented counter to: 199921
Thread incremented counter to: 199922
Thread incremented counter to: 199923

Final counter value: 199923
c@c:~/cos3105$

Mutex Lock

Process

#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <sys/mman.h>
#include <sys/wait.h>
#include <unistd.h>

#define ARRAY_SIZE 100000

struct Item {
  int buffers[ARRAY_SIZE];
  int in;
  int out;
  int counter;
  pthread_mutex_t lock;
};

int main() {
  struct Item *shared_items =
      mmap(NULL, sizeof(struct Item), PROT_READ | PROT_WRITE,
           MAP_SHARED | MAP_ANONYMOUS, -1, 0);

  pthread_mutexattr_t attr;
  pthread_mutexattr_init(&attr);
  pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
  pthread_mutex_init(&shared_items->lock, &attr);

  shared_items->counter = 0;
  shared_items->in = 0;   // Initialize
  shared_items->out = 0;  // Initialize

  if (fork() == 0) {
    // Child process (Producer)
    int value = 10;
    for (int i = 0; i < 100000; i++) {

      pthread_mutex_lock(&shared_items->lock);  // LOCK EARLIER!

      // Wait while buffer is full
      while (shared_items->counter == ARRAY_SIZE) {
        pthread_mutex_unlock(&shared_items->lock);  // Release while waiting
        pthread_mutex_lock(&shared_items->lock);    // Re-acquire
      }

      // Critical section - all buffer operations protected
      shared_items->buffers[shared_items->in] = value;
      shared_items->in = (shared_items->in + 1) % ARRAY_SIZE;
      shared_items->counter++;

      pthread_mutex_unlock(&shared_items->lock);  // UNLOCK AFTER ALL OPERATIONS

      printf("produce: %d, counter: %d\n", value, shared_items->counter);
      value += 10;
    }

  } else {
    // Parent process (Consumer)
    for (int i = 0; i < 100000; i++) {

      pthread_mutex_lock(&shared_items->lock);  // LOCK EARLIER!

      // Wait while buffer is empty
      while (shared_items->counter == 0) {
        pthread_mutex_unlock(&shared_items->lock);  // Release while waiting
        pthread_mutex_lock(&shared_items->lock);    // Re-acquire
      }

      // Critical section - all buffer operations protected
      int value = shared_items->buffers[shared_items->out];
      shared_items->out = (shared_items->out + 1) % ARRAY_SIZE;
      shared_items->counter--;

      pthread_mutex_unlock(&shared_items->lock);  // UNLOCK AFTER ALL OPERATIONS

      printf("consume: %d, counter: %d\n", value, shared_items->counter);
    }

    wait(NULL);
    printf("\nFinal counter: %d\n", shared_items->counter);
    pthread_mutex_destroy(&shared_items->lock);
    munmap(shared_items, sizeof(struct Item));
  }

  return 0;
}

OUTPUT

consume: 999860, counter: 14
consume: 999870, counter: 13
consume: 999880, counter: 12
consume: 999890, counter: 11
consume: 999900, counter: 10
consume: 999910, counter: 9
consume: 999920, counter: 8
consume: 999930, counter: 7
consume: 999940, counter: 6
consume: 999950, counter: 5
consume: 999960, counter: 4
consume: 999970, counter: 3
consume: 999980, counter: 2
consume: 999990, counter: 1
consume: 1000000, counter: 0

Final counter: 0
c@c:~/cos3105$

Thread

#include <stdio.h>
#include <pthread.h>

int counter = 0;           // Shared variable
pthread_mutex_t lock;      // Mutex lock

void* increment_function(void* arg) {
    for (int i = 0; i < 100000; i++) {

        pthread_mutex_lock(&lock);      // Lock (Entry section)

        counter++;                       // Critical section
        printf("Thread incremented counter to: %d\n", counter);

        pthread_mutex_unlock(&lock);    // Unlock (Exit section)
    }
    return NULL;
}

int main() {
    pthread_t thread1, thread2;

    // Step 1: Initialize the mutex
    pthread_mutex_init(&lock, NULL);

    // Step 2: Create two threads
    pthread_create(&thread1, NULL, increment_function, NULL);
    pthread_create(&thread2, NULL, increment_function, NULL);

    // Step 3: Wait for both threads to finish
    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    // Step 4: Print final result
    printf("\nFinal counter value: %d\n", counter);

    // Step 5: Destroy the mutex
    pthread_mutex_destroy(&lock);

    return 0;
}

OUTPUT

Thread incremented counter to: 199986
Thread incremented counter to: 199987
Thread incremented counter to: 199988
Thread incremented counter to: 199989
Thread incremented counter to: 199990
Thread incremented counter to: 199991
Thread incremented counter to: 199992
Thread incremented counter to: 199993
Thread incremented counter to: 199994
Thread incremented counter to: 199995
Thread incremented counter to: 199996
Thread incremented counter to: 199997
Thread incremented counter to: 199998
Thread incremented counter to: 199999
Thread incremented counter to: 200000

Final counter value: 200000
c@c:~/cos3105$

Semaphore

Busy Waiting (Spinlock Semaphore)

#include <stdio.h>
#include <stdbool.h>
#include <pthread.h>
#include <unistd.h>

bool ready = false;  // Shared flag

void* producer(void* arg) {
    sleep(3);  // Simulate work for 3 seconds
    printf("Producer: Data is ready!\n");
    ready = true;  // Signal that data is ready
    return NULL;
}

void* consumer(void* arg) {
    printf("Consumer: Waiting for data...\n");

    // BUSY WAITING - keeps spinning!
    while (!ready) {
        // Do nothing, just keep checking
        // CPU is at 100% usage here!
    }

    printf("Consumer: Got the data!\n");
    return NULL;
}

int main() {
    pthread_t t1, t2;

    pthread_create(&t2, NULL, consumer, NULL);
    pthread_create(&t1, NULL, producer, NULL);

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);

    return 0;
}

OUTPUT

c@c:~/cos3105$ ./semaphore_busy_waiting
Consumer: Waiting for data...
Producer: Data is ready!
Consumer: Got the data!

No Busy Waiting (Blocking Semaphore)

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

// Process structure (represents blocked thread)
typedef struct process {
    pthread_cond_t cond;     // For blocking/waking this process
    struct process *next;    // Link to next in queue
} process;

// Semaphore structure
typedef struct {
    int value;
    process *waiting_queue;  // Queue of blocked processes
    pthread_mutex_t lock;    // Protects the semaphore
} semaphore;

// Global variable to track current process (thread-local in real OS)
__thread process *current_process = NULL;

// Add process to waiting queue
void add_to_queue(process **queue, process *p) {
    p->next = *queue;
    *queue = p;
    printf("    [Added to waiting queue]\n");
}

// Remove process from waiting queue
process* remove_from_queue(process **queue) {
    if (*queue == NULL) return NULL;

    process *p = *queue;
    *queue = p->next;
    printf("    [Removed from waiting queue]\n");
    return p;
}

// Initialize semaphore
void sem_init_blocking(semaphore *S, int initial_value) {
    S->value = initial_value;
    S->waiting_queue = NULL;
    pthread_mutex_init(&S->lock, NULL);
    printf("Semaphore initialized: value = %d\n\n", initial_value);
}

// Wait operation
void wait(semaphore *S) {
    pthread_mutex_lock(&S->lock);

    S->value--;
    printf("  wait(): value = %d\n", S->value);

    if (S->value < 0) {
        // Create current process structure
        current_process = (process*)malloc(sizeof(process));
        pthread_cond_init(&current_process->cond, NULL);

        // Add this process to waiting queue
        add_to_queue(&S->waiting_queue, current_process);

        printf("  wait(): BLOCKING (going to sleep)...\n");

        // block() - Put process to sleep
        pthread_cond_wait(&current_process->cond, &S->lock);

        printf("  wait(): WOKEN UP!\n");

        // Cleanup
        pthread_cond_destroy(&current_process->cond);
        free(current_process);
        current_process = NULL;
    }

    pthread_mutex_unlock(&S->lock);
}

// Signal operation
void signal(semaphore *S) {
    pthread_mutex_lock(&S->lock);

    S->value++;
    printf("  signal(): value = %d\n", S->value);

    if (S->value <= 0) {
        // Wake up a waiting process
        process *P = remove_from_queue(&S->waiting_queue);
        if (P != NULL) {
            printf("  signal(): Waking up a process\n");
            pthread_cond_signal(&P->cond);  // wakeup(P)
        }
    }

    pthread_mutex_unlock(&S->lock);
}

// Destroy semaphore
void sem_destroy_blocking(semaphore *S) {
    pthread_mutex_destroy(&S->lock);
}

// ========== Test Program ==========

semaphore S;

void* thread_function(void* arg) {
    int id = *(int*)arg;

    printf("\nThread %d: Calling wait()\n", id);
    wait(&S);

    printf("\nThread %d: In critical section (working...)\n", id);
    sleep(2);
    printf("Thread %d: Done working\n", id);

    printf("\nThread %d: Calling signal()\n", id);
    signal(&S);

    return NULL;
}

int main() {
    pthread_t threads[4];
    int ids[4] = {1, 2, 3, 4};

    // Initialize semaphore with value 1
    sem_init_blocking(&S, 1);

    // Create 4 threads
    for (int i = 0; i < 4; i++) {
        pthread_create(&threads[i], NULL, thread_function, &ids[i]);
        usleep(200000);  // Small delay for clear output
    }

    // Wait for all threads to finish
    for (int i = 0; i < 4; i++) {
        pthread_join(threads[i], NULL);
    }

    printf("\n========== All threads finished ==========\n");
    printf("Final semaphore value: %d\n", S.value);

    sem_destroy_blocking(&S);

    return 0;
}

OUTPUT

Semaphore initialized: value = 1


Thread 1: Calling wait()
  wait(): value = 0

Thread 1: In critical section (working...)

Thread 2: Calling wait()
  wait(): value = -1
    [Added to waiting queue]
  wait(): BLOCKING (going to sleep)...

Thread 3: Calling wait()
  wait(): value = -2
    [Added to waiting queue]
  wait(): BLOCKING (going to sleep)...

Thread 4: Calling wait()
  wait(): value = -3
    [Added to waiting queue]
  wait(): BLOCKING (going to sleep)...
Thread 1: Done working

Thread 1: Calling signal()
  signal(): value = -2
    [Removed from waiting queue]
  signal(): Waking up a process
  wait(): WOKEN UP!

Thread 4: In critical section (working...)
Thread 4: Done working

Thread 4: Calling signal()
  signal(): value = -1
    [Removed from waiting queue]
  signal(): Waking up a process
  wait(): WOKEN UP!

Thread 3: In critical section (working...)
Thread 3: Done working

Thread 3: Calling signal()
  signal(): value = 0
    [Removed from waiting queue]
  signal(): Waking up a process
  wait(): WOKEN UP!

Thread 2: In critical section (working...)
Thread 2: Done working

Thread 2: Calling signal()
  signal(): value = 1

========== All threads finished ==========
Final semaphore value: 1