A First Look at Concurrency

// check_then_act.cpp: every access locked, still wrong
#include <cstdio>
#include <mutex>
#include <thread>

class LockedCount {
public:
    long get()       { std::lock_guard<std::mutex> g(m_);
                       return n_; }
    void set(long v) { std::lock_guard<std::mutex> g(m_);
                       n_ = v; }
private:
    std::mutex m_;
    long n_ = 0;
};

LockedCount count;

void add_many(long times) {
    for (long i = 0; i < times; ++i) {
        count.set(count.get() + 1);   // two locked steps
    }
}

int main(void) {
    std::thread a(add_many, 100000);
    std::thread b(add_many, 100000);
    a.join();
    b.join();
    std::printf("%ld\n", count.get());
    return 0;
}

// counter_atomic.cpp: each increment is indivisible
#include <atomic>
#include <cstdio>
#include <cstdlib>
#include <thread>

std::atomic<long> counter{0};  // each ++ is indivisible

void add_many(long times) {
    for (long i = 0; i < times; ++i) {
        ++counter;             // atomic read-modify-write
    }
}

int main(int argc, char* argv[]) {
    long times = (argc > 1) ? std::atol(argv[1]) : 1000;
    std::thread a(add_many, times);
    std::thread b(add_many, times);
    a.join();
    b.join();
    std::printf("%ld\n", counter.load());
    return 0;
}

// counter_mutex.cpp: the same counter, protected by a mutex
#include <cstdio>
#include <cstdlib>
#include <mutex>
#include <thread>

long counter = 0;          // shared; guarded by counter_mutex
std::mutex counter_mutex;      // protects counter, nothing else

void add_many(long times) {
    for (long i = 0; i < times; ++i) {
        std::lock_guard<std::mutex> guard(counter_mutex);
        ++counter;             // the critical section
    }                          // guard destroyed: unlock
}

int main(int argc, char* argv[]) {
    long times = (argc > 1) ? std::atol(argv[1]) : 1000;
    std::thread a(add_many, times);
    std::thread b(add_many, times);
    a.join();
    b.join();
    std::printf("%ld\n", counter);
    return 0;
}

// counter_race.cpp: two threads increment one shared counter
#include <cstdio>
#include <cstdlib>
#include <thread>

long counter = 0;              // shared, unprotected

void add_many(long times) {
    for (long i = 0; i < times; ++i) {
        ++counter;             // read, add one, write back
    }
}

int main(int argc, char* argv[]) {
    long times = (argc > 1) ? std::atol(argv[1]) : 1000;
    std::thread a(add_many, times);
    std::thread b(add_many, times);
    a.join();
    b.join();
    std::printf("%ld\n", counter);
    return 0;
}

// lock_cost.cpp: count readings above 25.0 C six ways and time each
#include <atomic>
#include <chrono>
#include <cstdio>
#include <mutex>
#include <thread>
#include <vector>

const std::size_t N = 10'000'000;
const double THRESHOLD = 25.0;
std::vector<double> temps(N);

long shared_count = 0;
std::mutex count_mutex;
std::atomic<long> atomic_count{0};

// Each worker scans temps[lo, hi).
void scan_plain(std::size_t lo, std::size_t hi, long* out) {
    long local = 0;
    for (std::size_t i = lo; i < hi; ++i) {
        if (temps[i] > THRESHOLD) ++local;
    }
    *out = local;                       // private result, no sharing
}
void scan_atomic(std::size_t lo, std::size_t hi) {
    for (std::size_t i = lo; i < hi; ++i) {
        if (temps[i] > THRESHOLD) ++atomic_count;  // atomic per hit
    }
}
void scan_mutex(std::size_t lo, std::size_t hi) {
    for (std::size_t i = lo; i < hi; ++i) {
        if (temps[i] > THRESHOLD) {
            std::lock_guard<std::mutex> g(count_mutex);
            ++shared_count;             // lock per hit
        }
    }
}
void scan_local_then_lock(std::size_t lo, std::size_t hi) {
    long local = 0;
    for (std::size_t i = lo; i < hi; ++i) {
        if (temps[i] > THRESHOLD) ++local;
    }
    std::lock_guard<std::mutex> g(count_mutex);
    shared_count += local;              // one lock per thread
}

template <typename F>
double best_ms(F f) {
    double best = 1e9;
    for (int rep = 0; rep < 11; ++rep) {
        shared_count = 0;
        atomic_count = 0;
        auto t0 = std::chrono::steady_clock::now();
        f();
        auto t1 = std::chrono::steady_clock::now();
        double ms = std::chrono::duration<double, std::milli>(t1 - t0).count();
        if (ms < best) best = ms;
    }
    return best;
}

