FreeRTOS 流缓冲与消息缓冲源码剖析:主体不用等待链表,也不用关中断的队列

/* FreeRTOS/stream_buffer.c StreamBuffer_t (节选) */typedef struct StreamBufferDef_t{    volatile size_t xTail;                       /* 读索引: 下一个读的位置 */    volatile size_t xHead;                       /* 写索引: 下一个写的位置 */    size_t xLength;                              /* 缓冲区总长(字节数+1, 见4.2) */    size_t xTriggerLevelBytes;                   /* 触发水位: 攒够几字节才叫醒读者 */    volatile TaskHandle_t xTaskWaitingToReceive; /* 等数据的任务(注意: 是句柄不是链表!) */    volatile TaskHandle_t xTaskWaitingToSend;    /* 等空间的任务(同上) */    uint8_t * pucBuffer;                         /* 存储区指针 */    uint8_t ucFlags;                              /* 流/消息/批量缓冲标记 */    UBaseType_t uxStreamBufferNumber;              /* 追踪编号(configUSE_TRACE_FACILITY=1 时才有) */    /* configUSE_SB_COMPLETED_CALLBACK=0：两个完成回调字段本工程不编入 */    UBaseType_t uxNotificationIndex;               /* 唤醒用的任务通知通道号,默认 0(见4.6) */} StreamBuffer_t;

/* FreeRTOS/include/message_buffer.h */#define xMessageBufferCreate( xBufferSizeBytes ) \    xStreamBufferGenericCreate( ( xBufferSizeBytes ), ( size_t ) 0, sbTYPE_MESSAGE_BUFFER, NULL, NULL )

#include"FreeRTOS.h"#include"task.h"#include"stream_buffer.h"/* 流缓冲 API */#include"message_buffer.h"/* 消息缓冲 API（注意：是它包含 stream_buffer.h，方向别记反） */#include<stdio.h>#include<string.h>/* strlen *//* 实验参数。注意本工程 configTICK_RATE_HZ=100，1 tick=10ms： *//* pdMS_TO_TICKS(<10) 会被取整为 0——所有延时参数都必须 ≥10ms */#define SBf_LOG_LINES        3       /* 实验一：每轮发 3 行变长日志 */#define SBf_LOG_ROUNDS       2       /* 实验一共 2 轮（6 行） */#define SBf_LOG_PERIOD_MS    500#define SBf_BURST_TIMES      5       /* 实验二：每轮写 5 次 */#define SBf_BURST_SIZE       4       /* 每次写 4 字节，一轮共 20 字节 */#define SBf_BURST_PERIOD_MS  100#define SBf_WATER_HIGH       20      /* 实验二第一轮水位：攒够 20 字节才醒 */#define SBf_ROUND_BYTES      ( SBf_BURST_TIMES * SBf_BURST_SIZE )/* printf 任务栈：标准库 printf 栈开销大，128 字容易触发栈溢出钩子，统一给 256 字 */#define SBf_PRINTF_STACK     ( configMINIMAL_STACK_SIZE * 2 )/* 两个内核对象：消息缓冲（实验一）+ 流缓冲（实验二） *//* MessageBufferHandle_t 本质就是 StreamBufferHandle_t 的别名（message_buffer.h） */static MessageBufferHandle_t xLogMb  = NULL;static StreamBufferHandle_t  xStream = NULL;

