Cleofwine | 开发者

后端技术_无锁消息队列实现RingBuffer

最近更新：2024-11-10 | 字数总计：3.6k | 阅读估时：15分钟 | 阅读量：次

为什么需要无锁队列
无锁队列的实现
1. ypipe_yqueue无锁队列
2. 循环数组无锁队列
无锁队列总结

前提：工作中能用有锁队列解决的问题尽量使用有锁解决，假设性能不能满足要求时，再考虑无锁队列。

为什么需要无锁队列

锁引起的问题：
1. cache损坏
  - 在保存和恢复上下文的过程中还隐藏了额外的开销：Cache中的数据会失效,因为它缓存的是将被换出任务的数据,这些数据对于新换进的任务是没用的。线程被频繁抢占产生的Cache损坏将导致应用程序性能下降。
  - 解决：减少线程切换
2. 在同步机制上的争抢队列
  - 任务将大量的时间(睡眠，等待，唤醒)浪费在获得保护队列数据的互斥锁，而不是处理队列中的数据上。
  - 解决：CAS
3. 动态内存分配
  - 当一个任务从堆中分配内存时，标准的内存分配机制会阻塞所有与这个任务共享地址空间的其它任务(进程中的所有线程)。这样做的原因是让处理更简单，且它工作得很好。两个线程不会被分配到一块相同的地址的内存，因为它们没办法同时执行分配请求。显然线程频繁分配内存会导致应用程序性能下降。
  - 解决：使用固定大小的数组；使用链表块；
无锁队列的限制：
- 假设cpu调度时间 << 任务耗时，那么无锁队列和有锁队列的性能几乎没有区别。
- 反过来，才是无锁队列的应用场景。

无锁队列的实现

ypipe_yqueue无锁队列

提取自zmq，可以直接用，适用于一读一写的应用场景
通过chunk模式批量分配节点，减少因为动态内存分配线程之间的互斥
- 分配链表块chunk
通过spare_chunk的作用（消息队列水位局部性原理，一般消息数量在一个位置上下波动）来降低chunk的频繁分配和释放
- 释放chunk的时候我们可以思考是直接调用free？还是可以用变量存起来，之后需要新的chunk时直接使用？
通过预写机制，批量更新写入位置，减少cas的调用
- 日志写操作：1.每写一次，直接flush，直接落盘（实时性强）；2.先fwrite写10条，再统一落盘；
- 在无锁队列中，有类似的操作，我们先yqueue.write几次后，才进行yqueue.flush，也就是一次write后我们并不急着更新可读指针的位置，而是多次后才更新，也就是write后读端并不能立即获取到最新的可读的位置。

巧妙的唤醒机制

读端没有数据可读时可以进行wait状态

写端在写入数据时可以根据返回值获知写入数据前消息队列是否为空，如果写入之前为空则可以唤醒读端

// 无锁队列的使用注意点，写端：
yqueue.write(...);
if(!yqueue.flush()){ // 这个函数只有在write之前队列是空的的时候才会返回false
    std::unique_lock<std::mutex> lock(ypipe_mutex_); 
    ypipe_cond_.notify_one(); // 只有这时我们才会去notify_one，注意和有锁队列的区别，有锁队列我们write一次就去notify一次，而不是等flush返回false才去notify
}
// 读端：
if(yqueue.read(&value)){
    ...
}
else{ // read返回false，说明是空的
    std::unique_lock<std::mutex> lock(ypipe_mutex_); 
    ypipe_cond_.wait(); // 此时我们wait
}

ypipe_yqueue无锁队列的使用

初始化ypipe_t<int, 100> yqueue 设置元素类型为int，每个chunk为100个元素
写端
- yqueue.write(count, false) 写入元素为count，false代表这次已经写完数据
- yqueue.flush 返回false，刷新之前队列为空（可以notify唤醒写端）；返回true说明队列写之前就有数据
读端
- yqueue.read读取元素返回true读到元素，返回false队列为空（可以让出cpu；或者wait写端唤醒）
写端，batch批量写入
- yqueue.write(count1, true)
- yqueue.write(count2, true)
- yqueue.write(count3, false)
- yqueue.flush flush后读端能看到更新后的数据

