基于CAS的低时延队列实现（原理示例）

https://github.com/Rtoax/test/tree/master/c/cas/cas-queue

/***  cas-queue.h*  *  提供简单低时延的 基于 CAS 操作的队列*  *  荣涛  2021年1月13日*  荣涛  2021年1月14日    添加枚举类型，封装简单接口，但是封装后时延显著增高，见 test-3.c*                      这套接口是按照 多入多出多消息传递 设计的。*  */
#ifndef __CAS_QUEUE_H
#define __CAS_QUEUE_H 1#ifndef __x86_64__
# error "Not support Your Arch, just support x86-64"
#endif#include <errno.h>
#include <assert.h>
#include <stdbool.h>/***  获取tick*/
#ifndef RDTSC
#define RDTSC() ({\register uint32_t a,d; \__asm__ __volatile__( "rdtsc" : "=a"(a), "=d"(d)); \(((uint64_t)a)+(((uint64_t)d)<<32)); \})
#endif/***  cas，默认 smp 架构*/
#ifndef CAS
#define CAS(ptr, val_old, val_new) ({ \char ret; \__asm__ __volatile__("lock; "\"cmpxchgl %2,%0; setz %1"\: "+m"(*ptr), "=q"(ret)\: "r"(val_new),"a"(val_old)\: "memory"); \ret;})
#endif#ifndef UNUSED
#define UNUSED __attribute__((unused))
#endif
#ifndef likely
#define likely(exp) __builtin_expect(!!exp, 1)
#endif
#ifndef unlikely
#define unlikely(exp) __builtin_expect(!!exp, 0)
#endif/***  队列状态*/
typedef enum  {READY_TO_CTOR = 1,  /* 准备填入参数 */CTORING,            /* 正在填入参数 */DONE_TO_CTOR,       /* 填写完参数 */READY_TO_ENQUEUE,   /* 准备入队 */ENQUEUING,          /* 正在入队 */READY_TO_DEQUEUE,   /* 准备出队 */DEQUEUING,          /* 正在出队 */
}QUEUE_STATUS;typedef unsigned long bits_set;
#define BITS_SET_INITIALIZER 0UL#define BITS_SETSIZE                    64
#define BITS_SET(bit, p_bits_set)       (*(p_bits_set) |= 1UL<<bit)
#define BITS_CLR(bit, p_bits_set)       (*(p_bits_set) &= ~(1UL<<bit))
#define BITS_ISSET(bit, p_bits_set)     (*(p_bits_set) & (1UL<<bit))
#define BITS_ZERO(p_bits_set)           (*(p_bits_set) = 0UL)/***  快速队列结构体*/
typedef struct fast_queue_s {unsigned long que_id;           //QUEUE_STATUSint bitmap_maxbits;             //bitmap_active 中的最大 bit 对应的十进制bits_set __attribute__((aligned(64))) bitmap_active;        //标记需要发送的消息 bitsvoid    *data[BITS_SETSIZE];    //传递的指针
#define FAST_QUEUE_INITIALIZER  {READY_TO_ENQUEUE, -1, BITS_SET_INITIALIZER, {NULL}}
}__attribute__((aligned(64))) fast_queue_t;static inline int UNUSED fast_queue_init(fast_queue_t *queue) {int i;if(unlikely(!queue)) {fprintf(stderr, "NULL pointer error\n");return -EINVAL;}memset(queue, 0, sizeof(fast_queue_t));queue->que_id = READY_TO_CTOR;queue->bitmap_maxbits = -1;return 0;
}static inline fast_queue_t *UNUSED fast_queue_create()
{fast_queue_t* queue = malloc(sizeof(fast_queue_t));assert(queue);fast_queue_init(queue);return queue;
}static inline int UNUSED fast_queue_destroy(fast_queue_t *queue)
{if(unlikely(!queue)) {fprintf(stderr, "NULL pointer error\n");return -EINVAL;}free(queue);return 0;
}static inline int fast_queue_can_add(fast_queue_t *queue) {return CAS(&queue->que_id, READY_TO_CTOR, READY_TO_CTOR);
}static inline int UNUSED fast_queue_add_msgs(fast_queue_t *queue, int idx, void *msg) {
#if 0    if(unlikely(!queue) || unlikely(!msg)) {fprintf(stderr, "NULL pointer error\n");return -EINVAL;}if(unlikely(idx<0) || unlikely(idx>BITS_SETSIZE)) {fprintf(stderr, "Message index must between %d and %d\n", 0, BITS_SETSIZE-1);return -EINVAL;}
#endif    while(1) {if(CAS(&queue->que_id, READY_TO_CTOR, CTORING)) {if(BITS_ISSET(idx, &queue->bitmap_active)) {fprintf(stderr, "Already set slot %d\n", idx);while(!CAS(&queue->que_id, CTORING, READY_TO_CTOR));break;}if(queue->bitmap_maxbits < idx) {queue->bitmap_maxbits = idx;}queue->data[idx] = (void*)msg;BITS_SET(idx, &queue->bitmap_active);while(!CAS(&queue->que_id, CTORING, READY_TO_CTOR));break;}}return 0;
}static inline int UNUSED fast_queue_add_msgs_done(fast_queue_t *queue) {while(!CAS(&queue->que_id, READY_TO_CTOR, READY_TO_ENQUEUE));
}static inline int UNUSED fast_queue_send(fast_queue_t *queue) {while(!CAS(&queue->que_id, READY_TO_ENQUEUE, READY_TO_DEQUEUE));
}static inline int UNUSED fast_queue_recv(fast_queue_t *queue, int (*msg_handler)(void *msg)) {#if 0    if(unlikely(!queue)) {fprintf(stderr, "NULL pointer error\n");return -EINVAL;}
#endif    int i = 0, nr_msg = 0;if(CAS(&queue->que_id, READY_TO_DEQUEUE, DEQUEUING)) {for(i=0; i<= queue->bitmap_maxbits; i++) {if(BITS_ISSET(i, &queue->bitmap_active)) {msg_handler(queue->data[i]);BITS_CLR(i, &queue->bitmap_active);nr_msg++;}}queue->bitmap_maxbits = -1;CAS(&queue->que_id, DEQUEUING, READY_TO_CTOR);            }return nr_msg;
}#endif /*<__CAS_QUEUE_H>*/

