参考论文 Implementing Lock-Free Queues
con_queue.hpp
#ifndef CON_QUEUE_H_
#define CON_QUEUE_H_#include <atomic>
#include <vector>using namespace std;
namespace eventsock{template<typename T>
class ConQueue{
private:T* m_data;int m_data_len;int m_tail;int m_head;atomic_uint m_size;public:ConQueue(T* data, int data_len): m_tail(0), m_size(0){m_data = data;m_data_len = data_len;// m_head表示第一个元素m_head = 0;// m_tail标记队列尾部，不包含元素m_tail = 0;}bool push(T value){int old_tail, new_tail;while(true){old_tail = m_tail;new_tail = (old_tail + 1) % m_data_len;//队列已满if(new_tail == m_head) return false;if(__sync_bool_compare_and_swap(&m_tail, old_tail, new_tail)){ m_data[old_tail] = value;m_size++;break;}}return true;}bool pop(){int old_head, new_head;while(true){old_head = m_head;if(old_head == m_tail) break;new_head = (old_head + 1) % m_data_len;if(__sync_bool_compare_and_swap(&m_head, old_head, new_head)){m_size--;return true;}}return false;}size_t size(){return m_size;}bool empty(){return m_head == m_tail;}void output(){int head = m_head, tail = m_tail;int pos = head;cout << "[";while(pos != tail){if(pos != head){cout << ",";}cout << m_data[pos];pos = (pos + 1) % m_data_len;}cout << "]" << endl;}vector<T> to_vector(){vector<T> vec;int head = m_head, tail = m_tail;int pos = head;while(pos != tail){vec.push_back(m_data[pos]);pos = (pos + 1) % m_data_len;}return vec;}};}
#endif测试例程
#include <iostream>
#include "con_queue.hpp"
#include <vector>
#include <thread>
#include <atomic>
#include <algorithm>
#include <queue>const int per_time = 500000;const int NUM = 100000;
int que[NUM];eventsock::ConQueue<int> con_queue(que, NUM);
void test_queue_push(int k){for(int i = k * per_time; i < k * per_time + per_time; i++){while(true){if(con_queue.push(i)) break;std::this_thread::sleep_for(std::chrono::milliseconds(10));};}
}
void test_queue_pop(int k){for(int i = k * per_time; i < k * per_time + per_time; i++){while(true){if(con_queue.pop()) break;std::this_thread::sleep_for(std::chrono::milliseconds(10));};}
}
void test_con_queue(){vector<thread> ths;for(int i = 0; i < 10; i++){ths.push_back(thread(test_queue_push, i));}for(int i = 0; i < 10; i++){ths.push_back(thread(test_queue_pop, i));}for(int i = 0; i < ths.size(); i++){ths[i].join();}int num = con_queue.size();con_queue.output();if(num == 0){printf("ConQueue success\n");}else{printf("EventConnection fail new_connection_num=%d\n", num);}
}
int main(int argc, char* argv[])
{test_con_queue();return 0;
}