int main(void) {
    for (std::size_t i = 0; i < N; ++i) {
        temps[i] = 20.0 + static_cast<double>((i * 7919) % 1000) / 100.0;
    }
    long r1 = 0;
    double a = best_ms([&] { scan_plain(0, N, &r1); });
    std::printf("1 thread, no lock:          %7.1f ms  count=%ld\n", a, r1);
    double b = best_ms([&] { scan_mutex(0, N); });
    std::printf("1 thread, mutex per hit:    %7.1f ms  count=%ld\n", b, shared_count);
    double b2 = best_ms([&] { scan_atomic(0, N); });
    std::printf("1 thread, atomic per hit:   %7.1f ms  count=%ld\n", b2, atomic_count.load());
    double c = best_ms([&] {
        std::thread t1(scan_mutex, 0, N / 2), t2(scan_mutex, N / 2, N);
        t1.join(); t2.join(); });
    std::printf("2 threads, mutex per hit:   %7.1f ms  count=%ld\n", c, shared_count);
    double d = best_ms([&] {
        std::thread t1(scan_atomic, 0, N / 2), t2(scan_atomic, N / 2, N);
        t1.join(); t2.join(); });
    std::printf("2 threads, atomic per hit:  %7.1f ms  count=%ld\n", d, atomic_count.load());
    double e = best_ms([&] {
        std::thread t1(scan_local_then_lock, 0, N / 2),
                    t2(scan_local_then_lock, N / 2, N);
        t1.join(); t2.join(); });
    std::printf("2 threads, local then lock: %7.1f ms  count=%ld\n", e, shared_count);
    return 0;
}

// lost_wakeup.cpp: the sensor finishes before the writer waits
#include <chrono>
#include <cstdio>
#include <thread>
#ifdef FIXED
#include "ring_buffer.h"
#else
#include "ring_buffer_lost.h"
#endif

int main(void) {
    stn::RingBuffer buf(64);
    std::thread sensor([&] {
        for (int t = 0; t < 3; ++t) buf.push({1001, t * 60, 21.0, 1});
        buf.close();                      // three readings, then done
    });
    std::thread writer([&] {
        std::this_thread::sleep_for(      // busy elsewhere for 100 ms
            std::chrono::milliseconds(100));
        stn::Reading r;
        while (buf.pop_wait(r)) {
            std::printf("wrote t=%lld\n", static_cast<long long>(r.time_s));
        }
        std::printf("writer done\n");
    });
    sensor.join();
    writer.join();
    return 0;
}

// no_join.cpp: a std::thread destroyed while still joinable
#include <cstdio>
#include <thread>

void hello() { std::printf("hello from the thread\n"); }

int main(void) {
    std::thread t(hello);
    std::printf("main returns without join\n");
    return 0;                       // t's destructor runs here
}

// overlap.cpp: how long do 1,000 increments take, compared with
// starting a thread? (Book build line; times are rough.)
#include <chrono>
#include <cstdio>
#include <thread>

long counter = 0;
using clk = std::chrono::steady_clock;

void add_many(long times) {
    for (long i = 0; i < times; ++i) ++counter;
}

double us_since(clk::time_point t0, clk::time_point t1) {
    return std::chrono::duration<double, std::micro>(t1 - t0).count();
}

int main(void) {
    for (long n : {1000L, 100000L}) {
        auto t0 = clk::now();
        add_many(n);
        auto t1 = clk::now();
        std::printf("n=%ld loop %.1f us\n", n, us_since(t0, t1));
    }
    for (int k = 0; k < 5; ++k) {
        clk::time_point started;
        auto t0 = clk::now();
        std::thread t([&] { started = clk::now(); });
        t.join();
        std::printf("start latency %.1f us\n", us_since(t0, started));
    }
    return 0;
}

#pragma once
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <vector>
namespace stn {
struct Reading { std::uint32_t sensor_id; std::int64_t time_s;
                 double value; std::uint8_t flags; };
class RingBuffer {
public:
    explicit RingBuffer(std::size_t cap) : data_(cap) {}
    bool push(const Reading& r) {           // sensor thread
        std::unique_lock<std::mutex> lock(m_);
        if (count_ == data_.size()) return false;   // full
        data_[(head_ + count_) % data_.size()] = r;
        ++count_;
        lock.unlock();                      // unlock, then wake
        not_empty_.notify_one();
        return true;
    }
    bool pop_wait(Reading& out) {           // writer thread
        std::unique_lock<std::mutex> lock(m_);
        not_empty_.wait(lock, [this] {  // re-checked on wake
            return count_ > 0 || closed_; });
        if (count_ == 0) return false;  // closed and drained
        out = data_[head_];
        head_ = (head_ + 1) % data_.size(); --count_;
        return true;
    }
    void close() { std::lock_guard<std::mutex> lock(m_);
                   closed_ = true; not_empty_.notify_all(); }
private:
    std::mutex m_;                          // guards all below
    std::condition_variable not_empty_;
    std::vector<Reading> data_;
    std::size_t head_ = 0, count_ = 0;
    bool closed_ = false;
};
}  // namespace stn

