摘要:前言內存數據結構,類似于的通道,底層基于共享內存互斥鎖實現,可實現用戶態的高性能內存隊列。是當前隊列占用的內存大小,用來指定是否使用共享內存是否使用鎖是否使用通知。
前言
內存數據結構 Channel,類似于 Go 的 chan 通道,底層基于 共享內存 + Mutex 互斥鎖實現,可實現用戶態的高性能內存隊列。Channel 可用于多進程環境下,底層在讀取寫入時會自動加鎖,應用層不需要擔心數據同步問題。
channel 在之前的文章中出現過,當時用于 manager 和 worker 進程之間進行通信的重要數據結構,主要用于 worker 進程通知 manager 進程重啟相應 worker 進程。
channel 數據結構channel 數據結構的屬性比較多,head 是隊列的頭部位置,tail 是隊列的尾部位置,size 是申請的隊列內存大小,maxlen 是每個隊列元素的大小,head_tag 和 tail_tag 用于指定隊列的頭尾是否循環被重置回頭部。bytes 是當前 channel 隊列占用的內存大小,flag 用來指定是否使用共享內存、是否使用鎖、是否使用 pipe 通知。mem 是 channel 的內存首地址。
typedef struct _swChannel_item { int length; char data[0]; } swChannel_item; typedef struct _swChannel { off_t head; off_t tail; size_t size; char head_tag; char tail_tag; int num; int max_num; /** * Data length, excluding structure */ size_t bytes; int flag; int maxlen; /** * memory point */ void *mem; swLock lock; swPipe notify_fd; } swChannel;channel 隊列 swChannel_new 創建隊列
創建隊列就是根據 flags 來初始化隊列的各個屬性,值得注意的是 maxlen,當申請內存的時候會多申請這些內存,用來防止內存越界。
swChannel* swChannel_new(size_t size, int maxlen, int flags) { assert(size >= maxlen); int ret; void *mem; //use shared memory if (flags & SW_CHAN_SHM) { mem = sw_shm_malloc(size + sizeof(swChannel) + maxlen); } else { mem = sw_malloc(size + sizeof(swChannel) + maxlen); } if (mem == NULL) { swWarn("swChannel_create: malloc(%ld) failed.", size); return NULL; } swChannel *object = mem; mem += sizeof(swChannel); bzero(object, sizeof(swChannel)); //overflow space object->size = size; object->mem = mem; object->maxlen = maxlen; object->flag = flags; //use lock if (flags & SW_CHAN_LOCK) { //init lock if (swMutex_create(&object->lock, 1) < 0) { swWarn("mutex init failed."); return NULL; } } //use notify if (flags & SW_CHAN_NOTIFY) { ret = swPipeNotify_auto(&object->notify_fd, 1, 1); if (ret < 0) { swWarn("notify_fd init failed."); return NULL; } } return object; }swChannel_push 入隊
入隊的時候,首先要先加鎖,然后調用 swChannel_in。
swChannel_in 邏輯很簡單,向隊列的尾部推送數據,如果當前 channel 尾部被重置,head 還未被重置,就需要先判斷剩余的內存是否夠用。
如果當前 channel 尾部未被重置,就可以放心的追加元素,因為 object->size 和真正申請的內存之前還有 maxlen 可以富余,不必考慮內存越界的問題。
int swChannel_push(swChannel *object, void *in, int data_length) { assert(object->flag & SW_CHAN_LOCK); object->lock.lock(&object->lock); int ret = swChannel_in(object, in, data_length); object->lock.unlock(&object->lock); return ret; } #define swChannel_full(ch) ((ch->head == ch->tail && ch->tail_tag != ch->head_tag) || (ch->bytes + sizeof(int) * ch->num == ch->size)) int swChannel_in(swChannel *object, void *in, int data_length) { assert(data_length <= object->maxlen); if (swChannel_full(object)) { return SW_ERR; } swChannel_item *item; int msize = sizeof(item->length) + data_length; if (object->tail < object->head) { //no enough memory space if ((object->head - object->tail) < msize) { return SW_ERR; } item = object->mem + object->tail; object->tail += msize; } else { item = object->mem + object->tail; object->tail += msize; if (object->tail >= object->size) { object->tail = 0; object->tail_tag = 1 - object->tail_tag; } } object->num++; object->bytes += data_length; item->length = data_length; memcpy(item->data, in, data_length); return SW_OK; }swChannel_push 出隊
swChannel_push 出隊的邏輯比較簡單,獲取隊列頭部位置,然后拷貝首部數據即可。當 head 超過 size 值,即可重置 head。
int swChannel_pop(swChannel *object, void *out, int buffer_length) { assert(object->flag & SW_CHAN_LOCK); object->lock.lock(&object->lock); int n = swChannel_out(object, out, buffer_length); object->lock.unlock(&object->lock); return n; } #define swChannel_empty(ch) (ch->num == 0) int swChannel_out(swChannel *object, void *out, int buffer_length) { if (swChannel_empty(object)) { return SW_ERR; } swChannel_item *item = object->mem + object->head; assert(buffer_length >= item->length); memcpy(out, item->data, item->length); object->head += (item->length + sizeof(item->length)); if (object->head >= object->size) { object->head = 0; object->head_tag = 1 - object->head_tag; } object->num--; object->bytes -= item->length; return item->length; }
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/30852.html
摘要:消息隊列的接受消息隊列的接受是利用函數,其中是消息的類型,該參數會取出指定類型的消息,如果設定的是爭搶模式,該值會統一為,否則該值就是消息發送目的的。環形隊列的消息入隊發送消息首先要確定環形隊列的隊尾。取模操作可以優化 前言 swoole 的底層隊列有兩種:進程間通信 IPC 的消息隊列 swMsgQueue,與環形隊列 swRingQueue。IPC 的消息隊列用于 task_wor...
摘要:修復添加超過萬個以上定時器時發生崩潰的問題增加模塊,下高性能序列化庫修復監聽端口設置無效的問題等。線程來處理網絡事件輪詢,讀取數據。當的三次握手成功了以后,由這個線程將連接成功的消息告訴進程,再由進程轉交給進程。此時進程觸發事件。 本文示例代碼詳見:https://github.com/52fhy/swoo...。 簡介 Swoole是一個PHP擴展,提供了PHP語言的異步多線程服務器...
摘要:表示的是兩個,當其中任意一個計算完并發編程之是線程安全并且高效的,在并發編程中經常可見它的使用,在開始分析它的高并發實現機制前,先講講廢話,看看它是如何被引入的。電商秒殺和搶購,是兩個比較典型的互聯網高并發場景。 干貨:深度剖析分布式搜索引擎設計 分布式,高可用,和機器學習一樣,最近幾年被提及得最多的名詞,聽名字多牛逼,來,我們一步一步來擊破前兩個名詞,今天我們首先來說說分布式。 探究...
閱讀 1648·2019-08-30 15:55
閱讀 973·2019-08-30 15:44
閱讀 866·2019-08-30 10:48
閱讀 2025·2019-08-29 13:42
閱讀 3179·2019-08-29 11:16
閱讀 1235·2019-08-29 11:09
閱讀 2053·2019-08-26 11:46
閱讀 611·2019-08-26 11:44