Lock-free Queue

কেন এই প্রজেক্ট

Atomics ও lock-free programming লেসনে আমরা CAS (compare-and-swap) কীভাবে কাজ করে দেখেছি, আর ABA problem-এর ধারণাটাও পরিচয় করানো হয়েছে — একটা মান A থেকে B হয়ে আবার A-তে ফিরে এলে, একটা stale CAS ভুলভাবে “কিছু বদলায়নি” ভেবে সফল হয়ে যায়। কিন্তু ABA শুধু একটা তাত্ত্বিক সতর্কতা হিসেবে পড়া, আর সেটা নিজের হাতে একটা প্রকৃত ডেটা স্ট্রাকচারে ঘটিয়ে, তারপর সত্যিকারের কোড দিয়ে সমাধান করা — সম্পূর্ণ ভিন্ন গভীরতার শিক্ষা।

এই প্রজেক্টে আমরা একটা Multi-Producer-Single-Consumer (MPSC) lock-free queue বানাব — একাধিক থ্রেড concurrently enqueue() করতে পারবে, একটা মাত্র থ্রেড dequeue() করবে। এই সীমাবদ্ধতা (single consumer) ইচ্ছাকৃত: এটা সমস্যাটাকে যথেষ্ট সহজ রাখে যাতে পুরো সমাধান একটা প্রজেক্টে আঁটে, অথচ ABA problem-টা এখনও পুরোপুরি বাস্তব থাকে (multi-producer contention-এর কারণে) — বাস্তব দুনিয়ায় ঠিক এই প্যাটার্নটাই দেখা যায় event queue, logging pipeline, আর job scheduler-এর মতো জায়গায়।

MPSC-তে কোথায় CAS লাগে, কোথায় লাগে না

এই একটা সিদ্ধান্তই পুরো ডিজাইনকে সহজ করে দেয়: যেহেতু শুধু একটা consumer থ্রেড কখনো dequeue() কল করে, head পয়েন্টার সরানোর জন্য কোনো CAS-ই লাগে না — দুইটা থ্রেড কখনো একই সাথে head এগোনোর চেষ্টা করবে না, তাই সাধারণ read/write যথেষ্ট। কিন্তু tail পয়েন্টার সম্পূর্ণ ভিন্ন গল্প — একাধিক producer থ্রেড একসাথে নতুন নোড জুড়তে চাইবে, তাই tail-এ প্রতিটা আপডেট একটা CAS-loop দিয়ে হতে হবে। ABA problem এখানে, tail-এর CAS-এই, বাসা বাঁধে — কারণ producer-রা এমন নোডের ঠিকানা নিয়ে কাজ করে যেগুলো consumer একই সময়ে “রিটায়ার” করে দিতে পারে।

ধাপ ১ — বেসিক CAS enqueue (এখনো ABA-vulnerable)

Michael আর Scott-এর ক্লাসিক ১৯৯৬ সালের lock-free queue-র কাঠামো অনুসরণ করব — একটা dummy/sentinel নোড দিয়ে শুরু, head->next থেকে আসল ডেটা পড়া হয়:

struct Node {
    int data;
    std::atomic<Node*> next{nullptr};
};

std::atomic<Node*> tail_naive;   // এখনো plain pointer — এখানেই সমস্যা লুকিয়ে

void enqueue_naive(Node* node) {
    Node* old_tail = tail_naive.load(std::memory_order_acquire);
    for (;;) {
        Node* next = old_tail->next.load(std::memory_order_acquire);
        if (next == nullptr) {
            Node* expected = nullptr;
            if (old_tail->next.compare_exchange_weak(expected, node,
                    std::memory_order_release, std::memory_order_relaxed)) {
                tail_naive.compare_exchange_weak(old_tail, node,
                    std::memory_order_acq_rel, std::memory_order_relaxed);
                return;
            }
        } else {
            tail_naive.compare_exchange_weak(old_tail, next,
                std::memory_order_acq_rel, std::memory_order_relaxed);   // "help" — কেউ আধা কাজ করে থেমেছে
        }
        old_tail = tail_naive.load(std::memory_order_acquire);
    }
}

