Chinaunix首页 | 论坛 | 博客

GP-Kinggpking.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

gongping11

  • 博客访问： 1892609
  • 博文数量： 152
  • 博客积分： 3730
  • 博客等级： 上尉
  • 技术积分： 3710
  • 用 户 组： 普通用户
  • 注册时间： 2011-06-02 14:36
个人简介

减肥，运动，学习，进步...

文章分类

全部博文（152）

文章存档

2016年（14）

2015年（17）

2014年（16）

2013年（4）

2012年（66）

2011年（35）

我的朋友
最近访客
推荐博文
相关博文
LibEvent代码阅读--报文收发过程分析

分类： LINUX

2015-02-07 23:05:44

网络编程中收发包处理是非常重要的过程，网络通信的本质就是报文的收发，在报文收发的基础上进行相关逻辑的处理和扩展。因此在libevent中也有报文收发处理相关的实现，在libevent中报文的发送和接收支持多缓冲区以及零拷贝相关的技术，本文就libevent的收发包处理进行分析。

在网络编程中需要对数据进行一定的管理和设计，在libevent中提供了一系列的数据结构：
struct evbuffer_chain：该结构体主要用于抽象一块内存空间。采用链表的方式进行管理，其中有该内存大小的相关字段。
struct evbuffer：该结构体主要用于抽象内存空间的管理器，其中指出了内存块的多少等重要信息，另外包含了相关信息，用于标识该管理归属的socket。在网络编程中收发包处理都是基于socket，
struct bufferevent：该结构体中主要是对某一个socket的收发包队列的抽象，其中包含了接收队列和发送队列、事件,每个socket可以包含一个读事件和一个写事件以及对应的回调函数。

