Lecture 23: 同步：高级 (Synchronization: Advanced)

/* prethreaded.c —— 预线程化回显服务器
 * gcc -g -Wall -std=c11 -lpthread prethreaded.c -o prethreaded
 * ./prethreaded 12345 4 8     # 端口、工作线程数、并发客户端数
 */
#define _GNU_SOURCE
#include <arpa/inet.h>
#include <netinet/in.h>
#include <pthread.h>
#include <semaphore.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

#define MAXLINE  1024
#define SBUFSIZE 16

typedef struct {                    /* 有界缓冲区（CS:APP 图 12.26） */
    int   buf[SBUFSIZE];
    int   front, rear, n;
    sem_t mutex, slots, items;
} sbuf_t;

static sbuf_t sbuf;

static void sbuf_init(sbuf_t *sp)
{
    sp->front = sp->rear = sp->n = 0;
    sem_init(&sp->mutex, 0, 1);
    sem_init(&sp->slots, 0, SBUFSIZE);
    sem_init(&sp->items, 0, 0);
}
static void sbuf_insert(sbuf_t *sp, int item)     /* 生产者：主线程 */
{
    sem_wait(&sp->slots);            /* ① 等一个空槽位 */
    sem_wait(&sp->mutex);            /* ② 加锁写结构 */
    sp->buf[sp->rear] = item;
    sp->rear = (sp->rear + 1) % SBUFSIZE;
    sp->n++;
    sem_post(&sp->mutex);            /* ③ 解锁 */
    sem_post(&sp->items);            /* ④ 通告"多了一个条目" */
}
static int sbuf_remove(sbuf_t *sp)                /* 消费者：工作线程 */
{
    int item;
    sem_wait(&sp->items);            /* ① 等一个条目 */
    sem_wait(&sp->mutex);            /* ② 加锁读结构 */
    item = sp->buf[sp->front];
    sp->front = (sp->front + 1) % SBUFSIZE;
    sp->n--;
    sem_post(&sp->mutex);
    sem_post(&sp->slots);            /* ③ 通告"多了一个空槽位" */
    return item;
}

static pthread_mutex_t stat_lock = PTHREAD_MUTEX_INITIALIZER;
static int served = 0, total = 0;

static void process(int connfd)
{
    char buf[MAXLINE], out[MAXLINE + 8];
    ssize_t n = read(connfd, buf, sizeof(buf) - 1);
    if (n <= 0) return;
    buf[n] = '\0';
    snprintf(out, sizeof(out), "ECHO: %s", buf);
    write(connfd, out, strlen(out));

    int v = atoi(buf);
    pthread_mutex_lock(&stat_lock);         /* 第 1 类不安全函数的修复：加锁 */
    served++; total += v;
    printf("[worker] connfd=%-3d value=%-4d served=%d total=%d\n",
           connfd, v, served, total);
    fflush(stdout);
    pthread_mutex_unlock(&stat_lock);
}

static void *worker(void *vargp)
{
    pthread_detach(pthread_self());          /* ★ 每个工作线程自我分离 */
    (void)vargp;
    for (;;) {
        int connfd = sbuf_remove(&sbuf);     /* 阻塞等待一个连接 */
        process(connfd);
        close(connfd);
    }
    return NULL;
}

static int g_port, g_nclient;

static void *one_client(void *vargp)         /* 内置客户端，制造并发连接 */
{
    long id = (long)vargp;
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in a; memset(&a, 0, sizeof(a));
    a.sin_family = AF_INET;
    a.sin_port = htons((unsigned short)g_port);
    inet_pton(AF_INET, "127.0.0.1", &a.sin_addr);
    if (connect(fd, (struct sockaddr *)&a, sizeof(a)) < 0) { close(fd); return NULL; }

    char msg[32], buf[MAXLINE];
    int len = snprintf(msg, sizeof(msg), "%ld\n", id * 100);
    write(fd, msg, len);
    ssize_t n = read(fd, buf, sizeof(buf) - 1);
    if (n > 0) { buf[n] = '\0'; printf("[client %ld] %s", id, buf); fflush(stdout); }
    close(fd);
    return NULL;
}

static void *client_driver(void *vargp)
{
    (void)vargp;
    struct timespec ts = {0, 300 * 1000 * 1000};   /* 让主线程先进 accept */
    nanosleep(&ts, NULL);
    pthread_t ct[64];
    for (long i = 0; i < g_nclient; i++) pthread_create(&ct[i], NULL, one_client, (void *)(i + 1));
    for (int i = 0; i < g_nclient; i++) pthread_join(ct[i], NULL);
    return NULL;
}