staticvoidvSbProducer( void *pvParameters ){    (void)pvParameters;    static const char *pcLines[ SBf_LOG_LINES ] = {        "boot ok",        "sensor ready ch0",        "voltage 3.27V temperature 26.5C pressure 1013hPa"    };    uint32_t i, round, k;    size_t   xLen;    uint32_t ulDummy;    for( ;; )    {        /* ---- 实验一：消息缓冲发变长日志，行长随内容变（7/16/48 字节） ---- */        for( round = 1; round <= SBf_LOG_ROUNDS; round++ )        {            for( i = 0; i < SBf_LOG_LINES; i++ )            {                xLen = strlen( pcLines[ i ] );                (void)xMessageBufferSend( xLogMb, pcLines[ i ], xLen,                                          pdMS_TO_TICKS( 100 ) );                printf( "任务A：发送 %u 字节日志\r\n", (unsigned int)xLen );                vTaskDelay( pdMS_TO_TICKS( SBf_LOG_PERIOD_MS ) );            }        }        vTaskDelay( pdMS_TO_TICKS( 1000 ) );   /* 留时间给消费者收尾打印 */        /* ---- 实验二：每轮写 5 次 × 4 字节；第一轮水位 20、第二轮水位 1 ---- */        (void)xStreamBufferSetTriggerLevel( xStream, SBf_WATER_HIGH );        for( round = 1; round <= 2; round++ )        {            printf( "任务A：第 %u 轮写入 5×4 字节\r\n", (unsigned int)round );            for( k = 0; k < SBf_BURST_TIMES; k++ )            {                (void)xStreamBufferSend( xStream, &ulDummy, SBf_BURST_SIZE,                                         pdMS_TO_TICKS( 100 ) );                vTaskDelay( pdMS_TO_TICKS( SBf_BURST_PERIOD_MS ) );            }            vTaskDelay( pdMS_TO_TICKS( 1000 ) );   /* 等消费者打印本轮统计 */            (void)xStreamBufferSetTriggerLevel( xStream, 1 );  /* 第二轮水位 1 */        }        /* 一轮结束，回到实验一重新开始 */    }}

staticvoidvSbConsumer( void *pvParameters ){    (void)pvParameters;    char     pcBuf[ 64 ];    uint32_t line, round, woken;    size_t   xNext, xGot, xTotal;    for( ;; )    {        /* ---- 实验一：整条收变长日志，行长 7/16/48 原样还原 ---- */        printf( "\r\n=== 实验一：消息缓冲传变长日志（行长原样还原）===\r\n" );        line = 0;                                    /* 已收行数: 收成一行才 +1 */        while( line < SBf_LOG_ROUNDS * SBf_LOG_LINES )        {            xNext = xStreamBufferNextMessageLengthBytes( xLogMb );  /* 只探长, 不消费 */            if( xNext == 0 || xNext > sizeof( pcBuf ) - 1 )            {                vTaskDelay( pdMS_TO_TICKS( 50 ) );   /* 暂时没消息, 歇 50ms 再探 */                continue;                            /* 没收成, 不计数 */            }            xMessageBufferReceive( xLogMb, pcBuf, xNext, portMAX_DELAY );            pcBuf[ xNext ] = '\0';            printf( "[LOG] %s\r\n", pcBuf );            line++;        }        /* ---- 实验二：收满 20 字节算一轮，统计从阻塞里被唤醒的次数 ---- */        printf( "\r\n=== 实验二：流缓冲触发水位（每轮写入 5×4=20 字节）===\r\n" );        for( round = 1; round <= 2; round++ )        {            woken  = 0;            xTotal = 0;            while( xTotal < SBf_ROUND_BYTES )            {                /* 请求 20 字节，有多少读多少——返回值是实拿字节数 */                xGot = xStreamBufferReceive( xStream, pcBuf, SBf_ROUND_BYTES,                                             portMAX_DELAY );                if( xGot > 0 )                {                    woken++;                    xTotal += xGot;                }            }            printf( "任务B：第 %u 轮被唤醒 %u 次，共收 %u 字节\r\n",                    (unsigned int)round, (unsigned int)woken,                    (unsigned int)xTotal );        }        /* 回到实验一，循环往复 */    }}

voidstream_buffer_demo( void ){    xLogMb  = xMessageBufferCreate( 256 );            /* 实验一：变长日志 */    xStream = xStreamBufferCreate( 128, SBf_WATER_HIGH );  /* 实验二：初始水位 20 */    xTaskCreate( vSbConsumer, "SbRcv", SBf_PRINTF_STACK, NULL,                tskIDLE_PRIORITY + 1, NULL );    xTaskCreate( vSbProducer, "SbSend", SBf_PRINTF_STACK, NULL,                tskIDLE_PRIORITY + 2, NULL );}