点击(此处)折叠或打开

  1. /* 缓存块*/
  2. /** A single item in an evbuffer. */
  3. struct evbuffer_chain {
  4.     /** points to next buffer in the chain */
  5.         /* 链表的实现 */
  6.     struct evbuffer_chain *next;

  8.     /** total allocation available in the buffer field. */
  9.     size_t buffer_len;

  11.     /** unused space at the beginning of buffer or an offset into a
  12.      * file for sendfile buffers. */
  13.     ev_off_t misalign;

  15.     /** Offset into buffer + misalign at which to start writing.
  16.      * In other words, the total number of bytes actually stored
  17.      * in buffer. */
  18.     size_t off;

  20.     /** Set if special handling is required for this chain */
  21.     unsigned flags;
  22.         ...
  23.          /* 实际的数据区 */
  24.     unsigned char *buffer;
  25. };

  27. struct evbuffer {
  28.     /** The first chain in this buffer's linked list of chains. */
  29.     /* 指向第一个chain */
  30.     struct evbuffer_chain *first;
  31.     /** The last chain in this buffer's linked list of chains. */
  32.     struct evbuffer_chain *last;

  34.     /** Pointer to the next pointer pointing at the 'last_with_data' chain.
  35.      *
  36.      * To unpack:
  37.      *
  38.      * The last_with_data chain is the last chain that has any data in it.
  39.      * If all chains in the buffer are empty, it is the first chain.
  40.      * If the buffer has no chains, it is NULL.
  41.      *
  42.      * The last_with_datap pointer points at _whatever 'next' pointer_
  43.      * points at the last_with_datap chain. If the last_with_data chain
  44.      * is the first chain, or it is NULL, then the last_with_datap pointer
  45.      * is &buf->first.
  46.      */
  47.      /* 指向最后一个 */
  48.     struct evbuffer_chain **last_with_datap;

  50.     /** Total amount of bytes stored in all chains.*/
  51.     size_t total_len;

  53.     /* 最后一次调用只有添加的字节和删除的字节 */
  54.     /** Number of bytes we have added to the buffer since we last tried to
  55.      * invoke callbacks. */
  56.     size_t n_add_for_cb;
  57.     /** Number of bytes we have removed from the buffer since we last
  58.      * tried to invoke callbacks. */
  59.     size_t n_del_for_cb;
  60.     ...
  61.     /* 延迟队列 */
  62.     /** Used to implement deferred callbacks. */
  63.     struct deferred_cb_queue *cb_queue;

  65.     /** A reference count on this evbuffer.     When the reference count
  66.      * reaches 0, the buffer is destroyed.    Manipulated with
  67.      * evbuffer_incref and evbuffer_decref_and_unlock and
  68.      * evbuffer_free. */
  69.     int refcnt;

  71.     /** A deferred_cb handle to make all of this buffer's callbacks
  72.      * invoked from the event loop. */
  73.      /* 延迟回调块 */
  74.     struct deferred_cb deferred;

  76.     /* 延迟队列回调函数链表 */
  77.     /** A doubly-linked-list of callback functions */
  78.     TAILQ_HEAD(evbuffer_cb_queue, evbuffer_cb_entry) callbacks;

  80.     /** The parent bufferevent object this evbuffer belongs to.
  81.      * NULL if the evbuffer stands alone. */
  82.      /* 归属的bufferevent */
  83.     struct bufferevent *parent;
  84. };

  86. /* buffer事件 */
  87. struct bufferevent {
  88.     /** Event base for which this bufferevent was created. */
  89.     struct event_base *ev_base;
  90.     /** Pointer to a table of function pointers to set up how this
  91.      bufferevent behaves. */
  92.      /* 设置bufferevent的行为的操作函数 */
  93.     const struct bufferevent_ops *be_ops;

  95.     /** A read event that triggers when a timeout has happened or a socket
  96.      is ready to read data. Only used by some subtypes of
  97.      bufferevent. */
  98.      /* 读事件 */
  99.     struct event ev_read;
  100.     /** A write event that triggers when a timeout has happened or a socket
  101.      is ready to write data. Only used by some subtypes of
  102.      bufferevent. */
  103.      /* 写事件 */
  104.     struct event ev_write;

  106.     /** An input buffer. Only the bufferevent is allowed to add data to
  107.      this buffer, though the user is allowed to drain it. */
  108.      /* 对应的接收缓存 */
  109.     struct evbuffer *input;

  111.     /** An input buffer. Only the bufferevent is allowed to drain data
  112.      from this buffer, though the user is allowed to add it. */
  113.      /* 发送缓存 */
  114.     struct evbuffer *output;

  116.     /* 用于读写限速控制的结构体 */
  117.     struct event_watermark wm_read;
  118.     struct event_watermark wm_write;

  120.     /* 读回调函数 */
  121.     bufferevent_data_cb readcb;
  122.     /* 写回调函数 */
  123.     bufferevent_data_cb writecb;
  124.     /* This should be called 'eventcb', but renaming it would break
  125.     void *cbarg;
  126.     ...
  127. }
代码实现过程:
接下来主要是fd与bufferevent的绑定过程:

点击(此处)折叠或打开

  1. /* 针对fd的bufferevent */
  2. struct bufferevent *
  3. bufferevent_socket_new(struct event_base *base, evutil_socket_t fd,
  4.     int options)
  5. {
  6.     struct bufferevent_private *bufev_p;
  7.     struct bufferevent *bufev;

  9.     if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private)))== NULL)
  10.         return NULL;

  12.     if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket,
  13.                  options) < 0) {
  14.         mm_free(bufev_p);
  15.         return NULL;
  16.     }
  17.     bufev = &bufev_p->bev;
  18.     evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);

  20.     /* 添加基于fd的读事件，对应的回调函数为bufferevent_readcb，参数为bufferent */
  21.     event_assign(&bufev->ev_read, bufev->ev_base, fd,
  22.      EV_READ|EV_PERSIST, bufferevent_readcb, bufev);
  23.     /* 添加基于fd的写事件，对应的回调函数为bufferevent_writecb，参数为bufferent */
  24.     event_assign(&bufev->ev_write, bufev->ev_base, fd,
  25.      EV_WRITE|EV_PERSIST, bufferevent_writecb, bufev);

  27.     /* 添加evbuffer的回调函数为bufferevent_socket_outbuf_cb，主要针对发送数据的 */
  28.     evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);

  30.     /* 允许接收数据 */
  31.     evbuffer_freeze(bufev->input, 0);
  32.     /* 冻结发送数据的缓存区 */
  33.     evbuffer_freeze(bufev->output, 1);

  35.     return bufev;
  36. }