int main(int argc, char **argv)
{
    int port    = (argc > 1) ? atoi(argv[1]) : 12345;
    int nthread = (argc > 2) ? atoi(argv[2]) : 4;
    int nclient = (argc > 3) ? atoi(argv[3]) : 8;
    signal(SIGPIPE, SIG_IGN);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    int one = 1;
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
    struct sockaddr_in addr; memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons((unsigned short)port);
    bind(listenfd, (struct sockaddr *)&addr, sizeof(addr));
    listen(listenfd, 1024);
    printf("listening on %d with %d worker threads\n", port, nthread); fflush(stdout);

    sbuf_init(&sbuf);

    /* ★★ 预线程化：主线程先建好一族工作线程，然后才开始 accept ★★ */
    pthread_t tid;
    for (int i = 0; i < nthread; i++)
        pthread_create(&tid, NULL, worker, NULL);

    g_port = port; g_nclient = nclient;
    pthread_t drv;
    pthread_create(&drv, NULL, client_driver, NULL);

    int accepted = 0;
    for (int i = 0; i < nclient; i++) {      /* 主线程 = 生产者 */
        struct sockaddr_in caddr; socklen_t clen = sizeof(caddr);
        int connfd = accept(listenfd, (struct sockaddr *)&caddr, &clen);
        if (connfd < 0) break;
        printf("[main] accepted connfd=%d\n", connfd); fflush(stdout);
        sbuf_insert(&sbuf, connfd);
        accepted++;
    }
    pthread_join(drv, NULL);
    sleep(1);
    printf("[main] accepted=%d served=%d total=%d\n", accepted, served, total);
    close(listenfd);
    return 0;
}

/* b_deadlock.c —— 相反加锁顺序导致死锁
 * gcc -g -Wall -std=c11 -lpthread b_deadlock.c -o b_deadlock
 * ./b_deadlock 0   # 不一致顺序：真死锁 -> timeout 5 退出码 124
 * ./b_deadlock 1   # 一致顺序：安全，退出码 0
 * ./b_deadlock 2   # 不一致顺序 + pthread_timedjoin_np 自检：退出码 3
 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

static pthread_mutex_t mA = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t mB = PTHREAD_MUTEX_INITIALIZER;
static atomic_int done_flag = 0, in_cs = 0;
static int g_mode = 0;

/* 让两个线程在拿到第一把锁之后精确对齐，使死锁可确定复现 */
static void rendezvous(void)
{
    struct timespec ts = {0, 30 * 1000 * 1000};   /* 30 ms */
    nanosleep(&ts, NULL);
}

static void *thread_1(void *arg)
{
    (void)arg;
    pthread_mutex_lock(&mA);
    rendezvous();
    printf("[T1] holds mA, now asks for mB ...\n"); fflush(stdout);
    pthread_mutex_lock(&mB);                      /* ← 若 T2 持有 mB 则卡死 */
    printf("[T1] acquired BOTH locks (in_cs=%d)\n", atomic_fetch_add(&in_cs, 1) + 1);
    fflush(stdout);
    atomic_fetch_sub(&in_cs, 1);
    pthread_mutex_unlock(&mB);
    pthread_mutex_unlock(&mA);
    atomic_fetch_add(&done_flag, 1);
    return NULL;
}

static void *thread_2(void *arg)
{
    (void)arg;
    if (g_mode == 0 || g_mode == 2) {
        pthread_mutex_lock(&mB);                  /* ★ 相反顺序：先 B 后 A */
        rendezvous();
        printf("[T2] holds mB, now asks for mA ...\n"); fflush(stdout);
        pthread_mutex_lock(&mA);                  /* ← 若 T1 持有 mA 则卡死 */
    } else {
        pthread_mutex_lock(&mA);                  /* ★ 一致顺序：先 A 后 B */
        rendezvous();
        printf("[T2] holds mA, now asks for mB ...\n"); fflush(stdout);
        pthread_mutex_lock(&mB);
    }
    printf("[T2] acquired BOTH locks (in_cs=%d)\n", atomic_fetch_add(&in_cs, 1) + 1);
    fflush(stdout);
    atomic_fetch_sub(&in_cs, 1);
    pthread_mutex_unlock(&mB);
    pthread_mutex_unlock(&mA);
    atomic_fetch_add(&done_flag, 1);
    return NULL;
}

