Arsitektur Concurrency Model CSP di C

Model konkurensi Go mengadopsi Communicating Sequential Processes (CSP), di mana thread bertukar pesan lewat channel daripada berbagi memori secara langsung. Di bahasa C, mengimplementasikan mekanisme ini membutuhkan manajemen manual menggunakan API POSIX Threads (pthread). Tanpa runtime Go yang mengotomatisasi penjadwalan dan pembersihan memori, pengembang C rentan menghadapi deadlock, race condition, buffer overflow pada antrean circular, serta bug use-after-free saat channel ditutup.

Untuk membangun abstraksi channel yang kuat di C, struktur data memerlukan kombinasi antrean melingkar berkapasitas tetap (bounded ring buffer), satu pthread_mutex_t untuk menjamin eksekusi atomik, dan dua pthread_cond_t untuk memberi sinyal ketersediaan slot tulis dan baca.

Struktur Data dan Validasi Kapasitas Antrean

Langkah pertama adalah mendefinisikan layout memori channel. Kapasitas channel harus divalidasi saat inisialisasi guna mencegah integer overflow dan alokasi memori berukuran nol.

#include <pthread.h>
#include <stdbool.h>
#include <stdlib.h>
#include <stdint.h>

typedef struct {
    void **buffer;
    size_t capacity;
    size_t size;
    size_t head;
    size_t tail;
    bool closed;
    pthread_mutex_t lock;
    pthread_cond_t cond_read;
    pthread_cond_t cond_write;
} channel_t;

channel_t *channel_create(size_t capacity) {
    if (capacity == 0 || capacity > SIZE_MAX / sizeof(void *)) {
        return NULL;
    }

    channel_t *ch = malloc(sizeof(channel_t));
    if (!ch) return NULL;

    ch->buffer = malloc(capacity * sizeof(void *));
    if (!ch->buffer) {
        free(ch);
        return NULL;
    }

    ch->capacity = capacity;
    ch->size = 0;
    ch->head = 0;
    ch->tail = 0;
    ch->closed = false;

    if (pthread_mutex_init(&ch->lock, NULL) != 0) goto err_buffer;
    if (pthread_cond_init(&ch->cond_read, NULL) != 0) goto err_mutex;
    if (pthread_cond_init(&ch->cond_write, NULL) != 0) goto err_cond_read;

    return ch;

err_cond_read:
    pthread_cond_destroy(&ch->cond_read);
err_mutex:
    pthread_mutex_destroy(&ch->lock);
err_buffer:
    free(ch->buffer);
    free(ch);
    return NULL;
}

Mitigasi Race Condition Channel ala Go

Pengiriman (send) dan penerimaan (receive) data harus dilindungi terhadap spurious wakeups. Penggunaan blok while saat memeriksa kondisi antrean wajib diterapkan, bukan pernyataan if biasa.

Operasi Send (Tulis)

Fungsi pengiriman memblokir eksekusi thread produsen jika antrean penuh, atau mengembalikan sinyal error jika channel telah ditutup.

bool channel_send(channel_t *ch, void *data) {
    if (!ch) return false;

    pthread_mutex_lock(&ch->lock);

    while (ch->size == ch->capacity && !ch->closed) {
        pthread_cond_wait(&ch->cond_write, &ch->lock);
    }

    if (ch->closed) {
        pthread_mutex_unlock(&ch->lock);
        return false;
    }

    ch->buffer[ch->tail] = data;
    ch->tail = (ch->tail + 1) % ch->capacity;
    ch->size++;

    pthread_cond_signal(&ch->cond_read);
    pthread_mutex_unlock(&ch->lock);
    return true;
}

Operasi Receive (Baca)

Fungsi penerima membaca data dari antrean circular. Jika buffer kosong dan channel ditutup, fungsi mengembalikan false yang mengindikasikan akhir dari data (mirip idiom val, ok := <-ch pada Go).