通用bufferevent初始化函数，主要完成了缓存管理器的创建，也就是接收缓存管理器和发送缓存管理器两个缓存的创建和bufferevent的绑定关系，该初始化是通用接口，不止是针对socket有效。

点击(此处)折叠或打开

  1. /* bufferevent的初始化接口，不同的bufferevent通用的接口 */
  2. int
  3. bufferevent_init_common(struct bufferevent_private *bufev_private,
  4.     struct event_base *base,
  5.     const struct bufferevent_ops *ops,
  6.     enum bufferevent_options options)
  7. {
  8.     struct bufferevent *bufev = &bufev_private->bev;
  9.     
  10.     /* 创建接收报文缓冲管理器 */
  11.     if (!bufev->input) {
  12.         if ((bufev->input = evbuffer_new()) == NULL)
  13.             return -1;
  14.     }
  15.     
  16.     /* 创建接收发送缓冲管理器 */
  17.     if (!bufev->output) {
  18.         if ((bufev->output = evbuffer_new()) == NULL) {
  19.             evbuffer_free(bufev->input);
  20.             return -1;
  21.         }
  22.     }

  24.     bufev_private->refcnt = 1;
  25.     bufev->ev_base = base;

  27.     /* Disable timeouts. */
  28.     evutil_timerclear(&bufev->timeout_read);
  29.     evutil_timerclear(&bufev->timeout_write);

  31.     bufev->be_ops = ops;

  33.     /*
  34.      * Set to EV_WRITE so that using bufferevent_write is going to
  35.      * trigger a callback. Reading needs to be explicitly enabled
  36.      * because otherwise no data will be available.
  37.      */
  38.     bufev->enabled = EV_WRITE;

  40. #ifndef _EVENT_DISABLE_THREAD_SUPPORT
  41.     if (options & BEV_OPT_THREADSAFE) {
  42.         if (bufferevent_enable_locking(bufev, NULL) < 0) {
  43.             /* cleanup */
  44.             evbuffer_free(bufev->input);
  45.             evbuffer_free(bufev->output);
  46.             bufev->input = NULL;
  47.             bufev->output = NULL;
  48.             return -1;
  49.         }
  50.     }
  51. #endif
  52.     if ((options & (BEV_OPT_DEFER_CALLBACKS|BEV_OPT_UNLOCK_CALLBACKS))
  53.      == BEV_OPT_UNLOCK_CALLBACKS) {
  54.         event_warnx("UNLOCK_CALLBACKS requires DEFER_CALLBACKS");
  55.         return -1;
  56.     }
  57.     if (options & BEV_OPT_DEFER_CALLBACKS) {
  58.         if (options & BEV_OPT_UNLOCK_CALLBACKS)
  59.             /* 初始化延迟队列的回调函数 */
  60.             event_deferred_cb_init(&bufev_private->deferred,
  61.              bufferevent_run_deferred_callbacks_unlocked,
  62.              bufev_private);
  63.         else
  64.             event_deferred_cb_init(&bufev_private->deferred,
  65.              bufferevent_run_deferred_callbacks_locked,
  66.              bufev_private);
  67.     }

  69.     bufev_private->options = options;

  71.     /* 设置两个缓存区的bufferevent，绑定了bufferent和evbuffer的关系 */
  72.     evbuffer_set_parent(bufev->input, bufev);
  73.     evbuffer_set_parent(bufev->output, bufev);

  75.     return 0;
  76. }
对于读事件的处理函数，主要完成了从socket接收报文，对应的回调函数：bufferevent_readcb(),而写事件的处理函数主要完成发送报文的操纵，对应的回调函数:bufferevent_writecb()。