কোডটা compile হয়, আর হালকা লোডে ঠিকঠাক কাজও করে। সমস্যাটা শুধু ভারী concurrent লোডে, খুব নির্দিষ্ট একটা টাইমিং-এ ধরা পড়ে।

ABA আসলে কীভাবে ঘটে — একটা concrete টাইমলাইন

ধরুন tail_naive বর্তমানে নোড D-কে পয়েন্ট করছে (ঠিকানা 0xA00)।

সময়	Producer P1	Producer P2	Consumer C
t1	`old_tail = tail_naive.load()` → `0xA00` (D) পড়ে ফেলে	—	—
t2	(OS দীর্ঘ সময়ের জন্য প্রিএম্পট করে P1-কে — কার্নেল-এ এটা বিরল না)	নিজের `enqueue()` সম্পন্ন করে: `D->next` সেট করে, তারপর `tail_naive`-কে D থেকে নতুন নোড E (`0xB00`)-এ CAS করে এগিয়ে দেয়	—
t3	—	—	`dequeue()` করে D-কে head-এর পেছনে ফেলে দেয়, `delete D` করে ফেলে (`0xA00` এখন মুক্ত মেমরি)
t4	—	আরেকটা `enqueue()` কল করে; `new Node()` — allocator-এর free-list থেকে দৈবক্রমে ঠিক `0xA00`-ই ফেরত পায় (এই নোডটার নাম দিই F); F, E-এর সাথে লিংক হয়ে নতুন tail হয়	—
t5	ঘুম থেকে জেগে ওঠে, `tail_naive.compare_exchange_weak(old_tail=0xA00, node=P1_new)` কল করে	—	—

t5-এ সমস্যাটা ঘটে: বর্তমান tail_naive-এর মান এখন 0xA00 (F-এর ঠিকানা) — আর P1-এর old_tail-ও 0xA00! ঠিকানা মিলে গেছে, CAS ভুলভাবে সফল হয় — যদিও 0xA00-এ থাকা নোডটা এখন লজিক্যালি সম্পূর্ণ ভিন্ন একটা entity (F, D না)। ফলাফল: tail_naive এখন P1-এর নোডে পয়েন্ট করছে, কিন্তু F থেকে সেই নোডের কোনো লিংক নেই — queue-র চেইন ছিঁড়ে গেছে, F-এর পরে যা-ই আসুক তা হারিয়ে যাবে বা মেমরি করাপশন ঘটবে।

মূল কথা: CAS শুধু “এই মেমরি লোকেশনের বর্তমান বিট-প্যাটার্ন কি আমার প্রত্যাশিত বিট-প্যাটার্নের সাথে মেলে” যাচাই করে — “এই ঠিকানার নোডটা কি এখনও সেই একই জিনিস যেটা আমি আগে দেখেছিলাম” যাচাই করে না। ঠিকানা রিইউজ হলে এই দুইটা প্রশ্নের উত্তর আলাদা হয়ে যায়।

সমাধান — Tagged pointer (versioned CAS)

ধারণা সহজ: প্রতিটা পয়েন্টারের সাথে একটা মনোটনিকভাবে বাড়তে-থাকা tag জুড়ে দিন, আর pointer+tag একসাথে, একটা মাত্র atomic ইউনিট হিসেবে CAS করুন (double-width CAS — x86-64-এ cmpxchg16b)। ঠিকানা রিইউজ হলেও, মাঝে যতগুলো tail-আপডেট হয়েছে ততবার tag বেড়ে গেছে — তাই ঠিকানা মিললেও tag মিলবে না, CAS সঠিকভাবে ব্যর্থ হবে।

#include <atomic>
#include <cstdint>
#include <cstddef>

template <typename T>
class MPSCQueue {
    struct Node {
        T data{};
        std::atomic<Node*> next{nullptr};
    };

