Jimmy's blog

#include <iostream>
#include <cassert>
#include <pthread.h>
 
using namespace std;
 
template<typename T>
class PipeQueue
{
public:
    PipeQueue() {
        head = new Item();
        head->next = NULL;
        tail = head;
    }
    ~PipeQueue() {
        assert(head->next == NULL);
        delete head;
    }
 
    void Push(const T& data) {
        Item* it = new Item();
        it->next = NULL;
 
        tail->data = data;
        __sync_synchronize(); // important
        tail->next = it;
        tail = it;
    }
 
    bool Pop(T& ret) {
        Item* tmp = NULL;
        __sync_synchronize(); // important
        if (head->next != NULL) {
            tmp = head;
            head = head->next;
            ret = tmp->data;
            delete tmp;
            return true;
        } else {
            return false;
        }
    }
 
private:
    struct Item {
        T data;
        Item* next;
    };
 
private:
    Item* head;
    Item* tail;
};
 
static void* thread_writer(void* param)
{
    PipeQueue<int>* queue = static_cast<PipeQueue<int>*>(param);
    for (int i = 0; i < 35000; i++) {
        queue->Push(i);
    }
 
    return NULL;
}
 
static void* thread_reader(void* param)
{
    PipeQueue<int>* queue = static_cast<PipeQueue<int>*>(param);
    int i = 0;
    int tmp;
    while (true) {
        if (queue->Pop(tmp)) {
            if (i != tmp) {
                cout<<"data error"<<endl;
                break;
            }
            i++;
            if (i > 30000) {
                break;
            }
        }
    }
    return NULL;
}
 
static void test1()
{
    PipeQueue<int> queue;
    queue.Push(1);
    queue.Push(2);
    int tmp;
    assert(queue.Pop(tmp));
    assert(tmp == 1);
    assert(queue.Pop(tmp));
    assert(tmp == 2);
    queue.Push(3);
    queue.Push(4);
    assert(queue.Pop(tmp));
    assert(tmp == 3);
    assert(queue.Pop(tmp));
    assert(tmp == 4);
    assert(!queue.Pop(tmp));
    assert(!queue.Pop(tmp));
    assert(!queue.Pop(tmp));
 
    cout<<" test1 completed... "<<endl;
}
 
static void test2()
{
    PipeQueue<int> queue;
    pthread_t writer, reader;
 
    pthread_create(&reader, NULL, thread_reader, &queue);
    pthread_create(&writer, NULL, thread_writer, &queue);
 
    pthread_join(reader, NULL);
    pthread_join(writer, NULL);
 
    // clean the queue
    int tmp;
    while (queue.Pop(tmp)) {
    }
 
    cout<<" test2 completed... "<<endl;
}
 
int main(int argc, char **argv)
{
    test1();
    test2();
    getchar();
    return 0;
}