点击(此处)折叠或打开

  1. /* fd读事件的回调函数，也就是说实际的读操作会调用该函数 */
  2. static void
  3. bufferevent_readcb(evutil_socket_t fd, short event, void *arg)
  4. {
  5.     struct bufferevent *bufev = arg; //间事件注册函数
  6.     struct bufferevent_private *bufev_p =
  7.      EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
  8.     struct evbuffer *input;
  9.     int res = 0;
  10.     short what = BEV_EVENT_READING;
  11.     ev_ssize_t howmuch = -1, readmax=-1;

  13.     _bufferevent_incref_and_lock(bufev);

  15.     if (event == EV_TIMEOUT) {
  16.         what |= BEV_EVENT_TIMEOUT;
  17.         goto error;
  18.     }

  20.     /* 获取缓存区 */
  21.     input = bufev->input;

  23.     /*
  24.      * If we have a high watermark configured then we don't want to
  25.      * read more data than would make us reach the watermark.
  26.      */
  27.     if (bufev->wm_read.high != 0) { /* 如果当前bufferevent设置有门限 */
  28.         howmuch = bufev->wm_read.high - evbuffer_get_length(input);
  29.         /* we somehow lowered the watermark, stop reading */
  30.         if (howmuch <= 0) { /* 暂停读的操作 */
  31.             bufferevent_wm_suspend_read(bufev);
  32.             goto done;
  33.         }
  34.     }

  36.     /* 获取能够读取的最大值 */
  37.     readmax = _bufferevent_get_read_max(bufev_p);
  38.     if (howmuch < 0 || howmuch > readmax) /* The use of -1 for "unlimited"
  39.                      * uglifies this code. XXXX */
  40.         howmuch = readmax;
  41.     /* 如果当前的读操作为暂停，则不读取啦 */
  42.     if (bufev_p->read_suspended)
  43.         goto done;
  44.     
  45.     evbuffer_unfreeze(input, 0);
  46.     /* 实际是完成数据的读写操作 */
  47.     res = evbuffer_read(input, fd, (int)howmuch); /* XXXX evbuffer_read would do better to take and return ev_ssize_t */
  48.     evbuffer_freeze(input, 0);

  50.     if (res == -1) {
  51.         int err = evutil_socket_geterror(fd);
  52.         if (EVUTIL_ERR_RW_RETRIABLE(err))
  53.             goto reschedule;
  54.         /* error case */
  55.         what |= BEV_EVENT_ERROR;
  56.     } else if (res == 0) {
  57.         /* eof case */
  58.         what |= BEV_EVENT_EOF;
  59.     }

  61.     if (res <= 0)
  62.         goto error;

  64.     /* 根据当前的 */
  65.     _bufferevent_decrement_read_buckets(bufev_p, res);

  67.     /* Invoke the user callback - must always be called last */
  68.     /* 当读取的数据量大于最小值以后必须调用使用者注册的回调函数 */
  69.     if (evbuffer_get_length(input) >= bufev->wm_read.low)
  70.         _bufferevent_run_readcb(bufev);

  72.     goto done;

  74.  reschedule:
  75.     goto done;

  77.  error:
  78.     bufferevent_disable(bufev, EV_READ);
  79.     _bufferevent_run_eventcb(bufev, what);

  81.  done:
  82.      /* 该接口会根据引用计数对内存进行删除等操作 */
  83.     _bufferevent_decref_and_unlock(bufev);
  84. }

  86. static void
  87. bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
  88. {
  89.     struct bufferevent *bufev = arg;
  90.     struct bufferevent_private *bufev_p =
  91.      EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
  92.     int res = 0;
  93.     short what = BEV_EVENT_WRITING;
  94.     int connected = 0;
  95.     ev_ssize_t atmost = -1;

  97.     _bufferevent_incref_and_lock(bufev);

  99.     if (event == EV_TIMEOUT) {
  100.         what |= BEV_EVENT_TIMEOUT;
  101.         goto error;
  102.     }

  104.     /* 如果正在进行链接操作 */
  105.     if (bufev_p->connecting) {/* 判断是否正在进行连接处理 */
  106.         int c = evutil_socket_finished_connecting(fd);
  107.         /* we need to fake the error if the connection was refused
  108.          * immediately - usually connection to localhost on BSD */
  109.          /* 如果拒绝连接 */
  110.         if (bufev_p->connection_refused) {
  111.          bufev_p->connection_refused = 0;
  112.          c = -1;
  113.         }

  115.         if (c == 0)
  116.             goto done;

  118.         bufev_p->connecting = 0;
  119.         if (c < 0) { /* 删除数据 */
  120.             event_del(&bufev->ev_write);
  121.             event_del(&bufev->ev_read);
  122.             _bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);
  123.             goto done;
  124.         } else { /* 连接成功或者正在连接 */
  125.             connected = 1;
  126.             _bufferevent_run_eventcb(bufev,
  127.                     BEV_EVENT_CONNECTED);
  128.             if (!(bufev->enabled & EV_WRITE) ||
  129.              bufev_p->write_suspended) {
  130.                 event_del(&bufev->ev_write);
  131.                 goto done;
  132.             }
  133.         }
  134.     }

  136.     /* 获取最大的数据量，也就是当前缓存中最大的报文量 */
  137.     atmost = _bufferevent_get_write_max(bufev_p);

  139.     if (bufev_p->write_suspended)
  140.         goto done;

  142.     /* 有数据需要发送 */
  143.     if (evbuffer_get_length(bufev->output)) {
  144.         evbuffer_unfreeze(bufev->output, 1);
  145.         /* 写数据，该函数实现报文的发送操作，尽最大能力的发送 */
  146.         res = evbuffer_write_atmost(bufev->output, fd, atmost);
  147.         evbuffer_freeze(bufev->output, 1);
  148.         if (res == -1) {
  149.             int err = evutil_socket_geterror(fd);
  150.             if (EVUTIL_ERR_RW_RETRIABLE(err))
  151.                 goto reschedule;
  152.             what |= BEV_EVENT_ERROR;
  153.         } else if (res == 0) {
  154.             /* eof case
  155.              XXXX Actually, a 0 on write doesn't indicate
  156.              an EOF. An ECONNRESET might be more typical.
  157.              */
  158.             what |= BEV_EVENT_EOF;
  159.         }
  160.         if (res <= 0)
  161.             goto error;

  163.         _bufferevent_decrement_write_buckets(bufev_p, res);
  164.     }

  166.     /* 如果没有数据需要再发送，则删除该事件 */
  167.     if (evbuffer_get_length(bufev->output) == 0) {
  168.         event_del(&bufev->ev_write);
  169.     }

  171.     /*
  172.      * Invoke the user callback if our buffer is drained or below the
  173.      * low watermark.
  174.      */
  175.      /* 当缓存再output中的报文量低于某一个值的情况下，通告用户再次进行发送操作 */
  176.     if ((res || !connected) &&
  177.      evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
  178.         _bufferevent_run_writecb(bufev);
  179.     }

  181.     goto done;

  183.  reschedule:
  184.     if (evbuffer_get_length(bufev->output) == 0) {
  185.         event_del(&bufev->ev_write);
  186.     }
  187.     goto done;

  189.  error:
  190.     bufferevent_disable(bufev, EV_WRITE);
  191.     _bufferevent_run_eventcb(bufev, what);

  193.  done: /* bufferevent的控制 */
  194.     _bufferevent_decref_and_unlock(bufev);
  195. }