// ring_buffer_agent.h: the agent's "thread-safe" buffer
#pragma once
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <vector>
namespace stn {
struct Reading { std::uint32_t sensor_id; std::int64_t time_s;
                 double value; std::uint8_t flags; };
class RingBuffer {
public:
    explicit RingBuffer(std::size_t cap) : data_(cap) {}
    bool push(const Reading& r) {           // write: locked
        std::lock_guard<std::mutex> lock(m_);
        if (count_ == data_.size()) return false;   // full
        data_[(head_ + count_) % data_.size()] = r;
        ++count_;
        return true;
    }
    bool pop(Reading& out) {                // write: locked
        std::lock_guard<std::mutex> lock(m_);
        if (count_ == 0) return false;
        out = data_[head_];
        head_ = (head_ + 1) % data_.size(); --count_;
        return true;
    }
    std::size_t size() const { return count_; }  // no lock
    bool empty() const { return count_ == 0; }   // no lock
private:
    std::mutex m_;
    std::vector<Reading> data_;
    std::size_t head_ = 0;     // index of the oldest reading
    std::size_t count_ = 0;    // readings currently stored
};
}  // namespace stn

#pragma once
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <vector>
namespace stn {
struct Reading { std::uint32_t sensor_id; std::int64_t time_s;
                 double value; std::uint8_t flags; };
class RingBuffer {
public:
    explicit RingBuffer(std::size_t cap) : data_(cap) {}
    bool push(const Reading& r) {           // sensor thread
        std::unique_lock<std::mutex> lock(m_);
        if (count_ == data_.size()) return false;   // full
        data_[(head_ + count_) % data_.size()] = r;
        ++count_;
        lock.unlock();                      // unlock, then wake
        not_empty_.notify_one();
        return true;
    }
    bool pop_wait(Reading& out) {           // writer thread
        std::unique_lock<std::mutex> lock(m_);
        not_empty_.wait(lock);              // wait for a notify
        if (count_ == 0) return false;  // closed and drained
        out = data_[head_];
        head_ = (head_ + 1) % data_.size(); --count_;
        return true;
    }
    void close() { std::lock_guard<std::mutex> lock(m_);
                   closed_ = true; not_empty_.notify_all(); }
private:
    std::mutex m_;                          // guards all below
    std::condition_variable not_empty_;
    std::vector<Reading> data_;
    std::size_t head_ = 0, count_ = 0;
    bool closed_ = false;
};
}  // namespace stn

// slow_io.cpp: how long does the sensor wait for the lock?
#include <atomic>
#include <chrono>
#include <cstdio>
#include <mutex>
#include <thread>
#include <utility>
#include <vector>

using clk = std::chrono::steady_clock;
std::mutex m;                       // guards batch
std::vector<double> batch;
std::atomic<bool> done{false};

void slow_write(const std::vector<double>& v) {   // the SD card
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    (void)v;
}

void writer(bool hold_lock) {           // flushes every 20 ms
    while (!done) {
        std::vector<double> mine;
        {
            std::lock_guard<std::mutex> lock(m);
            if (hold_lock) slow_write(batch); // slow, locked
            std::swap(mine, batch);           // fast, locked
        }
        if (!hold_lock) slow_write(mine);     // slow, unlocked
        std::this_thread::sleep_for(
            std::chrono::milliseconds(20));
    }
}