static int timed_join(pthread_t t, int timeout_sec)
{
    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec += timeout_sec;
    return pthread_timedjoin_np(t, NULL, &deadline);
}

int main(int argc, char **argv)
{
    g_mode = (argc > 1) ? atoi(argv[1]) : 0;
    printf("mode=%d (%s)\n", g_mode,
           g_mode == 1 ? "consistent order: every thread locks A then B"
                       : "inconsistent order: T1 does A->B, T2 does B->A");
    fflush(stdout);

    pthread_t t1, t2;
    pthread_create(&t1, NULL, thread_1, NULL);
    pthread_create(&t2, NULL, thread_2, NULL);

    if (g_mode == 1) {
        pthread_join(t1, NULL);
        pthread_join(t2, NULL);
        printf("OK: no deadlock, done_flag=%d, in_cs=%d\n",
               atomic_load(&done_flag), atomic_load(&in_cs));
        return 0;
    }
    int rc1 = timed_join(t1, 3);
    int rc2 = rc1 ? ETIMEDOUT : timed_join(t2, 3);
    if (rc1 == ETIMEDOUT || rc2 == ETIMEDOUT) {
        printf("*** DEADLOCK: done_flag=%d, threads blocked waiting for each other "
               "(rc1=%d rc2=%d) ***\n", atomic_load(&done_flag), rc1, rc2);
        fflush(stdout);
        if (g_mode == 0) for (;;) pause();        /* 真挂起，留给 timeout/gdb */
        _exit(3);
    }
    printf("no deadlock: done_flag=%d\n", atomic_load(&done_flag));
    return 0;
}

/* c_condvar.c —— 条件变量版生产者-消费者
 * gcc -g -Wall -std=c11 -lpthread c_condvar.c -o c_condvar
 * ./c_condvar while   # 正确：while 包裹等待
 * ./c_condvar if      # ⚠️ 仅供演示：if 的 bug
 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define NITEMS 20
#define CAPACITY 4
#define NPROD 2
#define NCONS 2

static int use_while = 1;

/* ---- 定向实验：一个条目、两个等待者、broadcast ---- */
static pthread_mutex_t m2 = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  cv2 = PTHREAD_COND_INITIALIZER;
static int slot = 0;
static int if_wrong_wakeups = 0;

static void *waiter_if(void *arg)
{
    (void)arg;
    pthread_mutex_lock(&m2);
    if (slot == 0)                       /* ❌ if：只检查一次 */
        pthread_cond_wait(&cv2, &m2);
    if (slot == 0) {                     /* 醒来后条件早已不成立 */
        __sync_fetch_and_add(&if_wrong_wakeups, 1);
        printf("[if-waiter] woke up but slot still empty --> check was stale\n");
    } else {
        slot = 0;
        printf("[if-waiter] consumed the only item\n");
    }
    fflush(stdout);
    pthread_mutex_unlock(&m2);
    return NULL;
}

static void *waiter_while(void *arg)
{
    (void)arg;
    pthread_mutex_lock(&m2);
    while (slot == 0)                     /* ✅ while：醒来后重查 */
        pthread_cond_wait(&cv2, &m2);
    slot = 0;
    printf("[while-waiter] consumed the only item\n"); fflush(stdout);
    pthread_mutex_unlock(&m2);
    return NULL;
}

static void directed_experiment(int use_while_local)
{
    printf("\n=== directed experiment: ONE item, TWO waiters, broadcast (%s) ===\n",
           use_while_local ? "while" : "if");
    fflush(stdout);
    if_wrong_wakeups = 0;
    slot = 0;
    pthread_t a, b;
    pthread_create(&a, NULL, use_while_local ? waiter_while : waiter_if, NULL);
    pthread_create(&b, NULL, use_while_local ? waiter_while : waiter_if, NULL);
    struct timespec ts = {0, 200 * 1000 * 1000};
    nanosleep(&ts, NULL);                 /* 保证两个等待者都已挂起 */

    pthread_mutex_lock(&m2);
    slot = 1;                             /* 只放入一个条目 */
    pthread_cond_broadcast(&cv2);         /* 两个等待者都被唤醒 */
    pthread_mutex_unlock(&m2);

    if (use_while_local) {                /* while 版必须再放一个才能让第二人拿到 */
        nanosleep(&ts, NULL);
        pthread_mutex_lock(&m2);
        slot = 1;
        pthread_cond_broadcast(&cv2);
        pthread_mutex_unlock(&m2);
    }
    pthread_join(a, NULL);
    pthread_join(b, NULL);
    printf("--> stale-condition wakeups: %d\n", if_wrong_wakeups);
    fflush(stdout);
}