接收报文的过程中实际上根据当前能够接收的报文量去接收报文，其中input是管理接收报文的缓存管理器。能够接收的报文量由管理器和对应的流控处理确定。实际的读报文函数为evbuffer_read()。

点击(此处)折叠或打开

  1. /* evbuffer的读取操作 */
  2. int
  3. evbuffer_read(struct evbuffer *buf, evutil_socket_t fd, int howmuch)
  4. {
  5.     struct evbuffer_chain **chainp;
  6.     int n;
  7.     int result;

  9. #ifdef USE_IOVEC_IMPL
  10.     int nvecs, i, remaining;
  11. #else
  12.     struct evbuffer_chain *chain;
  13.     unsigned char *p;
  14. #endif

  16.     EVBUFFER_LOCK(buf);

  18.     if (buf->freeze_end) {
  19.         result = -1;
  20.         goto done;
  21.     }
  22.     
  23.     /* 获取当前fd能够读取的最大报文量 */
  24.     n = get_n_bytes_readable_on_socket(fd);
  25.     if (n <= 0 || n > EVBUFFER_MAX_READ)
  26.         n = EVBUFFER_MAX_READ;
  27.     if (howmuch < 0 || howmuch > n) /* 读写长度 */
  28.         howmuch = n;

  30. #ifdef USE_IOVEC_IMPL
  31.     /* Since we can use iovecs, we're willing to use the last
  32.      * NUM_READ_IOVEC chains. */
  33.     if (_evbuffer_expand_fast(buf, howmuch, NUM_READ_IOVEC) == -1) {
  34.         result = -1;
  35.         goto done;
  36.     } else {
  37.         IOV_TYPE vecs[NUM_READ_IOVEC];
  38. #ifdef _EVBUFFER_IOVEC_IS_NATIVE
  39.         /* 实际是完成数据空间的预分配，即vecs空间的分配，4个vecs的空间，chainp是缓冲区的开始地址 */
  40.         nvecs = _evbuffer_read_setup_vecs(buf, howmuch, vecs,
  41.          NUM_READ_IOVEC, &chainp, 1);
  42. #else
  43.         /* We aren't using the native struct iovec. Therefore,
  44.          we are on win32. */
  45.         struct evbuffer_iovec ev_vecs[NUM_READ_IOVEC];
  46.         nvecs = _evbuffer_read_setup_vecs(buf, howmuch, ev_vecs, 2,
  47.          &chainp, 1);

  49.         for (i=0; i < nvecs; ++i)
  50.             WSABUF_FROM_EVBUFFER_IOV(&vecs[i], &ev_vecs[i]);
  51. #endif

  53.         /* 调用readv，采用多缓冲区的读写方式，linux的高级套接字，n是实际返回的长度 */
  54.         n = readv(fd, vecs, nvecs);
  55.     }

  57. #else /*!USE_IOVEC_IMPL*/
  58.     /* If we don't have FIONREAD, we might waste some space here */
  59.     /* XXX we _will_ waste some space here if there is any space left
  60.      * over on buf->last. */
  61.     if ((chain = evbuffer_expand_singlechain(buf, howmuch)) == NULL) {
  62.         result = -1;
  63.         goto done;
  64.     }

  66.     /* We can append new data at this point */
  67.     p = chain->buffer + chain->misalign + chain->off;

  69. #ifndef WIN32
  70.     n = read(fd, p, howmuch);
  71. #else
  72.     /* 常规的报文读取操作 */
  73.     n = recv(fd, p, howmuch, 0);
  74. #endif
  75. #endif /* USE_IOVEC_IMPL */

  77.     if (n == -1) {
  78.         result = -1;
  79.         goto done;
  80.     }
  81.     if (n == 0) {
  82.         result = 0;
  83.         goto done;
  84.     }

  86. #ifdef USE_IOVEC_IMPL
  87.     remaining = n;
  88.     /* nvecs是指多个缓冲区，但是不一定有那么多的数据 */
  89.     for (i=0; i < nvecs; ++i) {
  90.         /* 获取chain的长度 */
  91.         ev_ssize_t space = (ev_ssize_t) CHAIN_SPACE_LEN(*chainp);
  92.         if (space < remaining) { /* 若长度不够 */
  93.             (*chainp)->off += space; /* 则当前chain的内存使用完毕 */
  94.             remaining -= (int)space; /* 剩下的内存空间 */
  95.         } else {
  96.             (*chainp)->off += remaining; /* 当前空间已经足够 */
  97.             buf->last_with_datap = chainp;
  98.             break;
  99.         }
  100.         chainp = &(*chainp)->next;
  101.     }
  102. #else
  103.     chain->off += n;
  104.     advance_last_with_data(buf);
  105. #endif
  106.     /* 更新当前实际的有效长度 */
  107.     buf->total_len += n;
  108.     buf->n_add_for_cb += n;

  110.     /* Tell someone about changes in this buffer */
  111.     evbuffer_invoke_callbacks(buf);
  112.     result = n;
  113. done:
  114.     EVBUFFER_UNLOCK(buf);
  115.     return result;
  116. }
