POSIX thread হাতে-কলমে — তৈরি, join আর thread pool

১ · Pthreads-এর মানচিত্র — কোন ফাংশন কী করে

পর্ব ১৪-এ দেখেছেন থ্রেড কী: একটি process-এর ভেতরে একাধিক নির্বাহ-ধারা, যারা একই মেমরি ভাগ করে কিন্তু প্রত্যেকের নিজের stack আর register থাকে। এবার হাতে-কলমে কাজ। Linux, macOS, FreeBSD-এর মতো Unix-ধাঁচের সিস্টেমে থ্রেড তৈরির প্রচলিত উপায় হলো POSIX threads, সংক্ষেপে pthreads. এটি POSIX মানদণ্ডের অংশ — IEEE Std 1003.1c, ১৯৯৫ সালে প্রকাশিত। ফলে একই C কোড এসব OS-এ প্রায় অপরিবর্তিত চলে।

পুরো API-তে শতাধিক ফাংশন থাকলেও রোজকার কাজের জন্য অল্প কয়েকটিই যথেষ্ট:

শুরুর আগে তিনটি ব্যবহারিক কথা মনে রাখুন:

• Header আর compile: #include <pthread.h>, আর compile করার সময় gcc -pthread prog.c. glibc 2.34 (২০২১) থেকে pthread মূল C library-তেই, তাই আধুনিক Linux-এ flag ছাড়াও প্রায়ই link হয়; তবু -pthread দেওয়াই নিয়ম, কারণ পুরোনো সিস্টেম বা অন্য OS-এ এটি ছাড়া চলবে না।
• ত্রুটি আসে ফেরত মানে, errno-তে নয়: pthread ফাংশন সফল হলে 0 দেয়, ব্যর্থ হলে সরাসরি একটি error নম্বর ফেরত দেয় (যেমন EAGAIN — আর থ্রেড বানানোর সম্পদ নেই)। তাই perror নয়, strerror(rc) ব্যবহার করুন। পর্ব ৫-এর system call-এর নিয়ম থেকে এটি আলাদা।
• OS-ভেদে পার্থক্য: Linux আর macOS-এ pthreads আসল API. Windows-এর নিজস্ব API হলো CreateThread/_beginthreadex আর WaitForSingleObject (অংশ ৭); সেখানে pthreads পেতে MinGW-w64-এর winpthreads-এর মতো আলাদা library লাগে।

২ · pthread_create — function pointer আর void* argument

ফাংশনটির ঘোষণা প্রথম দেখায় ভয় লাগতে পারে:

int pthread_create(pthread_t *thread,              /* out: new thread's id   */
                   const pthread_attr_t *attr,     /* NULL = default settings */
                   void *(*start)(void *),         /* function the thread runs */
                   void *arg);                     /* the one argument it gets */

টুকরো করে পড়লে সহজ। প্রথমটিতে নতুন থ্রেডের পরিচয় লেখা হবে। দ্বিতীয়টি attribute — NULL দিলে সব ডিফল্ট। তৃতীয়টি একটি function pointer: এমন একটি ফাংশনের ঠিকানা, যা একটি void* নেয় আর একটি void* ফেরত দেয়। নতুন থ্রেড এই ফাংশন থেকেই জীবন শুরু করে, আর ফাংশনটি ফিরলেই থ্রেড শেষ। চতুর্থটি সেই ফাংশনকে দেওয়ার একমাত্র argument.

কেন void*? C-তে template নেই, আর library জানে না আপনার থ্রেডের কী লাগবে — একটি সংখ্যা, না দশটি জিনিসের একটি struct. void* হলো “যেকোনো কিছুর ঠিকানা”: আপনি ঠিকানা পাঠান, থ্রেডের ফাংশন সেটিকে আসল ধরনে ফিরিয়ে নেয়। বেশি তথ্য হলে সব একটি struct-এ ভরে তার ঠিকানা দিন। (Function pointer নিয়ে আরও অনুশীলন চাইলে দেখুন C কোর্সের function pointer পর্ব।)

নিচের প্রোগ্রামে তিনটি থ্রেড, প্রত্যেকের নিজের একটি struct job. থ্রেড নিজের struct পড়ে কাজ করে, ফলও সেখানেই লেখে, আর struct-এর ঠিকানাই ফেরত দেয়। সঙ্গে দেখা হবে, সব থ্রেডের getpid() এক, কিন্তু kernel-এর দেওয়া thread id (gettid(), Linux-এ glibc 2.30 থেকে) আলাদা:

#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

struct job { int id, n; long sum; pid_t pid, tid; };   /* one per thread */

static void *worker(void *arg) {
    struct job *j = arg;                  /* void* -> our own struct */
    for (int k = 1; k <= j->n; k++) j->sum += k;
    j->pid = getpid();
    j->tid = gettid();                    /* kernel thread id (Linux) */
    return j;                             /* comes back via pthread_join */
}