任务A：发送 7 字节日志        ← A(+2) 优先级高，调度器一启动就先打印发送=== 实验一：消息缓冲传变长日志（行长原样还原）===[LOG] boot ok                 ← 横幅由随后上 CPU 的 B 打印，之后 B 逐行收任务A：发送 16 字节日志[LOG] sensor ready ch0任务A：发送 48 字节日志[LOG] voltage 3.27V temperature 26.5C pressure 1013hPa（第二轮同上，6 行行长 7/16/48 原样进出）=== 实验二：流缓冲触发水位（每轮写入 5×4=20 字节）===任务A：第 1 轮写入 5×4 字节任务B：第 1 轮被唤醒 1 次，共收 20 字节        ← 水位 20：第 5 次写完才醒任务A：第 2 轮写入 5×4 字节任务B：第 2 轮被唤醒 5 次，共收 20 字节        ← 水位 1：每写 4 字节醒一次

/* FreeRTOS/stream_buffer.c xStreamBufferGenericCreate (节选) */prvInitialiseNewStreamBuffer( ( StreamBuffer_t * ) pvAllocatedMemory,       /* 结构体放在头部 */                              ( ( uint8_t * ) pvAllocatedMemory ) + sizeof( StreamBuffer_t ), /* 存储区紧随其后 */                              xBufferSizeBytes,                              xTriggerLevelBytes,                              ucFlags,                              pxSendCompletedCallback,                              pxReceiveCompletedCallback );

/* FreeRTOS/stream_buffer.c xStreamBufferGenericCreate (节选) */if( xBufferSizeBytes < ( xBufferSizeBytes + 1U + sizeof( StreamBuffer_t ) ) ){    xBufferSizeBytes++;    /* 申请 N 字节, 实际按 N+1 分配 */    pvAllocatedMemory = pvPortMalloc( xBufferSizeBytes + sizeof( StreamBuffer_t ) );}

