Playing with Intel's Restricted Transactional Memory.

double_ended_queue.cpp

#include <immintrin.h>
#include <iostream>
#include <thread>


template <class fun>
inline void atomic(fun const& block) {
    while(_xbegin() != _XBEGIN_STARTED) 
        printf("Txn retry\n");
    block();
    _xend();
}

template <class T>
class TX_queue {

    struct node_type;

    struct node_data {
        const T value;
        node_type* next;
        node_type* prev;
        explicit node_data(const T &value) : value(value), next(), prev() { }
    };

    struct node_type : node_data {
        explicit node_type(const T &value) : node_data(value) { }
        static const int CACHE_LINE_SIZE = 64;
        static const int PADDING_SIZE = sizeof(node_data) < CACHE_LINE_SIZE ? CACHE_LINE_SIZE - sizeof(node_data) : 0;
        // one node must fill at least one cache line
        char dummy[PADDING_SIZE];
    };

    typedef node_type Node;

public:

    TX_queue() {
        head_ = tail_ = nullptr;
    }

    ~TX_queue() {
        while(head_ != nullptr) {
            auto tmp = head_;
            head_ = tmp->next;
            delete tmp;
        }
    }

    void push_back(const T &value) {
        auto new_node = new Node(value);
        atomic([=] {
            if (!tail_) {
                head_ = tail_ = new_node;
            }
            else {
                tail_->next = new_node;
                new_node->prev = tail_;
                tail_ = new_node;
            }
        });
    }

    void push_front(const T &value) {
        auto new_node = new Node(value);
        atomic([=] {
            if (!head_) {
                head_ = tail_ = new_node;
            }
            else {
                head_->prev = new_node;
                new_node->next = head_;
                head_ = new_node;
            }
        });
    }

    bool pop_front(T *result) {
        bool success = false;
        Node* tmp = nullptr;
        atomic([&] {
            tmp = head_;
            if (head_ != nullptr) {
                *result = head_->value;
                success = true;
                head_ = head_->next;
                if (head_)
                    head_->prev = nullptr;
                else
                    tail_ = nullptr;
            }
        });
        if (success) delete tmp;
        return success;
    }

    bool pop_back(T *result) {
        bool success = false;
        Node* tmp = nullptr;
        atomic([&] {
            tmp = tail_;
            if (tail_ != nullptr) {
                *result = tail_->value;
                success = true;
                tail_ = tail_->prev;
                if (tail_)
                    tail_->next = nullptr;
                else
                    head_ = nullptr;
            }
        });
        if (success) delete tmp;
        return success;
    }

private:
    Node* head_;
    Node* tail_;
};


int main(int argc, char* argv[])
{
    TX_queue<int> chan;
    std::thread producer([&chan]{
        printf("producer started\n");
        for (int i = 0; i < 10000; i++) {
            chan.push_back(i);
        }
        printf("producer done\n");
    });
    std::thread consumer([&chan]{
        printf("consumer started\n");
        int count = 0;
        int value = 0;
        while(count < 10000) {
            if (chan.pop_front(&value)) {
                if (count != value) {
                    printf("Error at %d", count);
                    return;
                }
                count++;
            }
        }
        printf("consumer done\n");
    });
    consumer.join();
    producer.join();
    return 0;
}