宣告:此文件只做學習交流使用,請勿用作其他商業用途

author:朝陽_tony
E-mail : [email protected]
Create Date: 2013-8-5 19:31:33 Monday
Last Change: 2013-8-6 14:33:21 Tuesday

轉載請註明出處：http://blog.csdn.net/linzhaolover

intel DPDK交流群希望大家加入互相學習,QQ群號:289784125

本文章基於intel dpdk  的原始碼1.3.1 版本進行講解；

摘要

intel dpdk 提供了一套ring 佇列管理程式碼，支援單生產者產品入列，單消費者產品出列；多名生產者產品入列，多產品消費這產品出列操作；

我們以app/test/test_ring.c檔案中的程式碼進行講解，test_ring_basic_ex()函式完成一個基本功能測試函式；

1、ring的建立

    rp = rte_ring_create("test_ring_basic_ex", RING_SIZE, SOCKET_ID_ANY,
            RING_F_SP_ENQ | RING_F_SC_DEQ);

呼叫rte_ring_create函式去建立一個ring，

第一引數"test_ring_basic_ex"是這個ring的名字，

第二個引數RING_SIZE是ring的大小;

 第三個引數是在哪個socket id上建立 ，這指定的是任意；

第四個引數是指定此ring支援單入單出；

我看一下rte_ring_create函式主要完成了哪些操作；

rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);

執行讀寫鎖的加鎖操作；

mz = rte_memzone_reserve(mz_name, ring_size, socket_id, mz_flags);

預留一部分記憶體空間給ring，其大小就是RING_SIZE個sizeof(struct rte_ring)的尺寸；

        r = mz->addr;

        /* init the ring structure */
        memset(r, 0, sizeof(*r));
        rte_snprintf(r->name, sizeof(r->name), "%s", name);
        r->flags = flags;
        r->prod.watermark = count;
        r->prod.sp_enqueue = !!(flags & RING_F_SP_ENQ);
        r->cons.sc_dequeue = !!(flags & RING_F_SC_DEQ);
        r->prod.size = r->cons.size = count;
        r->prod.mask = r->cons.mask = count-1;
        r->prod.head = r->cons.head = 0;
        r->prod.tail = r->cons.tail = 0;

        TAILQ_INSERT_TAIL(ring_list, r, next);
 

將獲取到的虛擬地址給了ring，然後初始化她，prod 代表生成者，cons代表消費者；

生產者最大可以生產count個，其取模的掩碼是 count-1； 目前是0個產品，所以將生產者的頭和消費者頭都設定為0；其尾也設定未0；

rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);

執行讀寫鎖的寫鎖解鎖操作；

2、ring的單生產者產品入列

rte_ring_enqueue(rp, obj[i])

ring的單個入列；

__rte_ring_sp_do_enqueue

最終會呼叫到上面這個函式，進行單次入列，我們看一下它的實現；

    prod_head = r->prod.head;
    cons_tail = r->cons.tail;

暫時將生產者的頭索引和消費者的尾部索引交給臨時變數；

free_entries = mask + cons_tail - prod_head;

計算還有多少剩餘的儲存空間；

    prod_next = prod_head + n;
    r->prod.head = prod_next;

如果有足夠的剩餘空間，我們先將臨時變數prod_next 進行後移，同事將生產者的頭索引後移n個；

    /* write entries in ring */
    for (i = 0; likely(i < n); i++)
        r->ring[(prod_head + i) & mask] = obj_table[i];
    rte_wmb();

執行寫操作，將目標進行入隊操作，它並沒有任何大資料量的記憶體拷貝操作，只是進行指標的賦值操作，因此dpdk的記憶體操作很快，應該算是零拷貝；

r->prod.tail = prod_next;

成功寫入之後，將生產者的尾部索引賦值為prox_next ，也就是將其往後挪到n個索引；我們成功插入了n個產品；目前是單個操作，索引目前n=1；

3、ring的單消費者產品出列

rte_ring_dequeue(rp, &obj[i]);

同樣出隊也包含了好幾層的呼叫，最終定位到__rte_ring_sc_do_dequeue函式；

    cons_head = r->cons.head;
    prod_tail = r->prod.tail;

先將消費者的頭索引和生產者的頭索引賦值給臨時變數；

entries = prod_tail - cons_head;

計算目前ring中有多少產品；

    cons_next = cons_head + n;
    r->cons.head = cons_next;

如果有足夠的產品，就將臨時變數cons_next往後挪到n個值，指向你想取出幾個產品的位置；同時將消費者的頭索引往後挪到n個；這目前n=1；因為是單個取出；

    /* copy in table */
    rte_rmb();
    for (i = 0; likely(i < n); i++) {
        obj_table[i] = r->ring[(cons_head + i) & mask];
    }

執行讀取操作，同樣沒有任何的大的資料量拷貝，只是進行指標的賦值；

 r->cons.tail = cons_next;