/* FreeRTOS/stream_buffer.c xStreamBufferSend (完整骨架, ①~⑤ 对应上图) */size_txStreamBufferSend( StreamBufferHandle_t xStreamBuffer,                          const void * pvTxData,                          size_t xDataLengthBytes,                          TickType_t xTicksToWait ){    StreamBuffer_t * const pxStreamBuffer = xStreamBuffer;    size_t xReturn, xSpace = 0;                        /* xSpace 初始 0: 没进等待循环时靠 ③ 兜底 */    size_t xRequiredSpace = xDataLengthBytes;          /* 需求 = 要写的字节数 */    TimeOut_t xTimeOut;    size_t xMaxReportedSpace = 0;    xMaxReportedSpace = pxStreamBuffer->xLength - ( size_t ) 1;   /* 容量上限: 4.2 的 +1 字节不计入 */    /* ---------- ① 定需求 ---------- */    if( ( pxStreamBuffer->ucFlags & sbFLAGS_IS_MESSAGE_BUFFER ) != ( uint8_t ) 0 )    {        xRequiredSpace += sbBYTES_TO_STORE_MESSAGE_LENGTH;        /* 消息缓冲: 需求加上 4 字节前缀 */        configASSERT( xRequiredSpace > xDataLengthBytes );       /* 需求反而变小 = size_t 加法回绕溢出 */        if( xRequiredSpace > xMaxReportedSpace )        {            xTicksToWait = ( TickType_t ) 0;           /* 整条放不下: 等也白等 */        }    }    else    {        if( xRequiredSpace > xMaxReportedSpace )        {            xRequiredSpace = xMaxReportedSpace;        /* 流缓冲: 需求截断到缓冲大小(允许部分写) */        }    }    /* ---------- ② 等空间: 愿意等才进循环 ---------- */    if( xTicksToWait != ( TickType_t ) 0 )    {        vTaskSetTimeOutState( &xTimeOut );             /* 记下起始时刻: 多轮睡眠共用一个总超时 */        do        {            taskENTER_CRITICAL();                      /* 关中断只罩着"查+登记"几行, 不罩睡觉 */            {                xSpace = xStreamBufferSpacesAvailable( pxStreamBuffer );   /* 用 xTail 现场算空余 */                if( xSpace < xRequiredSpace )          /* 放不下: 清残信 → 登记 → 睡 */                {                    /* 睡前清掉先于这次查询到达的通知(唤醒你的那个早已被 wait 消费, 不会是它):*/                     /* 这些信号对应的空间变化已经算进上面这次查询里——*/                     /* 查完还不足, 说明全是过期信号, 不清会被它"假醒"白烧一轮(见 4.6) */                    ( void ) xTaskNotifyStateClearIndexed( NULL, pxStreamBuffer->uxNotificationIndex );                    configASSERT( pxStreamBuffer->xTaskWaitingToSend == NULL );        /* 单写者红线 */                    pxStreamBuffer->xTaskWaitingToSend = xTaskGetCurrentTaskHandle(); /* 登记进"等空间"槽位 */                }                else                {                    taskEXIT_CRITICAL();                    break;                              /* 空间够了: 退出循环, xSpace 是刚查的新值 */                }            }            taskEXIT_CRITICAL();                       /* 登记完立刻开中断——睡觉不占临界区 */            /* 睡在任务通知上(不是睡在链表上!): 读者腾空间时看到槽位里有你, 就朝这个通道发信号 */            ( void ) xTaskNotifyWaitIndexed( pxStreamBuffer->uxNotificationIndex,                                              ( uint32_t ) 0, ( uint32_t ) 0, NULL, xTicksToWait );            pxStreamBuffer->xTaskWaitingToSend = NULL; /* 醒了先注销: 下轮重新登记 */        } while( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE );  /* 总超时没到就回环重查 */    }    /* ---------- ③ 兜底查空间 ---------- */    if( xSpace == ( size_t ) 0 )    /* 还是 0: 没进等待循环(非阻塞), 或睡着前最后一次查到的就是 0——重查拿最终空余 */    {        xSpace = xStreamBufferSpacesAvailable( pxStreamBuffer );    }    /* ---------- ④ 搬数据(不在任何临界区!) ---------- */    xReturn = prvWriteMessageToBuffer( pxStreamBuffer, pvTxData, xDataLengthBytes, xSpace, xRequiredSpace );    /* ---------- ⑤ 按水位叫醒读者 ---------- */    if( xReturn > ( size_t ) 0 )    {        if( prvBytesInBuffer( pxStreamBuffer ) >= pxStreamBuffer->xTriggerLevelBytes )        {            prvSEND_COMPLETED( pxStreamBuffer );       /* 4.6 节那个宏: 发通知唤醒读者 */        }    }    return xReturn;    /* 实写字节数: 流缓冲可能被截断, 消息缓冲要么整条要么 0 */}

/* FreeRTOS/stream_buffer.c xStreamBufferReceive (节选) */if( ( pxStreamBuffer->ucFlags & sbFLAGS_IS_MESSAGE_BUFFER ) != ( uint8_t ) 0 ){    xBytesToStoreMessageLength = sbBYTES_TO_STORE_MESSAGE_LENGTH;   /* 消息缓冲: 门槛是 4 */}else{    xBytesToStoreMessageLength = 0;   /* 流缓冲: 有 1 字节就算有数据 */}

/* FreeRTOS/stream_buffer.c prvWriteBytesToBuffer (节选) */staticsize_tprvWriteBytesToBuffer( StreamBuffer_t * const pxStreamBuffer,                                     const uint8_t * pucData, size_t xCount, size_t xHead ){    /* 第一段: 从 xHead 写到缓冲区末尾 */    xFirstLength = configMIN( pxStreamBuffer->xLength - xHead, xCount );    ( void ) memcpy( ( void * ) ( &( pxStreamBuffer->pucBuffer[ xHead ] ) ),                     ( const void * ) pucData, xFirstLength );    /* 第二段: 放不下就回绕到开头写剩余的 */    if( xCount > xFirstLength )    {        ( void ) memcpy( ( void * ) pxStreamBuffer->pucBuffer,                         ( const void * ) &( pucData[ xFirstLength ] ), xCount - xFirstLength );    }    xHead += xCount;    if( xHead >= pxStreamBuffer->xLength ) { xHead -= pxStreamBuffer->xLength; }    return xHead;    /* 注意: 返回新位置, 不直接改字段(4.8 有原因) */}