ypipe_t<int, 100> yqueue;
// 生产者线程
void *yqueue_producer_thread(void *argv) {
    PRINT_THREAD_INTO();
    int count = 0;  // 这个只是赋值，目前没有作用，仅仅是为了和其他队列方式保持一致
    for (int i = 0; i < s_queue_item_num; i++) {
        yqueue.write(
            i,
            false); // enqueue的顺序是无法保证的，我们只能计算enqueue的个数
        count = s_count_push.fetch_add(1, std::memory_order_release);;
        yqueue.flush();
    }
    PRINT_THREAD_LEAVE();
    return NULL;
}

// 消费者线程
void *yqueue_consumer_thread(void *argv) {
    int last_value = 0;
    PRINT_THREAD_INTO();

    while (true) {
        int value = 0;
        if (yqueue.read(&value)) {
            if (s_consumer_thread_num == 1 && s_producer_thread_num == 1 &&
                (last_value + 1) != value) // 只有一入一出的情况下才有对比意义
            {
                // printf("pid:%lu, -> value:%d, expected:%d\n", pthread_self(),
                // value, last_value + 1);
            }
            s_count_pop.fetch_add(1, std::memory_order_release);
            test_work(s_test_work_for_value);
            last_value = value;
        } else {
            // printf("%s %lu no data, s_count_pop:%d\n", __FUNCTION__,
            // pthread_self(), s_count_pop); usleep(100);
            sched_yield(); // 让出cpu，也可以这里wait
        }

        if (s_count_pop >= s_queue_item_num * s_producer_thread_num) {
            // printf("%s dequeue:%d, s_count_pop:%d, %d, %d\n", __FUNCTION__,
            // last_value, s_count_pop, s_queue_item_num,
            // s_consumer_thread_num);
            break;
        }
    }
    PRINT_THREAD_LEAVE();
    return NULL;
}

ypipe_yqueue源码分析

yqueue_t

chunk数据结构
- N就是ypipe_t<int, 100> yqueue；初始化时传入的数量

队列首操作

begin_chunk指向队头的chunk
begin_pos用于指向当前元素位于当前chunk的位置

所以begin_pos会不断增加，假如begin_pos==N时，说明当前chunk用完了，此时我们需要begin_chunk位置=next，且begin_pos相应更改

inline void pop()
{
    if(++begin_pos == N)
    {
        chunk_t *o = begin_chunk;
        begin_chunk = begin_chunk->next;
        begin_chunk->prev = NULL;
        begin_pos = 0;
    }
    chunk_t *cs = spare_chunk.xchg(o); // 局部性原理, 先将这个chunk缓存到spare_chunk中，将之前缓存的chunk先free掉
    // 局部性原理：通常情况短时间，比如1分钟 队列的大小基本一致
    free(cs); // cs可能是空的，但是free(NULL)是没问题的，并不会做什么
}

队列尾操作

假如当前chunk的pos并不==N，说明不需要进行新的chunk的生成

inline void push()
{
    back_chunk = end_chunk;
    back_pos = end_pos; //

    if (++end_pos != N) //end_pos!=N表明这个chunk节点还没有满
        return;

    chunk_t *sc = spare_chunk.xchg(NULL); // 为什么设置为NULL？ 因为如果把之前值取出来了则没有spare chunk了，所以设置为NULL
    if (sc)                               // 如果有spare chunk则继续复用它
    {
        end_chunk->next = sc;
        sc->prev = end_chunk;
    }
    else // 没有则重新分配
    {
        // static int s_cout = 0;
        // printf("s_cout:%d\n", ++s_cout);
        end_chunk->next = (chunk_t *)malloc(sizeof(chunk_t)); // 分配一个chunk
        alloc_assert(end_chunk->next);
        end_chunk->next->prev = end_chunk;  
    }
    end_chunk = end_chunk->next;
    end_pos = 0;
}

ypipe_t

重点私有变量
1
2
3
4
T *w;
T *r;
T *f;
atomic_t<T> c;
1. c=NULL说明此时队列是空的，c控制唤醒、休眠的机制
2. r控制可读的位置
3. w用来识别队列是不是空，每次调用flush它都指向当前可写入的位置

初始化状态

inline ypipe_t()
{
    //  Insert terminator element into the queue.
    queue.push(); //yqueue_t的尾指针加1，开始back_chunk为空，现在back_chunk指向第一个chunk_t块的第一个位置

    //  Let all the pointers to point to the terminator.
    //  (unless pipe is dead, in which case c is set to NULL).
    r = w = f = &queue.back(); //就是让r、w、f、c四个指针都指向这个end迭代器
    c.set(&queue.back());
}

write状态变化