double sensor_worst_wait_ms() {     // 300 readings, one per ms
    double worst = 0.0;
    for (int i = 0; i < 300; ++i) {
        auto t0 = clk::now();
        { std::lock_guard<std::mutex> lock(m); batch.push_back(21.0); }
        std::chrono::duration<double, std::milli> w = clk::now() - t0;
        if (w.count() > worst) worst = w.count();
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
    return worst;
}

int main(void) {
    for (int mode = 0; mode < 2; ++mode) {
        done = false;
        std::thread w(writer, mode == 0);
        double worst = sensor_worst_wait_ms();
        done = true;
        w.join();
        std::printf("%s worst push wait: %.3f ms\n",
                    mode == 0 ? "lock across write:" : "swap, then write: ",
                    worst);
    }
    return 0;
}

// small_task.cpp: is it worth starting a thread for 1,440 readings?
#include <chrono>
#include <cstdio>
#include <thread>
#include <vector>

long count_above(const std::vector<double>& v, std::size_t lo, std::size_t hi) {
    long n = 0;
    for (std::size_t i = lo; i < hi; ++i) {
        if (v[i] > 25.0) ++n;
    }
    return n;
}

int main(void) {
    std::vector<double> day(1440);              // one sensor, one day
    for (std::size_t i = 0; i < day.size(); ++i) {
        day[i] = 20.0 + static_cast<double>(i % 600) / 60.0;
    }
    using clk = std::chrono::steady_clock;
    long total1 = 0, total2 = 0;
    auto t0 = clk::now();
    for (int rep = 0; rep < 1000; ++rep) {
        total1 += count_above(day, 0, day.size());
    }
    auto t1 = clk::now();
    for (int rep = 0; rep < 1000; ++rep) {
        long a = 0, b = 0;
        std::thread ta([&] { a = count_above(day, 0, 720); });
        std::thread tb([&] { b = count_above(day, 720, 1440); });
        ta.join();
        tb.join();
        total2 += a + b;
    }
    auto t2 = clk::now();
    std::chrono::duration<double, std::micro> one = t1 - t0, two = t2 - t1;
    std::printf("1 thread:  %.2f us per day (total %ld)\n", one.count() / 1000, total1);
    std::printf("2 threads: %.2f us per day (total %ld)\n", two.count() / 1000, total2);
    return 0;
}

// spin_flag.cpp: the writer spins on a plain bool
#include <chrono>
#include <cstdio>
#include <thread>

double reading = 0.0;     // shared, no protection
bool ready = false;       // shared, no protection

int main(void) {
    std::thread sensor([] {
        std::this_thread::sleep_for(
            std::chrono::milliseconds(50));
        reading = 21.75;
        ready = true;                 // "the reading is there"
    });
    std::thread writer([] {
        while (!ready) {}                 // spin until set
        std::printf("writer got %.2f\n", reading);
    });
    sensor.join();
    writer.join();
    return 0;
}

// station.cpp: sensor thread and writer thread, fixed buffer
#include <cstdio>
#include <thread>
#include "ring_buffer.h"

const int DAY = 1440;                 // one reading per minute

int main(void) {
    stn::RingBuffer buf(64);
    std::FILE* log = std::fopen("day.csv", "w");
    if (log == nullptr) { std::perror("day.csv"); return 1; }
    std::thread sensor([&] {
        for (int t = 0; t < DAY; ++t) {
            stn::Reading r{1001, t * 60, 20.0 + t % 10, 1};
            while (!buf.push(r)) std::this_thread::yield();
        }
        buf.close();                      // no more readings
    });
    int written = 0;                      // writer thread only
    double sum = 0.0;                     // writer thread only
    std::thread writer([&] {
        stn::Reading r;
        while (buf.pop_wait(r)) {         // sleeps while empty
            std::fprintf(log, "%u,%lld,%.2f\n", r.sensor_id,
                static_cast<long long>(r.time_s), r.value);
            sum += r.value;
            ++written;
        }
    });
    sensor.join();
    writer.join();                    // now main may read
    std::fclose(log);
    std::printf("written=%d sum=%.1f\n", written, sum);
    return 0;
}

// station_agent.cpp: the same two threads, agent's buffer
#include <cstdio>
#include <thread>
#include "ring_buffer_agent.h"

const int DAY = 1440;                 // one reading per minute

int main(void) {
    stn::RingBuffer buf(64);
    double sum = 0.0;
    std::thread sensor([&] {
        for (int t = 0; t < DAY; ++t) {
            stn::Reading r{1001, t * 60, 20.0 + t % 10, 1};
            while (!buf.push(r)) std::this_thread::yield();
        }
    });
    std::thread writer([&] {
        int written = 0;
        while (written < DAY) {
            if (buf.empty()) continue;    // unlocked read
            stn::Reading r;
            if (buf.pop(r)) { sum += r.value; ++written; }
        }
    });
    sensor.join();
    writer.join();
    std::printf("sum=%.1f (expected %.1f)\n", sum, DAY * 24.5);
    return 0;
}

// two_stacks.cpp: each thread runs on its own stack
#include <chrono>
#include <cstdio>
#include <thread>

int shared_total = 0;           // one copy, seen by all

void count_to(const char* name, int n) {
    int local = 0;                 // on THIS thread's stack
    for (int i = 0; i < n; ++i) {
        ++local;
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    std::printf("%s: local=%d at %p, shared_total at %p\n",
                name, local, static_cast<void*>(&local),
                static_cast<void*>(&shared_total));
}

int main(void) {
    int in_main = 7;
    std::printf("main: in_main at %p\n",
                static_cast<void*>(&in_main));
    std::thread t1(count_to, "t1", 3);  // starts running now
    std::thread t2(count_to, "t2", 5);  // runs at the same time
    t1.join();                          // wait for t1 to finish
    t2.join();                          // wait for t2 to finish
    return 0;
}