    // 16 বাইট, কোনো padding নেই (Node* ও uint64_t দুটোই 8-বাইট-অ্যালাইনড) —
    // এটা জরুরি, কারণ compare_exchange বিট-বাই-বিট (memcmp-এর মতো) তুলনা করে,
    // operator== কল করে না — padding-এ garbage বিট থাকলে মিথ্যা CAS-ব্যর্থতা ঘটত
    struct TaggedPtr {
        Node* ptr;
        uint64_t tag;
    };

    std::atomic<TaggedPtr> tail_;
    Node* head_;   // শুধু consumer থ্রেড ছোঁয় — plain pointer, কোনো atomic দরকার নেই

public:
    MPSCQueue() {
        Node* dummy = new Node();
        head_ = dummy;
        tail_.store(TaggedPtr{dummy, 0}, std::memory_order_relaxed);
        // 16-বাইট CAS lock-free কি না রানটাইমে যাচাই করে নেওয়া ভালো অভ্যাস:
        // assert(tail_.is_lock_free());  // মিথ্যা হলে glibc mutex-fallback ব্যবহার করছে
    }

    ~MPSCQueue() {
        Node* n = head_;
        while (n) { Node* next = n->next.load(std::memory_order_relaxed); delete n; n = next; }
    }

    MPSCQueue(const MPSCQueue&) = delete;
    MPSCQueue& operator=(const MPSCQueue&) = delete;

    // একাধিক producer থ্রেড থেকে concurrently কল করা নিরাপদ
    void enqueue(T value) {
        Node* node = new Node();
        node->data = std::move(value);

        TaggedPtr old_tail = tail_.load(std::memory_order_acquire);
        for (;;) {
            Node* next = old_tail.ptr->next.load(std::memory_order_acquire);

            if (next != nullptr) {
                // আরেকজন producer already old_tail->next সেট করে ফেলেছে কিন্তু
                // tail_ এখনো এগোয়নি — "help" করে এগিয়ে দিই, তারপর আবার চেষ্টা
                TaggedPtr advanced{next, old_tail.tag + 1};
                tail_.compare_exchange_weak(old_tail, advanced,
                        std::memory_order_acq_rel, std::memory_order_acquire);
                continue;   // ব্যর্থ হলেও old_tail এখন compare_exchange-এর নিয়মেই আপ-টু-ডেট
            }

            Node* expected_next = nullptr;
            if (old_tail.ptr->next.compare_exchange_weak(expected_next, node,
                    std::memory_order_release, std::memory_order_relaxed)) {
                // লিংক-ইন সফল — tail_ এগিয়ে দেওয়ার চেষ্টা (ব্যর্থ হলেও ঠিক আছে,
                // অন্য কোনো producer/দেরিতে-আসা helper পরে এগিয়ে দেবে)
                TaggedPtr advanced{node, old_tail.tag + 1};
                tail_.compare_exchange_weak(old_tail, advanced,
                        std::memory_order_acq_rel, std::memory_order_relaxed);
                return;
            }
            // next-এ CAS ব্যর্থ — অন্য producer জিতেছে; old_tail রিফ্রেশ করে আবার চেষ্টা
            old_tail = tail_.load(std::memory_order_acquire);
        }
    }

    // শুধুমাত্র ONE consumer থ্রেড থেকে কল করা উচিত — MPSC-এর "SC" অংশের প্রতিশ্রুতি
    bool dequeue(T& out) {
        Node* first = head_;
        Node* next = first->next.load(std::memory_order_acquire);
        if (next == nullptr) return false;   // খালি

        out = std::move(next->data);
        head_ = next;
        // ইচ্ছাকৃতভাবে `delete first` করা হচ্ছে না — নিচের
        // "ট্যাগিং কী সমাধান করে না" অংশ দেখুন
        return true;
    }
};

কম্পাইল করুন: g++ -std=c++17 -O2 -mcx16 -pthread -o queue_test queue_test.cpp — -mcx16 ফ্ল্যাগ ছাড়া GCC/Clang x86-64-এ ১৬-বাইট atomic-কে সত্যিকারের lock-free cmpxchg16b-এর বদলে একটা লুকানো mutex দিয়ে emulate করে ফেলতে পারে (তখনও সঠিক ফলাফল দেবে, কিন্তু আর “lock-free” থাকবে না)।