/* ---- 有界缓冲区本体 ---- */
typedef struct {
    int buf[CAPACITY];
    int count, in, out, produced, consumed, corrupt;
    pthread_mutex_t m;
    pthread_cond_t  not_empty, not_full;
} cb_t;

static cb_t cb;

static void *producer(void *arg)
{
    long id = (long)arg;
    for (int i = 0; i < NITEMS / NPROD; i++) {
        int item = (int)(id * 1000 + i);
        pthread_mutex_lock(&cb.m);
        if (use_while)
            while (cb.count == CAPACITY)                 /* ✅ */
                pthread_cond_wait(&cb.not_full, &cb.m);
        else
            if (cb.count == CAPACITY)                    /* ❌ */
                pthread_cond_wait(&cb.not_full, &cb.m);

        cb.buf[cb.in] = item;
        cb.in = (cb.in + 1) % CAPACITY;
        cb.count++; cb.produced++;
        printf("[P%ld] put %4d  count=%d\n", id, item, cb.count); fflush(stdout);
        pthread_cond_signal(&cb.not_empty);
        pthread_mutex_unlock(&cb.m);
    }
    return NULL;
}

static void *consumer(void *arg)
{
    long id = (long)arg;
    for (int i = 0; i < NITEMS / NCONS; i++) {
        pthread_mutex_lock(&cb.m);
        if (use_while) {
            while (cb.count == 0) {
                printf("[C%ld] buffer empty, waiting ...\n", id); fflush(stdout);
                pthread_cond_wait(&cb.not_empty, &cb.m);
            }
        } else {
            if (cb.count == 0) {
                printf("[C%ld] buffer empty, waiting ...\n", id); fflush(stdout);
                pthread_cond_wait(&cb.not_empty, &cb.m);
            }
            if (cb.count == 0) {            /* 检出 if 版的"陈旧条件"唤醒 */
                cb.corrupt++;
                printf("[C%ld] *** BUG: woke up but count==0 ***\n", id); fflush(stdout);
                pthread_mutex_unlock(&cb.m);
                continue;
            }
        }
        int item = cb.buf[cb.out];
        cb.out = (cb.out + 1) % CAPACITY;
        cb.count--; cb.consumed++;
        printf("[C%ld] got %4d  count=%d\n", id, item, cb.count); fflush(stdout);
        pthread_cond_signal(&cb.not_full);
        pthread_mutex_unlock(&cb.m);
    }
    return NULL;
}

int main(int argc, char **argv)
{
    use_while = !(argc > 1 && strcmp(argv[1], "if") == 0);
    printf("mode=%s\n", use_while ? "while" : "if(buggy)");

    memset(&cb, 0, sizeof(cb));
    cb.m = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
    cb.not_empty = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
    cb.not_full  = (pthread_cond_t)PTHREAD_COND_INITIALIZER;

    directed_experiment(use_while);

    printf("\n=== bounded buffer: %d producers, %d consumers, capacity %d ===\n",
           NPROD, NCONS, CAPACITY);
    fflush(stdout);
    pthread_t p[NPROD], c[NCONS];
    for (long i = 0; i < NPROD; i++) pthread_create(&p[i], NULL, producer, (void *)i);
    for (long i = 0; i < NCONS; i++) pthread_create(&c[i], NULL, consumer, (void *)i);
    for (int i = 0; i < NPROD; i++) pthread_join(p[i], NULL);
    for (int i = 0; i < NCONS; i++) pthread_join(c[i], NULL);

    printf("\nproduced=%d consumed=%d final count=%d corrupt=%d\n",
           cb.produced, cb.consumed, cb.count, cb.corrupt);
    return cb.corrupt ? 1 : 0;
}

/* d_ctime.c —— 第 3 类线程不安全函数：ctime 返回静态缓冲区
 * gcc -g -Wall -std=c11 -lpthread d_ctime.c -o d_ctime && ./d_ctime
 */
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#define NTHREADS 4

static pthread_barrier_t barrier;    /* 让"取结果"和"用结果"分离，放大竞态 */

/* ---------- ❌ 第 3 类：ctime 返回库内部静态缓冲区 ---------- */
static void *unsafe_thread(void *arg)
{
    long id = (long)arg;
    time_t t = (time_t)(1000000000L + id * 86400L * 365L);   /* 每人一个不同时间 */
    char *s = ctime(&t);                 /* ← 返回内部 static char buf[26] */
    pthread_barrier_wait(&barrier);      /* 所有线程在这里对齐 */
    printf("[unsafe %ld] my time should be 0x%lx, ctime says: %s",
           id, (unsigned long)t, s);     /* ← s 早已不是"自己"的字符串 */
    fflush(stdout);
    return NULL;
}