// 写入数据，incomplete参数表示写入是否还没完成，在没完成的时候不会修改flush指针，即这部分数据不会让读线程看到。
inline void write(const T &value_, bool incomplete_)
{
    //  Place the value to the queue, add new terminator element.
    queue.back() = value_;
    queue.push();

    //  Move the "flush up to here" poiter.
    if (!incomplete_)
    {
        f = &queue.back(); // 记录要刷新的位置     false 更新f
        // printf("1 f:%p, w:%p\n", f, w);
    }
    else
    {
        //  printf("0 f:%p, w:%p\n", f, w);
    }
}

flush状态变化

// 刷新所有已经完成的数据到管道，返回false意味着读线程在休眠，在这种情况下调用者需要唤醒读线程。
// 批量刷新的机制， 写入批量后唤醒读线程；
// 反悔机制 unwrite
inline bool flush()
{
    //  If there are no un-flushed items, do nothing.
    if (w == f) // 不需要刷新，即是还没有新元素加入
        return true;

    //  Try to set 'c' to 'f'.
    // c何时为NULL？read时如果没有数据可以读取则c的值会被置为NULL，此时必然不会==w
    /*
        cas函数，原子操作，线程安全的，将ptr（调用者）与cmp_(第一个参数)进行比较：
            相等：把ptr设置尾val_（第二个参数）的值，返回ptr设置之前的值
            不相等：直接返回ptr的值
    */
    if (c.cas(w, f) != w) // 尝试将c设置为f，即是准备更新w的位置
    {

        //  Compare-and-swap was unseccessful because 'c' is NULL.
        //  This means that the reader is asleep. Therefore we don't
        //  care about thread-safeness and update c in non-atomic
        //  manner. We'll return false to let the caller know
        //  that reader is sleeping.
        c.set(f); // 更新w的位置
        w = f;
        // 注意这里就是flush返回false的逻辑，所以返不返回false仅和c的状态有关系，所以c就是用来控制唤醒逻辑的变量
        return false; //线程看到flush返回false之后会发送一个消息给读线程，这需要写业务去做处理
    }
    else  // 读端还有数据可读取
    {
        //  Reader is alive. Nothing special to do now. Just move
        //  the 'first un-flushed item' pointer to 'f'.
        w = f;             // 只需要更新w的位置
        return true;
    }
}

read状态变化

//  Check whether item is available for reading.
// 这里面有两个点，一个是检查是否有数据可读，一个是预取
inline bool check_read()
{
    //  Was the value prefetched already? If so, return.
    if (&queue.front() != r && r) //判断是否在前几次调用read函数时已经预取数据了return true;
        return true;

    //  There's no prefetched value, so let us prefetch more values.
    //  Prefetching is to simply retrieve the
    //  pointer from c in atomic fashion. If there are no
    //  items to prefetch, set c to NULL (using compare-and-swap).
    // 两种情况
    // 1. 如果c值和queue.front()相等，此时还没有写入数据， 返回c值并将c值置为NULL，此时没有数据可读
    // 2. 如果c值和queue.front()不相等， 返回c值，此时可能有数据度的去
    r = c.cas(&queue.front(), NULL); //尝试预取数据

    //  If there are no elements prefetched, exit.
    //  During pipe's lifetime r should never be NULL, however,
    //  it can happen during pipe shutdown when items are being deallocated.
    if (&queue.front() == r || !r) //判断是否成功预取数据
        return false;

    //  There was at least one value prefetched.
    return true;
}

//  Reads an item from the pipe. Returns false if there is no value.
//  available.
inline bool read(T *value_)
{
    //  Try to prefetch a value.
    if (!check_read())
        return false;

    //  There was at least one value prefetched.
    //  Return it to the caller.
    *value_ = queue.front();
    queue.pop();
    return true;
}

ypipe_yqueue总结（重点）

chunk批量分配元素的方式，链表+数组；
spare_chunk局部性原理；
生产端在flush可以根据结果检测队列是否为NULL
消费端read队列检测是否为NULL

循环数组无锁队列

结构

入队

关键在与while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1)));这句话，当m_writeIndex被其他线程更改后，则不会实际写入数据，而是进入下一轮循环；
第二个关键点在于while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)))，这里就说明我们的多线程必须按照顺序去更新m_maximumReadIndex，否则只能进行循环检测；

这也是为什么这种实现方式在写线程过多的时候，性能反而下降很多的原因，一个是竞争失败的线程需要继续循环才能实际写入数据；第二个是m_maximumReadIndex必须按照顺序去更新，也就是1，2，3，4这四个线程，有可能执行顺序是4，3，2，1，但是更新的时候仍然必须按照1，2，3，4的顺序去执行；