/* FreeRTOS/stream_buffer.c prvBytesInBuffer */xCount = pxStreamBuffer->xLength + pxStreamBuffer->xHead;xCount -= pxStreamBuffer->xTail;              /* 无符号回绕安全 */if( xCount >= pxStreamBuffer->xLength ) { xCount -= pxStreamBuffer->xLength; }return xCount;    /* 没套圈(xHead>xTail): 差值被多加了一圈, 减掉; 套圈(xHead<xTail): 结果本来就对, 不动 */

volatile TaskHandle_t xTaskWaitingToReceive;   /* 等数据的任务: NULL=没人等 */volatile TaskHandle_t xTaskWaitingToSend;      /* 等空间的任务: NULL=没人等 */

/* FreeRTOS/stream_buffer.c sbSEND_COMPLETED 宏 (缺省实现, 节选) */vTaskSuspendAll();{    if( ( pxStreamBuffer )->xTaskWaitingToReceive != NULL )    {        ( void ) xTaskNotifyIndexed( ( pxStreamBuffer )->xTaskWaitingToReceive,                                     ( pxStreamBuffer )->uxNotificationIndex,                                     ( uint32_t ) 0,                                     eNoAction );      /* 只当信号用, 不带值 */        ( pxStreamBuffer )->xTaskWaitingToReceive = NULL;    }}( void ) xTaskResumeAll();

/* FreeRTOS/stream_buffer.c xStreamBufferSend 等待循环内 (发送侧等待) */( void ) xTaskNotifyStateClearIndexed( NULL, pxStreamBuffer->uxNotificationIndex );/* 然后才登记句柄、才去 xTaskNotifyWaitIndexed 睡 */

/* FreeRTOS/stream_buffer.c (节选) */if( prvBytesInBuffer( pxStreamBuffer ) >= pxStreamBuffer->xTriggerLevelBytes ){    prvSEND_COMPLETED( pxStreamBuffer );    /* 到水位才唤醒 */}

/* FreeRTOS/stream_buffer.c prvWriteMessageToBuffer (节选) */size_t xNextHead = pxStreamBuffer->xHead;       /* 先存在局部变量里! */if( xSpace >= xRequiredSpace ){    /* 先写 4 字节长度前缀——写进的是局部推进的 xNextHead */    xNextHead = prvWriteBytesToBuffer( pxStreamBuffer, ( const uint8_t * ) &( xMessageLength ),                                       sbBYTES_TO_STORE_MESSAGE_LENGTH, xNextHead );}if( xDataLengthBytes != ( size_t ) 0 ){    /* 数据写完, 最后一步才提交: 把推进完毕的 xNextHead 写回 xHead 字段 */    pxStreamBuffer->xHead = prvWriteBytesToBuffer( pxStreamBuffer, ( const uint8_t * ) pvTxData,                                                   xDataLengthBytes, xNextHead );}

/* FreeRTOS/stream_buffer.c xStreamBufferSendFromISR (节选) */xSpace = xStreamBufferSpacesAvailable( pxStreamBuffer );   /* 查空间 */xReturn = prvWriteMessageToBuffer( ... );                    /* 放得下多少写多少 */if( xReturn > ( size_t ) 0 ){    if( prvBytesInBuffer( pxStreamBuffer ) >= pxStreamBuffer->xTriggerLevelBytes )    {        prvSEND_COMPLETE_FROM_ISR( pxStreamBuffer, pxHigherPriorityTaskWoken );        /* 内部: taskENTER_CRITICAL_FROM_ISR 里 xTaskNotifyIndexedFromISR —— 中断篇的老规则 */    }}

— 写在最后 —

如果这篇文章对你有帮助，欢迎点赞、在看、转发三连支持，这是我持续更新的最大动力。

有问题欢迎在评论区指出，我都会认真看。

点个关注不迷路，更多硬核技术剖析持续更新中，下期再见。