读处理过程中还实现了类似流控的机制，当接受到的报文大于了当前设置的最小门限后，会通告用户进行报文的读取操作：

点击(此处)折叠或打开

  1. void
  2. _bufferevent_run_readcb(struct bufferevent *bufev)
  3. {
  4.     /* Requires that we hold the lock and a reference */
  5.     struct bufferevent_private *p =
  6.      EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
  7.     if (bufev->readcb == NULL)
  8.         return;
  9.     if (p->options & BEV_OPT_DEFER_CALLBACKS) {
  10.         p->readcb_pending = 1;
  11.         if (!p->deferred.queued)
  12.             SCHEDULE_DEFERRED(p);
  13.     } else {
  14.         /* 调用对应的读回调函数，对应的对调函数为bufferevent中注册的用户接口 */
  15.         bufev->readcb(bufev, bufev->cbarg);
  16.     }
  17. }
在发送报文的过程中通过将函数evbuffer_write_atmost()实现，该函数会尽可能的发送对应的报文：

点击(此处)折叠或打开

  1. /* 尽最大可能的发送，通常会采用ivec或者sendfile */
  2. int
  3. evbuffer_write_atmost(struct evbuffer *buffer, evutil_socket_t fd,
  4.     ev_ssize_t howmuch)
  5. {
  6.     int n = -1;

  8.     EVBUFFER_LOCK(buffer);

  10.     if (buffer->freeze_start) {
  11.         goto done;
  12.     }

  14.     /* 发送长度最大为当前buffer的长度 */
  15.     if (howmuch < 0 || (size_t)howmuch > buffer->total_len)
  16.         howmuch = buffer->total_len;

  18.     if (howmuch > 0) {
  19. #ifdef USE_SENDFILE /* 零拷贝的报文发送处理 */
  20.         struct evbuffer_chain *chain = buffer->first;
  21.         if (chain != NULL && (chain->flags & EVBUFFER_SENDFILE))
  22.             n = evbuffer_write_sendfile(buffer, fd, howmuch);
  23.         else {
  24. #endif
  25. #ifdef USE_IOVEC_IMPL
  26.         n = evbuffer_write_iovec(buffer, fd, howmuch);  /* 多缓冲区的数据发送操作，也能提高发送性能 */
  27. #elif defined(WIN32)
  28.         /* XXX(nickm) Don't disable this code until we know if
  29.          * the WSARecv code above works. */
  30.         void *p = evbuffer_pullup(buffer, howmuch);
  31.         n = send(fd, p, howmuch, 0);
  32. #else
  33.         /* 普通的报文发送处理 */
  34.         void *p = evbuffer_pullup(buffer, howmuch);
  35.         n = write(fd, p, howmuch);
  36. #endif
  37. #ifdef USE_SENDFILE
  38.         }
  39. #endif
  40.     }

  42.     if (n > 0) /* 报文发送完成之后将evbuffer丢弃 */
  43.         evbuffer_drain(buffer, n);

  45. done:
  46.     EVBUFFER_UNLOCK(buffer);
  47.     return (n);
  48. }