最後將消費者的尾部索引也像後挪動n個，最終等於消費者的頭索引；

4、ring的多生產者產品入列

 多生產者入列的實現是在 __rte_ring_mp_do_enqueue()函式中；在dpdk/lib/librte_ring/rte_ring.h 檔案中定義；其實這個函式和單入列函式很相似；

    /* move prod.head atomically */
    do {
        /* Reset n to the initial burst count */
        n = max;
.................

        prod_next = prod_head + n;
        success = rte_atomic32_cmpset(&r->prod.head, prod_head,
                          prod_next);
    } while (unlikely(success == 0));

在單生產者中時將生產者的頭部和消費者的尾部直接賦值給臨時變數，去求剩餘儲存空間；最後將生產者的頭索引往後移動n個，

但在多生產者中，要判斷這個頭部是否和其他的生產者發出競爭，

        success = rte_atomic32_cmpset(&r->prod.head, prod_head,
                          prod_next);

是否有其他生產者修改了prod.head，所以這要重新判斷一下prod.head是否還等於prod_head，如果等於，就將其往後移動n個，也就是將prod_next值賦值給prod.head;

如果不等於，就會失敗，就需要進入do while迴圈再次迴圈一次；重新重新整理一下prod_head和prod_next 以及prod.head的值 ；

    /* write entries in ring */
    for (i = 0; likely(i < n); i++)
        r->ring[(prod_head + i) & mask] = obj_table[i];
    rte_wmb();

執行產品寫入操作；

寫入操作完成之後，如是單生產者應該是直接修改生產者尾部索引，將其往後順延n個，但目前是多生產者操作；是怎樣實現的呢？

    /*
     * If there are other enqueues in progress that preceeded us,
     * we need to wait for them to complete
     */
    while (unlikely(r->prod.tail != prod_head))
        rte_pause();

    r->prod.tail = prod_next;

這也先進行判斷，判斷當前的生產者尾部索引是否還等於，儲存在臨時變數中的生產者頭索引，

如果不等於，說明，有其他的執行緒還在執行，而且應該是在它之前進行儲存，還沒來得及更新prod.tail;等其他的生產者更新tail後，就會使得prod.tail==prod_head;

之後再更新，prod.tail 往後挪動n個，最好實現 prod.tail==prod.head==prod_next==prod_head+n;

5、ring的多消費者產品出列

多個消費者同時取產品是在__rte_ring_mc_do_dequeue()函式中實現；定義在dpdk/lib/librte_ring/rte_ring.h檔案中；

    /* move cons.head atomically */
    do {
        /* Restore n as it may change every loop */
        n = max;

        cons_head = r->cons.head;
        prod_tail = r->prod.tail;
...................

        cons_next = cons_head + n;
        success = rte_atomic32_cmpset(&r->cons.head, cons_head,
                          cons_next);
    } while (unlikely(success == 0));

和多生產者一樣，在外面多包含了一次do while迴圈，防止多消費者操作發生競爭；

在迴圈中先將消費者的頭索引和生產者的為索引賦值給臨時變數；讓後判斷有多少剩餘的產品在迴圈佇列，

如有n個產品，就將臨時變數cons_next 往後挪動n個，然後判斷目前的消費者頭索引是否還等於剛才的儲存在臨時變數cons_head  中的值，如相等，說明沒有發生競爭，就將cons_next賦值給

消費者的頭索引   r->cons.head，如不相等，就需要重新做一次do while迴圈；

    /* copy in table */
    rte_rmb();
    for (i = 0; likely(i < n); i++) {
        obj_table[i] = r->ring[(cons_head + i) & mask];
    }

在成功更新消費者頭索引後，執行讀取產品操作，這並沒有大的資料拷貝操作，只是進行指標的重新賦值操作；

    /*
     * If there are other dequeues in progress that preceded us,
     * we need to wait for them to complete
     */
    while (unlikely(r->cons.tail != cons_head))
        rte_pause();

    __RING_STAT_ADD(r, deq_success, n);
    r->cons.tail = cons_next;

讀取完成後，就要更新消費者的尾部索引；

為了避免競爭，就要判是否有其他的消費者在更新消費者尾部索引；如果目前的消費者尾部索引不等於剛才儲存的在臨時變數cons_head 的值，就要等待其他消費者修改這個尾部索引；

如相等，機可以將當前消費者的尾部索引往後挪動n個索引值了，

實現  r->cons.tail=r->cons.head=cons_next=cons_head+n;

6、ring的其他判定函式

rte_ring_lookup("test_ring_basic_ex")

驗證以test_ring_basic_ex 為名的ring是否建立成功；

rte_ring_empty(rp)

判斷ring是否為空；

rte_ring_full(rp)

判斷ring是否已經滿；

rte_ring_free_count(rp)

判斷當前ring還有多少剩餘儲存空間；

技術水平有待提高，如果文章有錯誤的地方希望讀者指正，相互交流，互相學習；O(∩_∩)O~