প্রতিটা memory_order-এর justification

seq_cst সবজায়গায় বসিয়ে দেওয়া “নিরাপদ” মনে হতে পারে, কিন্তু memory consistency models লেসনের মূল শিক্ষা হলো — প্রতিটা সিঙ্ক্রোনাইজেশন পয়েন্টের একটা নির্দিষ্ট, ন্যূনতম প্রয়োজন আছে, আর সেটা justify করাই আসল কাজ:

অপারেশন	memory_order	কেন
`tail_.load()` (লুপের শুরুতে)	acquire	পরের লাইনেই `old_tail.ptr->next` dereference করব — যে থ্রেড এই tail_ মানটা লিখেছে, তার আগের সব লেখা (নোড initialization) visible হওয়া চাই
`old_tail.ptr->next.load()`	acquire	এই মানটা দিয়ে “help” করলে বা পরে dereference করলে, যে producer এটা লিখেছে তার কাজ visible থাকা দরকার
`next.compare_exchange_weak` — সফল হলে	release	এই মুহূর্তে node-টা (তার সম্পূর্ণ initialized `data`-সহ) অন্য থ্রেডের কাছে “প্রকাশ” করছি — এর সাথে জোড়া লাগা acquire-load-ই আসল synchronizes-with সম্পর্ক তৈরি করে
`next.compare_exchange_weak` — ব্যর্থ হলে	relaxed	ব্যর্থ হলে কিছুই প্রকাশ হয়নি, তাই কোনো ordering গ্যারান্টি লাগে না — শুধু retry হবে
`tail_.compare_exchange_weak` — সফল হলে	acq_rel	release অংশ: tail_-এর নতুন মান পরের acquire-load-এ সামঞ্জস্যপূর্ণভাবে দেখা যাক; acquire অংশ: এই একই কল যদি অভ্যন্তরীণভাবে রিফ্রেশ হয় তার জন্য প্রয়োজন
`tail_.compare_exchange_weak` — ব্যর্থ হলে	acquire	ব্যর্থ হলে `old_tail`-এ বর্তমান মান বসে যায় (C++ compare_exchange-এর নিয়মেই) — সেই রিফ্রেশড মান দিয়ে dereference করব বলে acquire দরকার
Consumer-এর `first->next.load()`	acquire	সবচেয়ে গুরুত্বপূর্ণ edge — producer-এর release-CAS-এর সাথে synchronizes-with তৈরি করে, নিশ্চিত করে `next->data` পড়ার সময় সেটা torn/অর্ধেক-লেখা মান না, সম্পূর্ণ initialized ডেটা
`head_` (consumer-exclusive)	কোনো atomic-ই না	মাত্র একটা থ্রেড কখনো এটা পড়ে বা লেখে — cross-thread visibility-র প্রশ্নই আসে না, তাই atomic নিষ্প্রয়োজন খরচ মাত্র

Tagging কী সমাধান করে, কী করে না

এইখানে সবচেয়ে সূক্ষ্ম আর প্রায়ই উপেক্ষিত বিষয়টা: tagging শুধু CAS-এর “তুলনা” ধাপটাকে নির্ভরযোগ্য করে — এটা কাউকে freed মেমরি dereference করা থেকে আটকায় না। লক্ষ্য করুন enqueue()-এ old_tail.ptr->next.load(...) লাইনটা CAS-এর আগেই old_tail.ptr-কে dereference করছে। যদি old_tail.ptr ইতিমধ্যে delete হয়ে যাওয়া মেমরি হয়, এই লাইনটা use-after-free — আর কোনো tag এটা ঠেকাতে পারে না, কারণ সমস্যাটা CAS-এর তুলনায় না, তার আগের dereference-এ।

