Concurrency in LLD: Locks and Thread Safety

package main

import (
	"fmt"
	"sync"
)

// ---- Lock placement: the check ("is it free?") and the write ("mark it taken") must be
// inside the SAME critical section, or two threads can both see "free".

type Hall struct {
	mu    sync.Mutex
	seats map[int]string // seat -> who booked it
}

func (h *Hall) Book(seat int, who string) bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	if _, taken := h.seats[seat]; taken {
		return false
	}
	h.seats[seat] = who
	return true
}

func main() {
	// 1. Eight booking counters race for the same ten seats.
	hall := &Hall{seats: map[int]string{}}
	var wg sync.WaitGroup
	var countMu sync.Mutex
	booked, refused := 0, 0
	for w := 1; w <= 8; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for seat := 1; seat <= 10; seat++ {
				ok := hall.Book(seat, fmt.Sprintf("counter-%d", w))
				countMu.Lock()
				if ok {
					booked++
				} else {
					refused++
				}
				countMu.Unlock()
			}
		}()
	}
	wg.Wait()
	fmt.Printf("seats: %d booked, %d requests refused, %d seats with an owner\n", booked, refused, len(hall.seats))

	// 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
	jobs := make(chan int, 5) // bounded: the producer waits when 5 orders are queued
	results := make(chan int)
	var workers sync.WaitGroup
	for range 4 {
		workers.Add(1)
		go func() {
			defer workers.Done()
			for order := range jobs {
				results <- order * 10 // "price" each order
			}
		}()
	}
	go func() {
		for order := 1; order <= 20; order++ {
			jobs <- order
		}
		close(jobs) // no more work: workers finish their loops
	}()
	go func() { workers.Wait(); close(results) }()
	count, total := 0, 0
	for r := range results {
		count++
		total += r
	}
	fmt.Printf("orders: %d processed by 4 workers through a queue of 5, total %d tk\n", count, total)

	// 3. A shared counter: 4 threads x 10,000 increments, protected by a lock.
	var mu sync.Mutex
	counter := 0
	var cw sync.WaitGroup
	for range 4 {
		cw.Add(1)
		go func() {
			defer cw.Done()
			for range 10000 {
				mu.Lock()
				counter++
				mu.Unlock()
			}
		}()
	}
	cw.Wait()
	fmt.Printf("counter with a lock: %d (expected %d)\n", counter, 4*10000)
}

import queue
import threading