从上述的过程中可以发现，在发送报文的流程中存在多种方式，其中主要是零拷贝的发送、多缓冲区发送、普通的发送，在报文发送完成之后将对应的缓存丢弃。使用这的报文处理实际上是通过回调的通告方式，当当前缓存中的报文量少于一定量的情况下就会回调注册的写函数进行报文的发送(使用者将报文写到缓存中的过程)。具体的如下所示:

点击(此处)折叠或打开

  1. /* 调用用户的发送回调函数 */
  2. void
  3. _bufferevent_run_writecb(struct bufferevent *bufev)
  4. {
  5.     /* Requires that we hold the lock and a reference */
  6.     struct bufferevent_private *p =
  7.      EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
  8.     if (bufev->writecb == NULL)
  9.         return;
  10.     if (p->options & BEV_OPT_DEFER_CALLBACKS) {
  11.         p->writecb_pending = 1;
  12.         if (!p->deferred.queued)
  13.             SCHEDULE_DEFERRED(p);
  14.     } else {
  15.         /* 调用写的回调函数,在该函数中用户可进行报文的发送处理 */
  16.         bufev->writecb(bufev, bufev->cbarg);
  17.     }
  18. }
以上的过程基本就完成了报文的接收和发送操作。

该系列的分析都是基于libevent2.0.20.

阅读(5572) | 评论(0) | 转发(0) |
0

上一篇：LibEvent代码阅读--服务器端Listen、Accept处理

下一篇：LibEvent代码阅读--多缓冲区和零拷贝技术

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6