/* ---------- ✅ 修复：lock-and-copy ---------- */
static pthread_mutex_t ctime_lock = PTHREAD_MUTEX_INITIALIZER;

static char *correct_ctime(const time_t *t, char *dst, size_t dstsz)
{
    pthread_mutex_lock(&ctime_lock);
    char *r = ctime(t);
    if (r != NULL && dstsz > 0)
        snprintf(dst, dstsz, "%s", r);   /* ★ 拷贝发生在锁内！ */
    pthread_mutex_unlock(&ctime_lock);
    return (r == NULL) ? NULL : dst;
}

static void *safe_thread(void *arg)
{
    long id = (long)arg;
    time_t t = (time_t)(1000000000L + id * 86400L * 365L);
    char mybuf[64];                      /* ★ 每个线程自己的存储 */
    char *s = correct_ctime(&t, mybuf, sizeof(mybuf));
    pthread_barrier_wait(&barrier);
    printf("[safe   %ld] my time should be 0x%lx, ctime says: %s",
           id, (unsigned long)t, s);
    fflush(stdout);
    return NULL;
}

/* ---------- 讲义原样出现的 itoa（static char buf[11]） ---------- */
static char *itoa_unsafe(int x)
{
    static char buf[11];
    snprintf(buf, sizeof(buf), "%d", x);
    return buf;
}
static void itoa_safe(int x, char *buf, size_t bufsz)
{
    snprintf(buf, bufsz, "%d", x);
}

static pthread_barrier_t b2;
static void *itoa_thread(void *arg)
{
    long id = (long)arg;
    char *s = itoa_unsafe((int)(id * 11111111L));
    pthread_barrier_wait(&b2);
    printf("[itoa   %ld] expected %8ld, got: %s\n", id, id * 11111111L, s);
    fflush(stdout);
    return NULL;
}
static void *itoa_r_thread(void *arg)
{
    long id = (long)arg;
    char buf[12];
    itoa_safe((int)(id * 11111111L), buf, sizeof(buf));
    pthread_barrier_wait(&b2);
    printf("[itoa_r %ld] expected %8ld, got: %s\n", id, id * 11111111L, buf);
    fflush(stdout);
    return NULL;
}

int main(void)
{
    pthread_t t[NTHREADS];

    printf("######## 1) ctime: UNSAFE (class 3, static buffer) ########\n");
    pthread_barrier_init(&barrier, NULL, NTHREADS);
    for (long i = 0; i < NTHREADS; i++) pthread_create(&t[i], NULL, unsafe_thread, (void *)i);
    for (int i = 0; i < NTHREADS; i++) pthread_join(t[i], NULL);
    pthread_barrier_destroy(&barrier);

    printf("\n######## 2) ctime: FIXED with lock-and-copy ########\n");
    pthread_barrier_init(&barrier, NULL, NTHREADS);
    for (long i = 0; i < NTHREADS; i++) pthread_create(&t[i], NULL, safe_thread, (void *)i);
    for (int i = 0; i < NTHREADS; i++) pthread_join(t[i], NULL);
    pthread_barrier_destroy(&barrier);

    printf("\n######## 3) itoa (from lecture slide): UNSAFE vs _r ########\n");
    pthread_barrier_init(&b2, NULL, NTHREADS);
    for (long i = 0; i < NTHREADS; i++) pthread_create(&t[i], NULL, itoa_thread, (void *)i);
    for (int i = 0; i < NTHREADS; i++) pthread_join(t[i], NULL);
    for (long i = 0; i < NTHREADS; i++) pthread_create(&t[i], NULL, itoa_r_thread, (void *)i);
    for (int i = 0; i < NTHREADS; i++) pthread_join(t[i], NULL);
    pthread_barrier_destroy(&b2);
    return 0;
}

pthread_mutex_lock(&m);
if (head == NULL)                   /* 队列为空 */
    pthread_cond_wait(&cv, &m);
Node *n = head;                     /* 取队首 */
head = head->next;
pthread_mutex_unlock(&m);

pthread_mutex_lock(&m);
while (head == NULL)                 /* ✅ 循环重查 */
    pthread_cond_wait(&cv, &m);
Node *n = head;
head = head->next;
pthread_mutex_unlock(&m);