基于CAS的低时延队列实现（原理示例）相关推荐

Linux C语言在用户态实现一个低时延通知（eventfd）+轮询（无锁队列ring）机制的消息队列
目录 fastq.c fastq.h test-0.c test-1.c https://github.com/Rtoax/test/tree/master/ipc/github/fastq fast ...
QUIC - 低时延互联网传输层协议
QUIC - 低时延互联网传输层协议 QUIC(Quick UDP Internet Connection)是谷歌制定的一种基于UDP的低时延的互联网传输层协议.在2016年11月国际互联网工程任务组 ...
harmonyos基于arm么,华为架构师解读：HarmonyOS低时延高可靠消息传输原理
华为架构师解读:HarmonyOS低时延高可靠消息传输原理 [复制链接] 本文作者:zhangkesi,华为软件架构设计工程师这是一篇HarmonyOS低时延高可靠消息传输原理的介绍,希望对你有所帮 ...
图解通信原理与案例分析-25：5G NR超可靠低时延通信URLLC是通过什么技术降低延时的？
前言: 5G的三大应用场景之一是超可靠低时延通信URLLC,这里有两个关键词:超可靠性和超低延时. 这句话是相对于LTE而言的,5G NR是通过什么技术实现超可靠性和超低延时的呢?本文探讨和拆解这个问 ...
物联网通信之5G工业级路由器，基于高速率、低时延、大带宽快速组网
随着2019年工信部与三大运营商举行的5G商用启动仪式,我们正式的进入了5G 时代,至今已经是第5年了,基于5G技术的高速率.低时延.大带宽特点,5G网络被广泛应用在智慧交通.智慧城市.自助终端.工业 ...
乐观锁-基于CAS原理
乐观锁理论基础乐观锁的操作过程中其实没有没有任何锁的参与,乐观锁只是和悲观锁相对,严格的说乐观锁不能称之为锁.下面我们就通过乐观锁与悲观锁的对比来更好的理解乐观锁. 乐观锁与悲观锁的概念乐观锁:总 ...
华为云SparkRTC面向低时延、大通量传输业务的技术探索
摘要:网络和移动设备高速发展的今天,人们开始思考如何用更短的时间下载更大的文件,追求更快的速度.当下在稳定的基础上有什么方法可以提升速度呢? 本文分享自华为云社区<华为云SparkRTC面向低时 ...
IDC网络低时延无阻塞拥塞控制随想
传统的拥塞控制千篇一律: 假设数据流都是长流. 目标端主机或交换机提供反馈信息给源端主机,源端主机根据这些信息限制发送量. 无论交换节点是否参与,本质上这还是端到端拥塞控制,注定是滞后的. 传统的拥塞 ...
AtomicInteger源码分析——基于CAS的乐观锁实现
原文出处: bestStyle 1. 悲观锁与乐观锁我们都知道,cpu是时分复用的,也就是把cpu的时间片,分配给不同的thread/process轮流执行,时间片与时间片之间,需要进行cpu切换, ...

基于CAS的低时延队列实现（原理示例）

基于CAS的低时延队列实现（原理示例）相关推荐

最新文章

热门文章