class Hall:
    """The check ("is it free?") and the write ("mark it taken") share one lock."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self.seats: dict[int, str] = {}

    def book(self, seat: int, who: str) -> bool:
        with self._lock:
            if seat in self.seats:
                return False
            self.seats[seat] = who
            return True


# 1. Eight booking counters race for the same ten seats.
hall = Hall()
count_lock = threading.Lock()
counts = {"booked": 0, "refused": 0}


def counter_desk(w: int) -> None:
    for seat in range(1, 11):
        ok = hall.book(seat, f"counter-{w}")
        with count_lock:
            counts["booked" if ok else "refused"] += 1


desks = [threading.Thread(target=counter_desk, args=(w,)) for w in range(1, 9)]
for t in desks:
    t.start()
for t in desks:
    t.join()
print(f"seats: {counts['booked']} booked, {counts['refused']} requests refused, {len(hall.seats)} seats with an owner")

# 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
jobs: "queue.Queue[int | None]" = queue.Queue(maxsize=5)  # put() waits when 5 are queued
results: "queue.Queue[int]" = queue.Queue()


def worker() -> None:
    while (order := jobs.get()) is not None:
        results.put(order * 10)  # "price" each order


pool = [threading.Thread(target=worker) for _ in range(4)]
for t in pool:
    t.start()
for order in range(1, 21):
    jobs.put(order)
for _ in pool:
    jobs.put(None)  # one "stop" per worker
for t in pool:
    t.join()
done = [results.get() for _ in range(results.qsize())]
print(f"orders: {len(done)} processed by 4 workers through a queue of 5, total {sum(done)} tk")

# 3. A shared counter: 4 threads x 10,000 increments, protected by a lock.
lock = threading.Lock()
counter = 0


def bump() -> None:
    global counter
    for _ in range(10000):
        with lock:
            counter += 1


threads = [threading.Thread(target=bump) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"counter with a lock: {counter} (expected {4 * 10000})")

// JavaScript runs this code on one thread. Concurrency comes from async tasks that take
// turns at every `await`, so a race can only happen across an await, never inside sync code.

class Hall {
  readonly seats = new Map<number, string>();
  // No await between the check and the write, so no other task can run in between.
  book(seat: number, who: string): boolean {
    if (this.seats.has(seat)) return false;
    this.seats.set(seat, who);
    return true;
  }
}

// A bounded async queue: put() waits while it's full, take() waits while it's empty.
class BoundedQueue<T> {
  readonly #items: T[] = [];
  readonly #capacity: number;
  #waitingTakers: ((v: T) => void)[] = [];
  #waitingPutters: (() => void)[] = [];
  constructor(capacity: number) {
    this.#capacity = capacity;
  }
  async put(v: T): Promise<void> {
    while (this.#items.length >= this.#capacity) await new Promise<void>((r) => this.#waitingPutters.push(r));
    const taker = this.#waitingTakers.shift();
    if (taker) taker(v);
    else this.#items.push(v);
  }
  async take(): Promise<T> {
    const v = this.#items.shift();
    if (v !== undefined) {
      this.#waitingPutters.shift()?.();
      return v;
    }
    return new Promise<T>((r) => this.#waitingTakers.push(r));
  }
}

const tick = () => new Promise<void>((r) => setTimeout(r, 0));

async function main(): Promise<void> {
  // 1. Eight booking counters race for the same ten seats.
  const hall = new Hall();
  let booked = 0;
  let refused = 0;
  await Promise.all(
    Array.from({ length: 8 }, async (_, i) => {
      for (let seat = 1; seat <= 10; seat++) {
        await tick(); // let other counters run between requests
        if (hall.book(seat, `counter-${i + 1}`)) booked++;
        else refused++;
      }
    }),
  );
  console.log(`seats: ${booked} booked, ${refused} requests refused, ${hall.seats.size} seats with an owner`);

  // 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
  const jobs = new BoundedQueue<number | null>(5);
  const results: number[] = [];
  const workers = Array.from({ length: 4 }, async () => {
    for (let order = await jobs.take(); order !== null; order = await jobs.take()) {
      results.push(order * 10); // "price" each order
    }
  });
  for (let order = 1; order <= 20; order++) await jobs.put(order);
  for (let i = 0; i < 4; i++) await jobs.put(null); // one "stop" per worker
  await Promise.all(workers);
  const total = results.reduce((a, b) => a + b, 0);
  console.log(`orders: ${results.length} processed by 4 workers through a queue of 5, total ${total} tk`);

  // 3. A shared counter: 4 tasks x 10,000 increments. counter++ has no await inside it, so the
  // single thread itself acts as the lock (the label below matches the other languages).
  let counter = 0;
  await Promise.all(
    Array.from({ length: 4 }, async () => {
      for (let i = 0; i < 10000; i++) {
        counter++;
        if (i % 1000 === 0) await tick();
      }
    }),
  );
  console.log(`counter with a lock: ${counter} (expected ${4 * 10000})`);
}

void main();

use std::collections::HashMap;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

// The check ("is it free?") and the write ("mark it taken") happen under one lock.
// In Rust the map lives INSIDE the Mutex, so it can't be touched without locking.
struct Hall { seats: Mutex<HashMap<u32, String>> }

impl Hall {
    fn book(&self, seat: u32, who: &str) -> bool {
        let mut seats = self.seats.lock().unwrap();
        if seats.contains_key(&seat) { return false; }
        seats.insert(seat, who.to_string());
        true
    }
}

fn main() {
    // 1. Eight booking counters race for the same ten seats.
    let hall = Arc::new(Hall { seats: Mutex::new(HashMap::new()) });
    let counts = Arc::new(Mutex::new((0u32, 0u32))); // (booked, refused)
    let desks: Vec<_> = (1..=8).map(|w| {
        let (hall, counts) = (Arc::clone(&hall), Arc::clone(&counts));
        thread::spawn(move || {
            for seat in 1..=10 {
                let ok = hall.book(seat, &format!("counter-{w}"));
                let mut c = counts.lock().unwrap();
                if ok { c.0 += 1 } else { c.1 += 1 }
            }
        })
    }).collect();
    for d in desks { d.join().unwrap(); }
    let (booked, refused) = *counts.lock().unwrap();
    println!("seats: {booked} booked, {refused} requests refused, {} seats with an owner", hall.seats.lock().unwrap().len());

    // 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
    let (job_tx, job_rx) = mpsc::sync_channel::<u32>(5); // send() blocks when 5 are queued
    let job_rx = Arc::new(Mutex::new(job_rx));               // workers share one receiver
    let (res_tx, res_rx) = mpsc::channel::<u32>();
    let pool: Vec<_> = (0..4).map(|_| {
        let (rx, tx) = (Arc::clone(&job_rx), res_tx.clone());
        thread::spawn(move || loop {
            let next = rx.lock().unwrap().recv();
            match next {
                Ok(order) => tx.send(order * 10).unwrap(), // "price" each order
                Err(_) => break,                           // channel closed: no more work
            }
        })
    }).collect();
    drop(res_tx);
    for order in 1..=20 { job_tx.send(order).unwrap(); }
    drop(job_tx); // close the queue
    for p in pool { p.join().unwrap(); }
    let done: Vec<u32> = res_rx.iter().collect();
    println!("orders: {} processed by 4 workers through a queue of 5, total {} tk", done.len(), done.iter().sum::<u32>());

    // 3. A shared counter: 4 threads x 10,000 increments, protected by a lock.
    let counter = Arc::new(Mutex::new(0u32));
    let threads: Vec<_> = (0..4).map(|_| {
        let counter = Arc::clone(&counter);
        thread::spawn(move || for _ in 0..10000 { *counter.lock().unwrap() += 1; })
    }).collect();
    for t in threads { t.join().unwrap(); }
    println!("counter with a lock: {} (expected {})", *counter.lock().unwrap(), 4 * 10000);
}

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Main {
    // The check ("is it free?") and the write ("mark it taken") share one lock.
    static final class Hall {
        private final Map<Integer, String> seats = new HashMap<>();
        synchronized boolean book(int seat, String who) {
            if (seats.containsKey(seat)) return false;
            seats.put(seat, who);
            return true;
        }
        synchronized int owned() { return seats.size(); }
    }

    private static int counter = 0;
    private static final Object LOCK = new Object();

    public static void main(String[] args) throws InterruptedException {
        // 1. Eight booking counters race for the same ten seats.
        Hall hall = new Hall();
        AtomicInteger booked = new AtomicInteger(), refused = new AtomicInteger();
        ExecutorService desks = Executors.newFixedThreadPool(8);
        for (int w = 1; w <= 8; w++) {
            final int id = w;
            desks.submit(() -> {
                for (int seat = 1; seat <= 10; seat++) {
                    if (hall.book(seat, "counter-" + id)) booked.incrementAndGet(); else refused.incrementAndGet();
                }
            });
        }
        desks.shutdown();
        desks.awaitTermination(10, TimeUnit.SECONDS);
        System.out.printf("seats: %d booked, %d requests refused, %d seats with an owner%n", booked.get(), refused.get(), hall.owned());

        // 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
        BlockingQueue<Integer> jobs = new ArrayBlockingQueue<>(5); // put() waits when 5 are queued
        AtomicInteger count = new AtomicInteger(), total = new AtomicInteger();
        final int stop = -1;                                       // one "stop" per worker
        ExecutorService pool = Executors.newFixedThreadPool(4);
        for (int i = 0; i < 4; i++) {
            pool.submit(() -> {
                try {
                    for (int order = jobs.take(); order != stop; order = jobs.take()) {
                        count.incrementAndGet();
                        total.addAndGet(order * 10); // "price" each order
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        for (int order = 1; order <= 20; order++) jobs.put(order);
        for (int i = 0; i < 4; i++) jobs.put(stop);
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.printf("orders: %d processed by 4 workers through a queue of 5, total %d tk%n", count.get(), total.get());

        // 3. A shared counter: 4 threads x 10,000 increments, protected by a lock.
        Thread[] threads = new Thread[4];
        for (int i = 0; i < 4; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 10000; j++) synchronized (LOCK) { counter++; }
            });
            threads[i].start();
        }
        for (Thread t : threads) t.join();
        System.out.printf("counter with a lock: %d (expected %d)%n", counter, 4 * 10000);
    }
}

#include <condition_variable>
#include <cstdio>
#include <map>
#include <mutex>
#include <optional>
#include <queue>
#include <string>
#include <thread>
#include <vector>

// The check ("is it free?") and the write ("mark it taken") share one lock.
class Hall {
public:
    bool book(int seat, const std::string& who) {
        std::lock_guard lock(mu_);
        if (seats_.count(seat)) return false;
        seats_[seat] = who;
        return true;
    }
    size_t owned() {
        std::lock_guard lock(mu_);
        return seats_.size();
    }

private:
    std::mutex mu_;
    std::map<int, std::string> seats_;
};

// A bounded queue: put() waits while full, take() waits while empty.
template <class T>
class BoundedQueue {
public:
    explicit BoundedQueue(size_t cap) : cap_(cap) {}
    void put(T v) {
        std::unique_lock lock(mu_);
        notFull_.wait(lock, [&] { return q_.size() < cap_; });
        q_.push(std::move(v));
        notEmpty_.notify_one();
    }
    T take() {
        std::unique_lock lock(mu_);
        notEmpty_.wait(lock, [&] { return !q_.empty(); });
        T v = std::move(q_.front());
        q_.pop();
        notFull_.notify_one();
        return v;
    }

private:
    size_t cap_;
    std::mutex mu_;
    std::condition_variable notFull_, notEmpty_;
    std::queue<T> q_;
};

int main() {
    // 1. Eight booking counters race for the same ten seats.
    Hall hall;
    std::mutex countMu;
    int booked = 0, refused = 0;
    std::vector<std::thread> desks;
    for (int w = 1; w <= 8; w++) {
        desks.emplace_back([&, w] {
            for (int seat = 1; seat <= 10; seat++) {
                bool ok = hall.book(seat, "counter-" + std::to_string(w));
                std::lock_guard lock(countMu);
                (ok ? booked : refused)++;
            }
        });
    }
    for (auto& d : desks) d.join();
    std::printf("seats: %d booked, %d requests refused, %zu seats with an owner\n", booked, refused, hall.owned());

    // 2. Producer-consumer: one producer, a bounded queue of 5, a pool of 4 workers.
    BoundedQueue<std::optional<int>> jobs(5);
    std::mutex resMu;
    int count = 0, total = 0;
    std::vector<std::thread> pool;
    for (int i = 0; i < 4; i++) {
        pool.emplace_back([&] {
            while (auto order = jobs.take()) {  // std::nullopt means "stop"
                std::lock_guard lock(resMu);
                count++;
                total += *order * 10;  // "price" each order
            }
        });
    }
    for (int order = 1; order <= 20; order++) jobs.put(order);
    for (int i = 0; i < 4; i++) jobs.put(std::nullopt);  // one "stop" per worker
    for (auto& p : pool) p.join();
    std::printf("orders: %d processed by 4 workers through a queue of 5, total %d tk\n", count, total);

    // 3. A shared counter: 4 threads x 10,000 increments, protected by a lock.
    std::mutex mu;
    int counter = 0;
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; i++)
        threads.emplace_back([&] {
            for (int j = 0; j < 10000; j++) {
                std::lock_guard lock(mu);
                counter++;
            }
        });
    for (auto& t : threads) t.join();
    std::printf("counter with a lock: %d (expected %d)\n", counter, 4 * 10000);
}