ঠিক এই কারণেই এই প্রজেক্টের dequeue() ইচ্ছাকৃতভাবে delete first করে না — retired নোডগুলো লিক হয়ে যায় (শিক্ষার উদ্দেশ্যে একটা সচেতন, নিয়ন্ত্রিত simplification, প্রোডাকশন-রেডি সমাধান না)। বাস্তব দুনিয়ায় দুইটা রাস্তা আছে:

কখনো সত্যিকারের free() না করা — retired নোডগুলো একটা type-stable internal free-list-এ রিসাইকেল করা, যেটা শুধু এই queue-র নিজের নোড হিসেবেই আবার বরাদ্দ হয়। তখন dereference কখনো সত্যিকারের invalid মেমরিতে পড়ে না (সবসময় বৈধ Node লেআউট), tagging-ই যথেষ্ট।
Hazard pointer (বা RCU / epoch-based reclamation) — প্রতিটা থ্রেড dereference করার আগে একটা globally-visible “আমি এই ঠিকানাটা এখন ব্যবহার করছি” ঘোষণা করে; কোনো থ্রেড কোনো নোড free করার আগে সব hazard pointer স্ক্যান করে নিশ্চিত হয় কেউ এখনো সেটা ধরে নেই। এটাই libcds, Folly, আর বেশিরভাগ প্রোডাকশন-গ্রেড lock-free লাইব্রেরির আসল উত্তর — জটিলতা বেশি, কিন্তু সাধারণ allocator-এ নিরাপদে ফিরে যাওয়া যায়। এই প্রজেক্টে শুধু ধারণাটা উল্লেখ করা হলো; সম্পূর্ণ বাস্তবায়ন একটা স্বতন্ত্র, বড় প্রজেক্ট।

টেস্ট মেথডোলজি — মাল্টি-প্রোডিউসার স্ট্রেস টেস্ট

#include <thread>
#include <vector>
#include <cassert>
#include <cstdio>

int main() {
    constexpr int NUM_PRODUCERS = 8;
    constexpr int ITEMS_PER_PRODUCER = 100000;
    constexpr uint64_t ID_STRIDE = 1000000;   // থ্রেড-আইডি অনুযায়ী নেমস্পেস আলাদা রাখতে

    MPSCQueue<uint64_t> q;
    std::vector<std::thread> producers;
    for (int t = 0; t \< NUM_PRODUCERS; t++) {
        producers.emplace_back([&q, t]() {
            for (int i = 0; i \< ITEMS_PER_PRODUCER; i++)
                q.enqueue((uint64_t)t * ID_STRIDE + (uint64_t)i);   // globally-ইউনিক ID
        });
    }

    size_t total_expected = (size_t)NUM_PRODUCERS * ITEMS_PER_PRODUCER;
    std::vector<char> seen(NUM_PRODUCERS * ITEMS_PER_PRODUCER, 0);
    size_t received = 0;
    uint64_t value;

    while (received \< total_expected) {
        if (q.dequeue(value)) {
            int t = (int)(value / ID_STRIDE);
            uint64_t i = value % ID_STRIDE;
            assert(t >= 0 && t \< NUM_PRODUCERS && i < (uint64_t)ITEMS_PER_PRODUCER);
            size_t idx = (size_t)t * ITEMS_PER_PRODUCER + i;
            assert(!seen[idx] && "ডুপ্লিকেট এলিমেন্ট!");
            seen[idx] = 1;
            received++;
        }
        // খালি হলে busy-poll — প্রোডাকশন কোডে এখানে exponential backoff থাকা উচিত
    }

    for (auto &t : producers) t.join();
    for (char s : seen) assert(s && "কোনো এলিমেন্ট হারিয়ে গেছে!");

    printf("PASS: %zu এলিমেন্ট, কোনো লস/ডুপ্লিকেট/করাপশন ছাড়াই\n", received);
    return 0;
}