template <typename ELEM_T, QUEUE_INT Q_SIZE>
bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::enqueue(const ELEM_T &a_data)
{
	QUEUE_INT currentWriteIndex;		// 获取写指针的位置
	QUEUE_INT currentReadIndex;
	// 1. 获取可写入的位置
	do
	{
		currentWriteIndex = m_writeIndex;
		currentReadIndex = m_readIndex;
		if(countToIndex(currentWriteIndex + 1) ==
			countToIndex(currentReadIndex))
		{
			return false;	// 队列已经满了	
		}
		// 目的是为了获取一个能写入的位置,因为存在多个线程请求写入位置，所以同一个位置不能被多个线程同时获得，通过cas去判断
		/*
		 比如线程1: currentWriteIndex = 0, currentReadIndex =0
		 	线程2: currentWriteIndex = 0, currentReadIndex =0
			此时就只有一个线程能CAS更新成功，比如此时线程1正常更新后，m_writeIndex=1(被设置，实际是(currentWriteIndex+1)=0+1=1)，更新成功返回true，线程1退出循环
			此时线程2 去做CAS比如的时候，变成了m_writeIndex=1和currentWriteIndex=0比如，需要返回50行更新新的currentWriteIndex = m_writeIndex = 1
		*/
	//  CAS(a_ptr, a_oldVal, a_newVal)
	// 如果 a_ptr == a_oldVal 则 更新为 a_ptr = a_newVal，并返回true
	// 如果 a_ptr != a_oldVal 则 直接返回false
	} while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1)));
	// 获取写入位置后 currentWriteIndex 是一个临时变量，保存我们写入的位置
	// We know now that this index is reserved for us. Use it to save the data
	m_thequeue[countToIndex(currentWriteIndex)] = a_data;  // 把数据更新到对应的位置

	// 2. 更新可读的位置，按着m_maximumReadIndex+1的操作， 这里要更新可读的位置，同样可能存在多线程来操作，这里让让给currentWriteIndex顺序的写入
 	// update the maximum read index after saving the data. It wouldn't fail if there is only one thread 
	// inserting in the queue. It might fail if there are more than 1 producer threads because this
	// operation has to be done in the same order as the previous CAS
	while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1))) // 更新可读取的位置
	{
		 // this is a good place to yield the thread in case there are more
		// software threads than hardware processors and you have more
		// than 1 producer thread
		// have a look at sched_yield (POSIX.1b)
		sched_yield();		// 当线程超过cpu核数的时候如果不让出cpu导致一直循环在此。
	}

	AtomicAdd(&m_count, 1);

	return true;

}

出队

template <typename ELEM_T, QUEUE_INT Q_SIZE>
bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::dequeue(ELEM_T &a_data)
{
	QUEUE_INT currentMaximumReadIndex;
	QUEUE_INT currentReadIndex;

	do
	{
		 // to ensure thread-safety when there is more than 1 producer thread
       	// a second index is defined (m_maximumReadIndex)
		currentReadIndex = m_readIndex;
		currentMaximumReadIndex = m_maximumReadIndex;

		if(countToIndex(currentReadIndex) ==
			countToIndex(currentMaximumReadIndex))		// 如果不为空，获取到读索引的位置
		{
			// the queue is empty or
			// a producer thread has allocate space in the queue but is 
			// waiting to commit the data into it
			return false;
		}
		// retrieve the data from the queue
		a_data = m_thequeue[countToIndex(currentReadIndex)]; // 从临时位置读取的

		// try to perfrom now the CAS operation on the read index. If we succeed
		// a_data already contains what m_readIndex pointed to before we 
		// increased it
		if(CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))
		{
			AtomicSub(&m_count, 1);	// 真正读取到了数据，元素-1
			return true;
		}
	} while(true);

	assert(0);
	 // Add this return statement to avoid compiler warnings
	return false;

}

无锁队列总结

能用有锁尽量用有锁
什么时候使用无锁队列，满足以下的条件可以考虑
- 所处理的任务没有阻塞（如果有mysql数据库操作，有redis操作就没必要）
- 每秒处理的任务超过百万条（ops>100w）
- 生产者不存在io阻塞（每秒产生任务1万多也没必要，如果有锁队列影响io的吞吐量也可以使用无锁）
- 生产者最好和消费者是1：1的关系，不要出现多写1读的情况

2024-11-10 该篇文章被 Cleofwine 归为分类: 服务端