bool channel_recv(channel_t *ch, void **out) {
    if (!ch || !out) return false;

    pthread_mutex_lock(&ch->lock);

    while (ch->size == 0 && !ch->closed) {
        pthread_cond_wait(&ch->cond_read, &ch->lock);
    }

    if (ch->size == 0 && ch->closed) {
        pthread_mutex_unlock(&ch->lock);
        *out = NULL;
        return false;
    }

    *out = ch->buffer[ch->head];
    ch->head = (ch->head + 1) % ch->capacity;
    ch->size--;

    pthread_cond_signal(&ch->cond_write);
    pthread_mutex_unlock(&ch->lock);
    return true;
}

Lifecycle Cleanup dan Sanitasi Memory Pointer

Pembersihan memori pada channel melibatkan dua aspek: status operasional channel dan kepemilikan (ownership) pointer yang dikirim.

  • Penutupan Saluran (Close): Mengubah flag closed dan memanggil pthread_cond_broadcast untuk membangunkan seluruh thread pengirim dan penerima yang tertahan di condition variable.
  • Destruksi Buffer: Fungsi channel_destroy hanya boleh dieksekusi setelah seluruh thread pekerja selesai melakukan join. Menerapkan destruksi saat masih ada thread yang beroperasi memicu undefined behavior.
void channel_close(channel_t *ch) {
    if (!ch) return;
    pthread_mutex_lock(&ch->lock);
    ch->closed = true;
    pthread_cond_broadcast(&ch->cond_read);
    pthread_cond_broadcast(&ch->cond_write);
    pthread_mutex_unlock(&ch->lock);
}

void channel_destroy(channel_t *ch) {
    if (!ch) return;
    pthread_mutex_destroy(&ch->lock);
    pthread_cond_destroy(&ch->cond_read);
    pthread_cond_destroy(&ch->cond_write);
    free(ch->buffer);
    free(ch);
}
Peringatan Ownership: Channel hanya mentransfer pointer generik void *. Pastikan produsen mengalokasikan payload dan konsumen membebaskannya (free) untuk mencegah memory leak, atau kirimkan nilai yang berada di segmen static/stack global yang valid sepanjang masa hidup aplikasi.

Verifikasi Stress Test dengan ThreadSanitizer

Data race sering kali tidak tampak pada pengujian fungsional sederhana dengan beban rendah. Gunakan ThreadSanitizer (TSan) yang tertanam pada LLVM/Clang dan GCC untuk mendeteksi konflik sinkronisasi pada akses memori.

Kompilasi source code menggunakan argumen flag instrumentasi:

gcc -fsanitize=thread -g -O1 -pthread main.c -o channel_test

Uji program dengan skenario multi-producer multi-consumer (MPMC):

#include <stdio.h>
#include <assert.h>

#define NUM_ITEMS 100000

void *producer(void *arg) {
    channel_t *ch = (channel_t *)arg;
    for (intptr_t i = 1; i <= NUM_ITEMS; i++) {
        channel_send(ch, (void *)i);
    }
    return NULL;
}

void *consumer(void *arg) {
    channel_t *ch = (channel_t *)arg;
    void *val;
    size_t count = 0;
    while (channel_recv(ch, &val)) {
        count++;
    }
    return (void *)(intptr_t)count;
}

int main(void) {
    channel_t *ch = channel_create(64);
    pthread_t prod, cons;

    pthread_create(&prod, NULL, producer, ch);
    pthread_create(&cons, NULL, consumer, ch);

    pthread_join(prod, NULL);
    channel_close(ch);

    void *ret;
    pthread_join(cons, &ret);
    assert((intptr_t)ret == NUM_ITEMS);

    channel_destroy(ch);
    printf("Semua pengujian lolos tanpa data race.\n");
    return 0;
}

Jika sinkronisasi salah atau terdapat akses variabel non-atomik di luar perlindungan mutex, ThreadSanitizer akan mencetak laporan WARNING: ThreadSanitizer: data race lengkap dengan jejak stack pemanggil. Output yang bersih menandakan mekanisme sinkronisasi primitif telah terisolasi dengan benar.