实现一个lockfree的队列——错误修改

上篇文章：实现一个lockfree的队列

写了上一个文章后，对于速度不满意，于是又琢磨了一下，发现了一个很严重的错误。

这个错误是同事发现的。他说曾经发现，有时某个线程能被挂起很长时间。

如果有这样的情况，就可能会造成第一轮的A线程和第二轮的B线程访问同一个下标，这无疑是会出问题的。

于是我把下面放值和取值的时候也做了个限定，只有一个线程能通过该锁，这样就安全了。

修改了标记的设定，不用预先设定的空值，而用另一个char队列实现标记，因为是char类型的，不但实现标记，还能实现标示状态。

同时用基础库的原子操作函数代替了自己写的，并做了些优化（这些优化是抄袭的），速度有所提高。10次读写在176秒左右。

static inline void prefetch(void *x) {
asm volatile("prefetcht0 %0" :: "m" (*(unsigned long *)x));
}

// 临界锁，线程安全
// ，
template <class T_Key>
class CQQueue_Lockfree
{
public:
CQQueue_Lockfree()
{
m_lMaxQueueSize = -1;
m_tpQueue = NULL;
m_tpFlgQueue = NULL;
m_lBegPos = 0;
m_lEndPos = 0;
};

~CQQueue_Lockfree()
{
if(m_tpQueue)
free(m_tpQueue);
if( m_tpFlgQueue )
free(m_tpFlgQueue);
};

var_4 InitQueue(var_u8 lMaxQueueSize)
{
m_lMaxQueueSize = lMaxQueueSize;
posix_memalign((void**)&m_tpQueue, 64, (lMaxQueueSize) * sizeof(T_Key));
if(m_tpQueue == NULL)
return -1;
posix_memalign((void**)&m_tpFlgQueue, 64, (lMaxQueueSize));
if(m_tpFlgQueue == NULL)
return -1;

m_lBegPos = 0;
m_lEndPos = 0;
memset(m_tpFlgQueue,0,m_lMaxQueueSize);
return 0;
};

void ResetQueue()
{
m_lBegPos = 0;
m_lEndPos = 0;
};

void ClearQueue()
{
if(m_tpQueue)
{
free(m_tpQueue);
m_tpQueue = NULL;
}
if(m_tpFlgQueue)
{
free(m_tpFlgQueue);
m_tpQueue = NULL;
}

m_lMaxQueueSize = -1;
m_lBegPos = 0;
m_lEndPos = 0;
};

void PushData(T_Key tKey)
{
register var_u8 cnt = 8;
register var_u8 pos = __sync_fetch_and_add(&m_lEndPos, cnt)%m_lMaxQueueSize;
while( !__sync_bool_compare_and_swap(m_tpFlgQueue+pos, 0, 1))
sched_yield();

m_tpQueue[pos] = tKey;
*(m_tpFlgQueue+pos) = 2;
};

T_Key PopData()
{
register var_u8 cnt =8;
register var_u8 pos = __sync_fetch_and_add(&m_lBegPos, cnt)%m_lMaxQueueSize;
while( !__sync_bool_compare_and_swap(m_tpFlgQueue+pos, 2, 3 ) )
sched_yield();

T_Key ret = m_tpQueue[pos];
*(m_tpFlgQueue+pos) = 0;
prefetch(&m_tpQueue[(pos+8*60)%m_lMaxQueueSize]);
prefetch(&m_tpFlgQueue[(pos+8*60)%m_lMaxQueueSize]);
return ret;
};
private:
var_u8 m_lMaxQueueSize  __attribute__((aligned(64)));
T_Key* m_tpQueue __attribute__((aligned(64)));
var_1* m_tpFlgQueue __attribute__((aligned(64)));
var_u8 m_lBegPos __attribute__((aligned(64)));
var_u8 m_lEndPos __attribute__((aligned(64)));
};

其实还有个更高速的优化，就是设置cpu的相关性，时间能减少三分之二，就是能达到60+秒。不过我觉得在实际应用中作用不大，就没加上。

至此lockfree学习告一段落，相对于waitlock，时间减少了一半（waitlock 10亿次是350秒）。