int main(void) {
    pthread_t t[3];
    struct job jobs[3];
    for (int i = 0; i < 3; i++) {
        jobs[i] = (struct job){ .id = i, .n = (i + 1) * 100 };
        int rc = pthread_create(&t[i], NULL, worker, &jobs[i]);
        if (rc != 0) { printf("create: %s\n", strerror(rc)); return 1; }
    }
    for (int i = 0; i < 3; i++) {
        void *ret;
        pthread_join(t[i], &ret);         /* wait, and collect the result */
        struct job *j = ret;
        printf("job %d: sum 1..%d = %ld | same pid as main: %s | tid != pid: %s\n",
               j->id, j->n, j->sum, j->pid == getpid() ? "yes" : "no",
               j->tid != j->pid ? "yes" : "no");
    }
    printf("main: tid == pid? %s\n", gettid() == getpid() ? "yes" : "no");
    return 0;
}

job 0: sum 1..100 = 5050 | same pid as main: yes | tid != pid: yes
job 1: sum 1..200 = 20100 | same pid as main: yes | tid != pid: yes
job 2: sum 1..300 = 45150 | same pid as main: yes | tid != pid: yes
main: tid == pid? yes

সবচেয়ে পরিচিত বাগ — লুপের &i পাঠানো

নতুনরা (আর অনেক সময় AI-ও) প্রথমে এভাবে লেখেন — আলাদা struct-এর ঝামেলা না করে সরাসরি লুপের ভ্যারিয়েবলের ঠিকানা:

static void *worker(void *p) {
    int id = *(int *)p;          /* reads main's i -- whenever it gets the CPU */
    printf("thread %d\n", id);
    return NULL;
}
/* in main: */
int i;
for (i = 0; i < 3; i++)
    pthread_create(&t[i], NULL, worker, &i);   /* BUG: same &i for all */
for (int j = 0; j < 3; j++)
    pthread_join(t[j], NULL);

আশা ছিল thread 0, thread 1, thread 2. এই কোডটি একই অনলাইন রানারে তিনবার চালিয়ে আসলে পাওয়া গেছে:

run 1: thread 1, thread 3, thread 3
run 2: thread 1, thread 3, thread 3
run 3: thread 3, thread 3, thread 3

একবারও 0 বা 2 আসেনি, অথচ 3 এসেছে — যে মানে কোনো থ্রেডই তৈরি হয়নি! কারণ pthread_create থ্রেডকে শুধু “চলার জন্য প্রস্তুত” করে ফেরে; কখন CPU পাবে, ঠিক করে scheduler. ততক্ষণে main-এর লুপ এগোয়, আর সবার হাতে একই &i; থ্রেড *p পড়ার মুহূর্তে i-তে যা আছে, তা-ই পায়। নিচের ডায়াগ্রামে run 1-এর মতো একটি সময়রেখা ধাপে ধাপে দেখুন:

1void *worker(void *p) {2  int id = *(int *)p;  // read later!3  printf("thread %d\n", id); }4int i;5for (i = 0; i < 3; i++)6  pthread_create(&t[i], NULL, worker, &i);7for (int j = 0; j < 3; j++) pthread_join(t[j], NULL);8// fix: ids[i] = i;  pass &ids[i]

৩ · Join, ফেরত মান, pthread_exit আর detach

pthread_join — অপেক্ষা আর ফল নেওয়া

pthread_join(t, &ret) ডাকলে ডাকা থ্রেডটি ঘুমিয়ে পড়ে, যতক্ষণ না থ্রেড t শেষ হয়। তারপর t-এর ফেরত দেওয়া void* এসে বসে ret-এ। এটি পর্ব ১০-এর waitpid-এর মতো, তবে যেকোনো থ্রেড যেকোনো joinable থ্রেডকে join করতে পারে — শুধু “প্যারেন্ট” নয়।

একটি বিপজ্জনক ভুল: থ্রেডের ফাংশনের local ভ্যারিয়েবলের ঠিকানা ফেরত দেওয়া। থ্রেড শেষ হলেই তার stack অবৈধ। নিরাপদ উপায়: argument-এর struct-এ ফল লেখা (উপরের প্রোগ্রামের মতো), malloc করা মেমরির ঠিকানা ফেরত দেওয়া (join-কারী পরে free করবে), অথবা ছোট পূর্ণসংখ্যা হলে (void *)(intptr_t)value.

pthread_exit আর main-এর return

থ্রেড শেষ হয় তিনভাবে: start ফাংশন থেকে return, যেকোনো গভীর ফাংশন থেকে pthread_exit(value), অথবা পুরো প্রসেস শেষ হলে। শেষেরটি নিয়ে সাবধান: main থেকে return মানে exit(), আর তাতে প্রসেসের সব থ্রেড সঙ্গে সঙ্গে শেষ — কাজ শেষ হোক বা না হোক। তাই main অন্যদের join করে তারপর ফেরে। আর main নিজে pthread_exit ডাকলে শুধু main শেষ হয়; বাকিরা চলতে থাকে, শেষজন শেষ হলে প্রসেস শেষ।

Joinable বনাম detached

ডিফল্টভাবে প্রতিটি থ্রেড joinable. এমন থ্রেড শেষ হলেও তার কিছু তথ্য (ফেরত মান, পরিচয়, stack-এর মেমরি) রয়ে যায়, যতক্ষণ না কেউ join করে — অনেকটা পর্ব ১০-এর zombie প্রসেসের মতো। কেউ কখনো join না করলে সেই সম্পদ জমতেই থাকে। যে থ্রেডের ফল কারও লাগবে না, তাকে detached করুন: pthread_detach(t) ডাকলে বা attribute-এ PTHREAD_CREATE_DETACHED দিলে থ্রেড শেষ হওয়ার সঙ্গে সঙ্গে নিজের সম্পদ ফিরিয়ে দেয়।

Attribute — stack size আর অন্যান্য

প্রতিটি থ্রেডের নিজের stack লাগে, আর সেটি আসে প্রসেসের address space থেকে। কত বড় হবে, তা ঠিক হয় thread attribute দিয়ে। নিচের কোডটি ডিফল্ট stack size জানে, ২৫৬ KiB stack-এর একটি detached থ্রেড বানায়, আর pthread_exit দিয়ে একটি মান ফেরত পাঠায়। (এই পাতায় এটি চালানো যাবে না; ডান পাশের মন্তব্যে Wandbox-এর একটি পরীক্ষা-রানের মান দেওয়া হলো।)

static void *bg(void *arg) { (void)arg; return NULL; }
static void *ex(void *arg) { (void)arg; pthread_exit((void *)42); }
/* inside main: */
pthread_attr_t attr;
size_t sz;
pthread_attr_init(&attr);
pthread_attr_getstacksize(&attr, &sz);            /* test run: 8192 KiB      */
pthread_attr_setstacksize(&attr, 256 * 1024);      /* this thread: 256 KiB    */
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_create(&t, &attr, bg, NULL);              /* rc = 0                  */
pthread_attr_destroy(&attr);                      /* attr no longer needed   */
/* pthread_join(t, NULL) here is a bug: glibc returned EINVAL in our test */

pthread_create(&t2, NULL, ex, NULL);
pthread_join(t2, &ret);                           /* (long)ret == 42         */
sysconf(_SC_THREAD_STACK_MIN);                     /* test run: 16384 bytes   */

Linux-এ glibc ডিফল্ট stack size নেয় ulimit -s থেকে — সাধারণত ৮ MiB, পরীক্ষাতেও তা-ই। এটি ভার্চুয়াল মেমরির হিসাব; যতটুকু সত্যিই ছোঁয়া হয়, ততটুকুই RAM নেয় (পর্ব ৩৫)। অন্য সিস্টেমে ডিফল্ট আলাদা: Windows-এ সাধারণত ১ MB, macOS-এ main ছাড়া অন্য থ্রেডে ৫১২ KiB, আর musl libc (Alpine Linux)-এ অনেক ছোট, সংস্করণভেদে প্রায় ১২৮ KiB. তাই বড় local অ্যারে বা গভীর recursion-ওয়ালা কোড glibc-তে ঠিক চললেও Alpine container-এ stack overflow-তে ভাঙতে পারে। প্রতিটি থ্রেডের stack-এর শেষে glibc একটি guard page রাখে, যাতে overflow হলে নীরবে অন্যের মেমরি নষ্ট না করে সঙ্গে সঙ্গে segmentation fault হয়।

থ্রেডের পরিচয় — pthread_t, TID আর ps -L

থ্রেডের দুই রকম পরিচয় আছে। pthread_self() দেয় একটি pthread_t — library-র নিজস্ব হাতল, যার ভেতরের রূপ OS-ভেদে আলাদা (glibc-তে আসলে একটি ঠিকানা)। তাই দুটি pthread_t == দিয়ে নয়, pthread_equal() দিয়ে তুলনা করতে হয়। অন্যদিকে Linux kernel প্রতিটি থ্রেডকে একটি সংখ্যা দেয় — TID, যা gettid() দেয় আর ps, top-এ দেখা যায়। Linux-এর ভাষায় প্রতিটি থ্রেড একটি “task”; প্রসেসের সব থ্রেড মিলে একটি thread group, আর সেই group-এর id-ই PID. থ্রেডের একটি নামও দেওয়া যায় (pthread_setname_np, Linux-এ ১৫ অক্ষর পর্যন্ত), যা debugging-এ খুব কাজে লাগে।

$ ps -L -o pid,lwp,nlwp,comm -p 4321
    PID     LWP NLWP COMMAND
   4321    4321    4 server
   4321    4322    4 server
   4321    4323    4 server
   4321    4324    4 server
$ ls /proc/4321/task
4321  4322  4323  4324

LWP (light-weight process) কলামটিই TID, আর NLWP মোট থ্রেড সংখ্যা। চারটি সারির PID একই — একটি প্রসেস, চারটি থ্রেড; প্রথম সারিতে PID আর TID সমান, এটিই main থ্রেড। Windows-এ একই তথ্য দেখায় Task Manager-এর Details ট্যাবের “Threads” কলাম বা Process Explorer.

৪ · কাজ ভাগ করা — একটি অ্যারের যোগফল কয়েকটি থ্রেডে

থ্রেডের সবচেয়ে সরল ব্যবহার: একটি বড় কাজকে কয়েকটি টুকরোয় ভাগ করে আলাদা core-এ একসাথে চালানো, অর্থাৎ parallelism. উদাহরণ — ১০ লাখ সংখ্যার যোগফল। T সংখ্যক থ্রেড থাকলে থ্রেড id নেবে সূচক id·N/T থেকে (id+1)·N/T-এর ঠিক আগ পর্যন্ত। এই সূত্রে টুকরোগুলো একে অপরের ওপর পড়ে না, কোনো সূচক বাদ যায় না, আর N যদি T দিয়ে ভাগ না যায়, তবু শেষ টুকরো ঠিক N-এ থামে।

এই নকশার মূল কৌশল একটি: প্রতিটি থ্রেড শুধু নিজের ঘরে লেখে, আর main সব join শেষে যোগ করে. কোনো দুই থ্রেড একই মেমরিতে লেখে না, তাই কোনো lock লাগে না। আর pthread_join শুধু অপেক্ষাই করে না, একটি নিশ্চয়তাও দেয়: join ফেরার পর থ্রেডটি যা যা লিখেছিল, সবই main দেখতে পাবে। ছোট একটি উদাহরণে ধাপে ধাপে দেখি — ১০টি সংখ্যা, ৩টি থ্রেড:

1a = {3,1,4,1,5,9,2,6,5,3}; N = 10; T = 32lo = id*N/T; hi = (id+1)*N/T;3for (i = lo; i < hi; i++) s += a[i];4partial[id] = s;5pthread_join(t[id], NULL);6total += partial[id];

এবার আসল আকারে — ১০ লাখ সংখ্যা (1, 2, …, 1000000), ৪টি থ্রেড। থ্রেডের নম্বরটি এখানে ঠিকানা হিসেবে নয়, void*-এর ভেতরে মান হিসেবে পাঠানো হয়েছে (অংশ ২-এর নিয়মের তৃতীয় পথ), আর ফল মেলানো হয়েছে সূত্র N(N+1)/2-এর সাথে:

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

#define N 1000000
#define T 4
static int a[N];
static long partial[T];                   /* one slot per thread */

static void *sum_part(void *arg) {
    long id = (long)(intptr_t)arg;        /* small integer carried in void* */
    long lo = id * N / T, hi = (id + 1) * N / T;
    long s = 0;                           /* local, private to this thread */
    for (long i = lo; i < hi; i++) s += a[i];
    partial[id] = s;                      /* each thread writes ONLY its slot */
    return NULL;
}

int main(void) {
    for (long i = 0; i < N; i++) a[i] = (int)(i + 1);      /* 1, 2, ..., N */
    pthread_t t[T];
    for (long id = 0; id < T; id++)
        pthread_create(&t[id], NULL, sum_part, (void *)(intptr_t)id);
    long total = 0;
    for (int id = 0; id < T; id++) {
        pthread_join(t[id], NULL);        /* after join, partial[id] is ready */
        printf("thread %d: a[%ld..%ld) -> %ld\n", id,
               (long)id * N / T, (long)(id + 1) * N / T, partial[id]);
        total += partial[id];
    }
    long expect = (long)N * (N + 1) / 2;
    printf("total = %ld, formula N(N+1)/2 = %ld, match: %s\n",
           total, expect, total == expect ? "yes" : "no");
    return 0;
}

thread 0: a[0..250000) -> 31250125000
thread 1: a[250000..500000) -> 93750125000
thread 2: a[500000..750000) -> 156250125000
thread 3: a[750000..1000000) -> 218750125000
total = 500000500000, formula N(N+1)/2 = 500000500000, match: yes

৪টি থ্রেড মানেই ৪ গুণ দ্রুত নয়: থ্রেড তৈরি আর join-এর খরচ আছে, অ্যারে যোগে আসল বাধা প্রায়ই মেমরির bandwidth, আর core কম থাকলে থ্রেডগুলো পালা করে একই core-এ চলে — তখন concurrency আছে, parallelism নেই। এই রানারের container-এ মাত্র ২টি CPU (পরের প্রোগ্রামে দেখবেন)। তাত্ত্বিক সীমা (Amdahl-এর সূত্র) পর্ব ১৪-এ।

৫ · Data race — একই counter-এ দুই থ্রেড

এবার উল্টো নকশা: দুটি থ্রেড একই global counter-এ প্রত্যেকে ১০ লাখবার counter++ করবে। গণিত বলে ফল ২০ লাখ। একই প্রোগ্রামে দ্বিতীয়বার একই কাজ হবে mutex দিয়ে। চালিয়ে দেখুন — একাধিকবার চালালে প্রথম সংখ্যাটি প্রায় প্রতিবার আলাদা আসবে:

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#define LOOPS 1000000
static long counter = 0;                  /* shared by both threads */
static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static void *inc_racy(void *arg) {
    (void)arg;
    for (int i = 0; i < LOOPS; i++) counter++;          /* load, add, store */
    return NULL;
}
static void *inc_locked(void *arg) {
    (void)arg;
    for (int i = 0; i < LOOPS; i++) {
        pthread_mutex_lock(&m);           /* only one thread at a time */
        counter++;
        pthread_mutex_unlock(&m);
    }
    return NULL;
}
static long run(void *(*fn)(void *)) {
    pthread_t x, y;
    counter = 0;
    pthread_create(&x, NULL, fn, NULL);
    pthread_create(&y, NULL, fn, NULL);
    pthread_join(x, NULL); pthread_join(y, NULL);
    return counter;
}
int main(void) {
    printf("online CPUs: %ld\n", sysconf(_SC_NPROCESSORS_ONLN));
    long r1 = run(inc_racy), r2 = run(inc_locked);
    printf("without lock: %ld  (== 2000000? %s)\n", r1, r1 == 2000000 ? "yes" : "no");
    printf("with mutex  : %ld  (== 2000000? %s)\n", r2, r2 == 2000000 ? "yes" : "no");
    return 0;
}

online CPUs: 2
without lock: 1407041  (== 2000000? no)
with mutex  : 2000000  (== 2000000? yes)

হারানো যোগ গেল কোথায়? counter++ C-তে এক লাইন, কিন্তু CPU-র কাছে তিন ধাপ: load, add, store. আর পর্ব ৯-এ দেখেছেন, প্রতিটি থ্রেডের register আলাদা — context switch-এ সেগুলো সংরক্ষিত হয়। নিচের ডায়াগ্রামে counter = 5 থেকে শুরু করে দুই থ্রেড একবার করে যোগ করছে; ফল হওয়ার কথা 7:

1// T1 on core 0        (counter++)2mov rax, [counter]     ; load3add rax, 1             ; add4mov [counter], rax     ; store5// T2 on core 1        (counter++)6mov rax, [counter]     ; load7add rax, 1             ; add8mov [counter], rax     ; store

কত কম হতে পারে? — একটি ছোট প্রমাণ

দুটি থ্রেড প্রত্যেকে N বার (N ≥ ২) load-add-store করলে শেষ ফল সর্বোচ্চ ২N — যখন কোনো যোগই মেশে না। কিন্তু সর্বনিম্ন কত? চমকে দেওয়ার মতো উত্তর: ২, N যত বড়ই হোক। এটি বিশ্ববিদ্যালয় ও ইন্টারভিউর একটি প্রিয় প্রশ্ন।

• ২ সম্ভব (নির্মাণ): T1 load করে 0. তারপর T2 একটানা N−১ বার পুরো ++ করে (counter = N−১). T1 store করে 1 (T2-এর সব কাজ মুছে গেল)। এবার T2 load করে 1. তারপর T1 বাকি N−১ বার পুরো ++ করে শেষ করে। শেষে T2 store করে 1 + 1 = 2. মোট ফল 2.
• ২-এর কম অসম্ভব (প্রমাণ): প্রতিটি store লেখে (পড়া মান + ১), আর পড়া মান কখনো ঋণাত্মক নয় — তাই প্রতিটি store-এর মান ≥ ১. শেষ ফল হলো সবশেষ store-এর মান; ধরা যাক সেটি থ্রেড X-এর N-তম (শেষ) ধাপের store. X-এর N-তম load ঘটেছে X-এর নিজের (N−১)-তম store-এর পরে, তাই তখন মেমরিতে কোনো একটি store-এর লেখা মান ছিল (প্রাথমিক 0 আর নেই), অর্থাৎ পড়া মান ≥ ১. তাহলে শেষ store ≥ ১ + ১ = ২. ∎

অর্থাৎ ১০ লাখ করে যোগের ফল তাত্ত্বিকভাবে ২ থেকে ২০,০০,০০০-এর যেকোনো কিছু হতে পারে। (প্রমাণটি ধরে নেয় যে load আর store মেমরিতে কোনো একটি ক্রমে ঘটে; বাস্তবের CPU আর compiler আরও অদ্ভুত আচরণের সুযোগ রাখে, যা নিচে দেখবেন।)

-O2 দিলে বাগ “সেরে যায়”?

একই প্রোগ্রাম gcc -O2 দিয়ে compile করে Wandbox-এ চালানো হয়েছিল (এই পাতার রানার optimization ছাড়া compile করে)। ফল:

online CPUs: 2
without lock: 2000000  (== 2000000? yes)
with mutex  : 2000000  (== 2000000? yes)

বাগ উধাও? না, লুকিয়েছে। Optimizer ধরে নেয় লুপ চলার সময় counter-কে আর কেউ বদলায় না, তাই পুরো লুপকে কার্যত একবার load, একবার ১০ লাখ যোগ, একবার store-এ নামিয়ে আনে। Race-এর জানালা প্রায় শূন্য, তাই প্রায় সবসময় ২০ লাখ — কিন্তু ওই একটি load-store মিশে গেলে ফল ঠিক ১০ লাখ। C11 আর C++11 মানদণ্ড বলে, data race থাকা প্রোগ্রামের আচরণ undefined, তাই optimization বদলালে ফলও বদলাতে পারে। শিক্ষা: সব optimization level-এ, বারবার, load দিয়ে পরীক্ষা করুন, আর ভরসা রাখুন race detector-এ (অংশ ৮), ভাগ্যে নয়।

Mutex নিয়ে এই পর্বে তিনটি অভ্যাসই যথেষ্ট: lock যত ছোট অংশ ঘিরে রাখা যায় তত ভালো; ফাংশন থেকে যে পথেই বের হোন (error-এর return-সহ), প্রতিটি lock-এর একটি unlock চাই; আর একাধিক mutex হলে সব থ্রেড একই ক্রমে নেবে, নইলে deadlock (পর্ব ২৭)। Race condition, critical section আর Peterson-এর সমাধানের পূর্ণ তত্ত্ব পর্ব ২২-এ।

৬ · ছোট একটি thread pool — mutex, condition variable আর work queue

ধরুন একটি ওয়েব সার্ভার প্রতিটি অনুরোধে নতুন থ্রেড বানায় আর কাজ শেষে ফেলে দেয়। প্রতিটি থ্রেড তৈরিতে stack বরাদ্দ, kernel-এ clone — খরচ আছে; আর হঠাৎ ১০ হাজার অনুরোধ এলে ১০ হাজার থ্রেড মেমরি আর context switch-এ সার্ভারকে ডুবিয়ে দিতে পারে। সমাধান: আগে থেকে অল্প কয়েকটি থ্রেড বানিয়ে রাখা, কাজ এলে একটি সারিতে রাখা, আর খালি থ্রেড সারি থেকে কাজ তুলে নেবে। একে বলে thread pool, আর সারিটি work queue.

এখানে নতুন হাতিয়ার condition variable. Mutex-এর কাজ “একবারে একজন” নিশ্চিত করা; condition variable-এর কাজ শর্ত পূরণ না হওয়া পর্যন্ত থ্রেডকে ঘুম পাড়িয়ে রাখা, আর কেউ সংকেত দিলে জাগানো। pthread_cond_wait(&cv, &mu) একটি অবিভাজ্য কাজে mutex ছেড়ে দিয়ে ঘুমায়, আর জাগার সময় mutex আবার নিয়ে তবেই ফেরে। নিচের pool-এ ৩টি worker আর ৮টি কাজ; কাজটি সরল (সংখ্যার বর্গ), যাতে কাঠামোটি স্পষ্ট দেখা যায়। প্রতিটি ফল নিজের ঘরে, আর ছাপা হয় সবাই join হওয়ার পরে — তাই আউটপুট প্রতিবার একই:

#include <pthread.h>
#include <stdio.h>
#define WORKERS 3
#define TASKS 8
static int queue[TASKS], head = 0, tail = 0, closed = 0;   /* work queue */
static long result[TASKS];
static pthread_mutex_t mu = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cv = PTHREAD_COND_INITIALIZER;
static void *worker(void *arg) {
    (void)arg;
    for (;;) {
        pthread_mutex_lock(&mu);
        while (head == tail && !closed)           /* empty: sleep, release mu */
            pthread_cond_wait(&cv, &mu);
        if (head == tail) { pthread_mutex_unlock(&mu); return NULL; }
        int task = queue[head++];                 /* take one job */
        pthread_mutex_unlock(&mu);
        result[task] = (long)task * task;         /* do the work, no lock held */
    }
}
int main(void) {
    pthread_t w[WORKERS];
    for (int i = 0; i < WORKERS; i++) pthread_create(&w[i], NULL, worker, NULL);
    for (int t = 0; t < TASKS; t++) {             /* submit jobs */
        pthread_mutex_lock(&mu);
        queue[tail++] = t;
        pthread_cond_signal(&cv);                 /* wake one sleeping worker */
        pthread_mutex_unlock(&mu);
    }
    pthread_mutex_lock(&mu); closed = 1;          /* no more jobs */
    pthread_cond_broadcast(&cv);                  /* wake everyone to exit */
    pthread_mutex_unlock(&mu);
    for (int i = 0; i < WORKERS; i++) pthread_join(w[i], NULL);
    for (int t = 0; t < TASKS; t++) printf("task %d -> %ld\n", t, result[t]);
}

task 0 -> 0
task 1 -> 1
task 2 -> 4
task 3 -> 9
task 4 -> 16
task 5 -> 25
task 6 -> 36
task 7 -> 49

এটি শেখার জন্য ইচ্ছে করে সরল। বাস্তবের pool-এ queue হয় নির্দিষ্ট আকারের বৃত্তাকার buffer, “সারি ভর্তি” হলে কাজ জমা দেওয়া থ্রেডও অপেক্ষা করে (দ্বিতীয় condition variable — পর্ব ২৬-এর producer-consumer), আর কাজ হিসেবে থাকে একটি function pointer ও একটি void*. Condition variable-এর পূর্ণ তত্ত্ব পর্ব ২৫-এ।

কতগুলো থ্রেড রাখবেন? — একটি হিসাব

CPU-নির্ভর কাজে থ্রেড সংখ্যা মোটামুটি core-এর সমান রাখলেই যথেষ্ট; বেশি দিলে শুধু context switch বাড়ে। কিন্তু যে কাজে থ্রেড বেশিরভাগ সময় অপেক্ষা করে (ডেটাবেস, নেটওয়ার্ক), সেখানে বেশি লাগে। Brian Goetz-দের বই Java Concurrency in Practice-এ প্রচলিত একটি আনুমানিক সূত্র:

উদাহরণ: ৮ core, পুরো CPU ব্যবহারের লক্ষ্য (১), প্রতিটি অনুরোধে ২০ ms হিসাব আর ৮০ ms ডেটাবেসের অপেক্ষা। থ্রেড ≈ ৮ × ১ × (১ + ৮০ ÷ ২০) = ৮ × ৫ = ৪০টি. প্রতিটি থ্রেড সময়ের ১/৫ ভাগ CPU চায়, তাই ৮টি core ভরাতে ৫ গুণ থ্রেড লাগে। এটি শুরুর অনুমান মাত্র — আসল সংখ্যা ঠিক হয় load test করে।

৭ · অন্য ভাষা আর OS-এ একই ধারণা

Pthreads শিখলে বাকি সব থ্রেড-API সহজ লাগে, কারণ নিচে ধারণা একই: তৈরি, অপেক্ষা, ফল নেওয়া, আর shared ডেটা রক্ষা। পার্থক্য মূলত সুবিধায় — ভাষা কতটা নিরাপত্তা নিজে দেয়। উদাহরণগুলো এই পাতায় চালানো যায় না (Windows-এরটি ছাড়া সবগুলো আলাদাভাবে চালিয়ে যাচাই করা); মন্তব্যে প্রত্যাশিত ফল।

#include <stdio.h>
#include <threads.h>

static int work(void *arg) {               /* returns int, not void* */
    int id = *(int *)arg;
    printf("C11 thread %d\n", id);
    return id * 10;
}
int main(void) {
    thrd_t t;
    int id = 7, res;
    if (thrd_create(&t, work, &id) != thrd_success) return 1;
    thrd_join(t, &res);                    /* res = 70 */
    printf("result %d\n", res);
    return 0;
}

#include <iostream>
#include <thread>
#include <vector>

int main() {
    std::vector<long> part(4, 0);
    {
        std::vector<std::jthread> pool;            // C++20
        for (int id = 0; id < 4; ++id)
            pool.emplace_back([id, &part] {         // lambda: no void* casts
                for (long i = id; i < 1000; i += 4) part[id] += i;
            });
    }                                              // every jthread joins here
    long total = 0;
    for (long p : part) total += p;
    std::cout << "total = " << total << '\n';      // 0+1+...+999 = 499500
}

import java.util.concurrent.*;

class Main {
    public static void main(String[] args) throws Exception {
        try (ExecutorService pool = Executors.newFixedThreadPool(3)) {  // a thread pool
            Future<Integer> f = pool.submit(() -> 6 * 7);               // a task
            System.out.println("result " + f.get());                     // waits, like join
        }                                                                // close(): Java 19+
        Thread v = Thread.ofVirtual().start(() -> System.out.println("hello from a virtual thread"));
        v.join();                                                        // Java 21+
    }
}

import sys, threading

def work(i, n, out):
    out[i] = sum(range(i, n, 4))        # each thread writes its own slot

out = [0] * 4
ts = [threading.Thread(target=work, args=(i, 1000, out)) for i in range(4)]
for t in ts: t.start()
for t in ts: t.join()
print("total =", sum(out))              # 499500
# Python 3.13+: is the GIL on in this interpreter?
print("GIL enabled:", sys._is_gil_enabled() if hasattr(sys, "_is_gil_enabled") else True)

#include <windows.h>
#include <process.h>
#include <stdio.h>

static unsigned __stdcall work(void *arg) {      /* Win32 thread function */
    int id = *(int *)arg;
    printf("Windows thread %d\n", id);
    return (unsigned)(id * 10);
}
int main(void) {
    int id = 7;
    HANDLE h = (HANDLE)_beginthreadex(NULL, 0, work, &id, 0, NULL); /* 0 = default stack */
    WaitForSingleObject(h, INFINITE);             /* like pthread_join */
    DWORD code;
    GetExitCodeThread(h, &code);                  /* the return value: 70 */
    CloseHandle(h);                               /* release our handle */
    printf("result %lu\n", (unsigned long)code);
    return 0;
}

৮ · থ্রেডের বাগ ধরা — ThreadSanitizer আর Helgrind

অংশ ৫-এর race ধরা গেছে, কারণ সঠিক উত্তর আগে থেকে জানা ছিল। বাস্তবে উত্তর জানা থাকে না, আর race হয়তো হাজারে একবার ঘটে — তাই টুল লাগে।

ThreadSanitizer (TSan)

GCC আর Clang-এ -fsanitize=thread দিয়ে compile করলে compiler প্রতিটি মেমরি-ব্যবহারের পাশে ছোট পরীক্ষার কোড বসিয়ে দেয়। প্রোগ্রাম চলার সময় ThreadSanitizer হিসাব রাখে কোন থ্রেড কোন ঠিকানা কখন ছুঁয়েছে, আর দুটি ব্যবহারের মাঝে কোনো happens-before সম্পর্ক (যেমন একই mutex, বা join) আছে কি না। না থাকলে — race. বড় সুবিধা: race-টি সেই রানে ভুল ফল না দিলেও TSan ধরে ফেলে, কারণ সে ফল নয়, সমন্বয়ের অভাব খোঁজে। অংশ ৫-এর প্রোগ্রামে TSan যা বলে, তার একটি উদাহরণ (সংক্ষেপিত; ঠিকানা আর লাইন নম্বর আপনার মেশিনে ভিন্ন হবে):

$ gcc -fsanitize=thread -g -O1 race_vs_mutex.c -o race && ./race
online CPUs: 8
==================
WARNING: ThreadSanitizer: data race (pid=31207)
  Write of size 8 at 0x55c3d8a0e018 by thread T2:
    #0 inc_racy race_vs_mutex.c:9 (race+0x12b4)

  Previous write of size 8 at 0x55c3d8a0e018 by thread T1:
    #0 inc_racy race_vs_mutex.c:9 (race+0x12b4)

  Location is global 'counter' of size 8 at 0x55c3d8a0e018 (race+0x4018)

  Thread T2 (tid=31210, running) created by main thread at:
    #0 pthread_create ... (libtsan.so.2+0x5f1a6)
    #1 run race_vs_mutex.c:25 (race+0x1386)
    #2 main race_vs_mutex.c:31 (race+0x13e9)
SUMMARY: ThreadSanitizer: data race race_vs_mutex.c:9 in inc_racy
==================
without lock: 1998071  (== 2000000? no)
with mutex  : 2000000  (== 2000000? yes)
ThreadSanitizer: reported 1 warnings

পড়ার নিয়ম: কোন ঠিকানায় (global counter), কোন দুই থ্রেড (T1, T2), কোন লাইনে (৯, অর্থাৎ counter++), আর থ্রেডগুলো কোথায় তৈরি। Mutex-ওয়ালা অংশের জন্য কোনো সতর্কবার্তা নেই — lock একটি happens-before সম্পর্ক তৈরি করে। TSan-এ প্রোগ্রাম কয়েক গুণ ধীর হয় আর মেমরিও কয়েক গুণ বেশি লাগে, তাই এটি টেস্ট আর CI-এর জন্য। এই পাতার রানারে TSan চলে না (container-এ তার বিশাল shadow memory বরাদ্দ করা যায়নি), তাই নিজের Linux বা WSL-এ চালিয়ে দেখুন।

Helgrind (Valgrind)

Recompile না করেই race খুঁজতে আছে Valgrind-এর Helgrind. এটি প্রোগ্রামকে একটি ভার্চুয়াল CPU-তে চালিয়ে প্রতিটি মেমরি-ব্যবহার আর pthread কল দেখে, আর race ছাড়াও ধরে lock নেওয়ার ক্রমের অসংগতি (সম্ভাব্য deadlock) ও pthread API-র ভুল ব্যবহার। উদাহরণ আউটপুট (সংক্ষেপিত):

$ gcc -g race_vs_mutex.c -o race && valgrind --tool=helgrind ./race
==31452== Possible data race during read of size 8 at 0x10C018 by thread #3
==31452== Locks held: none
==31452==    at 0x1091C6: inc_racy (race_vs_mutex.c:9)
==31452==    by 0x485396A: ??? (in /usr/libexec/valgrind/vgpreload_helgrind-amd64-linux.so)
==31452==    by 0x4909AC2: start_thread (pthread_create.c:442)
==31452==
==31452== This conflicts with a previous write of size 8 by thread #2
==31452== Locks held: none
==31452==    at 0x1091D1: inc_racy (race_vs_mutex.c:9)
==31452==  Address 0x10c018 is 0 bytes inside data symbol "counter"

“Locks held: none” — দুই দিকেই কোনো lock ধরা ছিল না, এটিই race-এর প্রমাণ। Helgrind (আর তার সঙ্গী DRD) TSan-এর চেয়ে অনেক ধীর, তবে recompile লাগে না। দুটি টুলই শুধু পরীক্ষার রানে সত্যিই চলা কোড-পথ দেখে, তাই ভালো টেস্ট-ইনপুট লাগে। Helgrind চলে Linux-এ (আর কিছু অন্য Unix-এ), macOS-এ Clang-এর TSan চলে, কিন্তু Windows-এর নিজস্ব compiler-এ এদের সাধারণত পাওয়া যায় না — সেখানে WSL-এ Linux হিসেবে পরীক্ষা করা সহজ পথ।

৯ · AI-এর যুগে থ্রেডের কোড

AI কয়েক সেকেন্ডে একটি thread pool লিখে দেবে, আর আপনার ল্যাপটপে সম্ভবত চলবেও। কিন্তু থ্রেডের কোডে “চলছে” আর “ঠিক আছে” এক কথা নয় — race-ওয়ালা কোড -O2-এ নিখুঁত উত্তর দিয়েছে, দেখেছেন। AI-এর concurrent কোডের সাধারণ ভুল:

AI-এর লেখা থ্রেডের কোড ব্যবহারের আগে — চেকলিস্ট

• প্রতিটি shared ভ্যারিয়েবলের জন্য বলতে পারি কোন mutex তাকে রক্ষা করে, বা কেন lock লাগে না (যেমন নিজের ঘর + join)
• -fsanitize=thread দিয়ে compile করে চালিয়েছি, কোনো warning আসেনি
• -O0 আর -O2 দুই ভাবেই, আর একাধিকবার (লুপে শতবার) চালিয়ে ফল একই পেয়েছি
• প্রতিটি থ্রেড হয় join হয়, নয় detach; থ্রেডের argument আর ফেরত মানের মেমরি যথেষ্ট সময় বৈধ থাকে
• কোড আমার OS-এর জন্য (POSIX বনাম Win32), আর ভাষার যে সংস্করণ বলা হয়েছে (যেমন Python-এর GIL বা free-threaded) তা আমার মেশিনেরটাই
• core কম এমন মেশিনে বা container-এ, আর অনেক বেশি load দিয়েও পরীক্ষা করেছি