এই টেস্ট তিনটা জিনিস একসাথে যাচাই করে: (১) গণনা মিলছে (received == total_expected), (২) কোনো ডুপ্লিকেট নেই (assert(!seen[idx]) — কোনো ভ্যালু দুইবার দেখা গেলে সাথে সাথে ক্র্যাশ করবে), (৩) কোনো করাপশন নেই (decode করা t/i সবসময় বৈধ রেঞ্জে থাকতে হবে — না থাকলে মানে bit-pattern নষ্ট হয়ে গেছে)।

ThreadSanitizer দিয়ে চালান — এটা race condition ধরার সবচেয়ে নির্ভরযোগ্য টুল, শুধু “টেস্ট পাস করলো” তার চেয়ে অনেক বেশি আত্মবিশ্বাস দেয়:

g++ -std=c++17 -O1 -g -fsanitize=thread -mcx16 -pthread -o queue_tsan queue_test.cpp
./queue_tsan

উচ্চ iteration count-এ (৮ producer × ১ লক্ষ আইটেম, কয়েকবার রান) চালিয়ে দেখুন — লক্ষ রাখুন assert কখনো ট্রিগার না হয়, আর ThreadSanitizer কোনো data race রিপোর্ট না করে।

নিজেকে চ্যালেঞ্জ করুন

ABA পুনরুৎপাদন করুন — উপরের naive সংস্করণে ইচ্ছাকৃতভাবে std::this_thread::sleep_for() দিয়ে একটা producer থ্রেডকে ঠিক old_tail লোড করার পরপরই ঘুম পাড়িয়ে দিন, আর অন্য থ্রেডগুলোকে প্রচুর enqueue/dequeue করান — কতক্ষণে করাপশন ধরা পড়ে দেখুন
Hazard pointer (basic) — একটা single-slot hazard-pointer array বানিয়ে dequeue-তে আসল delete নিরাপদে যোগ করুন — শুধু তখনই free করুন যখন কোনো hazard slot সেই ঠিকানা ধরে নেই
Bounded queue — একটা fixed-capacity সংস্করণ বানান, producer-দের জন্য queue-ভর্তি হলে backpressure (block বা reject) যোগ করুন
Latency measurement — Cache Effect Measurement প্রজেক্টের টাইমিং কৌশল ব্যবহার করে enqueue/dequeue-এর per-operation latency মাপুন, contention বাড়ার সাথে (producer সংখ্যা বাড়িয়ে) latency কীভাবে বদলায় দেখুন
MPMC-এ সম্প্রসারণ — head_-কেও tagged-pointer CAS দিয়ে সুরক্ষিত করে একাধিক consumer সাপোর্ট করা পূর্ণাঙ্গ Michael-Scott queue বানান

এটা যেখানে গিয়ে মিশবে

এখানে যা শিখলেন	পরে কোথায় লাগবে
Tagged pointer দিয়ে CAS-level ABA সমাধান	Level 11-এর Memory Consistency Models লেসনের সাথে সরাসরি সংযুক্ত — এই প্রজেক্টই সেই তত্ত্বের ব্যবহারিক প্রয়োগ
প্রতিটা atomic op-এর memory_order justify করা	এই মডিউলেরই বাকি concurrency-নির্ভর কাজ — false sharing মাপা, profiling করার সময় সঠিক synchronization বোঝা জরুরি
Safe memory reclamation-এর সীমাবদ্ধতা বোঝা	Level 8 (Database Systems)-এর concurrent B+ Tree/buffer pool — একই ধরনের “কখন free করা নিরাপদ” সমস্যা, disk page-এর প্রেক্ষাপটে
ThreadSanitizer-ভিত্তিক verification methodology	এই মডিউলেরই Benchmarking Methodology লেসন — টুল দিয়ে দাবি যাচাই করা, শুধু “মনে হচ্ছে ঠিক আছে” না
Lock-free queue-র বাস্তব ব্যবহার	Level 12 (Virtualization/Cloud)-এর Mini Orchestrator প্রজেক্ট — scheduler-এর ভেতরের event/task queue প্রায়ই এই একই প্যাটার্নে বানানো হয়

মাইলস্টোন

আগে যা পড়া দরকার