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

最近项目赶着发版本，连续好几天通宵赶工，瓶颈出现在了性能问题，各种方案验证和修订。周末闲下来还是对之前学习的libevent进行总结。
在libevent的报文收发处理过程中采用了一系列提高收发性能的技术，其中多缓冲区的接收和发送以及零拷贝技术等，本篇主要分析这些技术在libevent中的运用。

首先简要的介绍一下两种技术:
多缓存的收据发送和接收:

struct iovec {
    ptr_t iov_base; /* Starting address */
    size_t iov_len; /* Length in bytes */
};

int readv(int fd, const struct iovec *vector, int count);
int writev(int fd, const struct iovec *vector, int count);

其中的iovec是指一个缓冲区，包含了数据区的地址和对应的长度，在两个函数中的count是指iovec的个数。这种多缓冲区的发送和接收操作性能相对更好。而iovec需要在使用前分配好相关的内存空间。

零拷贝技术：
零拷贝能够减少数据之间无效的数据拷贝，而直接进行数据的发送，通常在数据发送的过程中使用,特别是在文件发送的过程中被经常使用。通常情况下要将文件a.txt中的内容发送出去，需要进行如下的操作:
读取文件内容: read(),然后发送读取的内容send()。因此一个完整的过程会出现一个读取再发送的操作，往往文件的IO操作是相对费时的操作，因此零拷贝技术实际上就是较少了read()的处理过程，即在发送数据前不需要进行文件的读取操作，这样相对而言就会提高处理的性能。关于零拷贝的技术有很多方式，这里主要介绍sendfile和mmap.
其中的mmap是采用映射的方式将文件内容映射到内存中，在发送报文时直接读取内存中的内容，这样就能提高发送效率。
sendfile则是直接将读取到文件fd的内容发送到输出的fd中，也不需要文件读取的过程，性能也会提高。
以上两种处理方式实际上都是内核协助完成。

void *mmap(void *start,size_t length,int prot,int flags,int fd,off_t offsize）;

ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

Libevent中多缓存的分析
关于多缓冲区的数据发送和接收主要是在报文的接收和发送过程中使用，多缓冲区减少了调用send和recv的次数。

/* 如果支持多缓冲区的写操作 */
int
evbuffer_write_iovec(struct evbuffer *buffer, evutil_socket_t fd,
    ev_ssize_t howmuch)
{
    IOV_TYPE iov[NUM_WRITE_IOVEC];
    struct evbuffer_chain *chain = buffer->first;
    int n, i = 0;

    if (howmuch < 0)
        return -1;

    ASSERT_EVBUFFER_LOCKED(buffer);
    /* XXX make this top out at some maximal data length? if the
     * buffer has (say) 1MB in it, split over 128 chains, there's
     * no way it all gets written in one go. */
     /* 从evbuffer中将对应的数据拷贝出来 */
    while (chain != NULL && i < NUM_WRITE_IOVEC && howmuch) {
#ifdef USE_SENDFILE
        /* we cannot write the file info via writev */
        if (chain->flags & EVBUFFER_SENDFILE)
            break;
#endif
        /*iov[i].iov_base XXX的含义*/
        iov[i].IOV_PTR_FIELD = (void *) (chain->buffer + chain->misalign);
        if ((size_t)howmuch >= chain->off) {
            /* XXXcould be problematic when windows supports mmap*/
            /*iov[i++].iov_len是指长度*/
            iov[i++].IOV_LEN_FIELD = (IOV_LEN_TYPE)chain->off;
            howmuch -= chain->off;
        } else {
            /* XXXcould be problematic when windows supports mmap*/
            iov[i++].IOV_LEN_FIELD = (IOV_LEN_TYPE)howmuch;
            break;
        }
        chain = chain->next;
    }
    if (! i)
        return 0;
    /* 采用多缓冲区发送，因此fd应该是发送的接口 */
    n = writev(fd, iov, i);

    return (n);
}

接收报文的的处理过程:

int
evbuffer_read(struct evbuffer *buf, evutil_socket_t fd, int howmuch)
{
    struct evbuffer_chain **chainp;
    int n;
    int result;

    int nvecs, i, remaining;

    EVBUFFER_LOCK(buf);

    if (buf->freeze_end) {
        result = -1;
        goto done;
    }

    n = get_n_bytes_readable_on_socket(fd);
    if (n <= 0 || n > EVBUFFER_MAX_READ)
        n = EVBUFFER_MAX_READ;
    if (howmuch < 0 || howmuch > n) /* 读写长度 */
        howmuch = n;

    /* Since we can use iovecs, we're willing to use the last
     * NUM_READ_IOVEC chains. */
    if (_evbuffer_expand_fast(buf, howmuch, NUM_READ_IOVEC) == -1) {
        result = -1;
        goto done;
    } else {
        IOV_TYPE vecs[NUM_READ_IOVEC];
        /* 实际是完成数据空间的预分配，即vecs空间的分配，4个vecs的空间，chainp是缓冲区的开始地址 */
        nvecs = _evbuffer_read_setup_vecs(buf, howmuch, vecs,
            NUM_READ_IOVEC, &chainp, 1);

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

    if (n == -1) {
        result = -1;
        goto done;
    }
    if (n == 0) {
        result = 0;
        goto done;
    }

    remaining = n;
    /* nvecs是指多个缓冲区，但是不一定有那么多的数据 */
    for (i=0; i < nvecs; ++i) {
        /* 获取chain的长度 */
        ev_ssize_t space = (ev_ssize_t) CHAIN_SPACE_LEN(*chainp);
        if (space < remaining) { /* 若长度不够 */
            (*chainp)->off += space; /* 则当前chain的内存使用完毕 */
            remaining -= (int)space; /* 剩下的内存空间 */
        } else {
            (*chainp)->off += remaining; /* 当前空间已经足够 */
            buf->last_with_datap = chainp;
            break;
        }
        chainp = &(*chainp)->next;
    }

    /* 更新当前实际的有效长度 */
    buf->total_len += n;
    buf->n_add_for_cb += n;

    /* Tell someone about changes in this buffer */
    evbuffer_invoke_callbacks(buf);
    result = n;
done:
    EVBUFFER_UNLOCK(buf);
    return result;
}

设置缓存队列的过程如下所示:

int
_evbuffer_read_setup_vecs(struct evbuffer *buf, ev_ssize_t howmuch,
    struct evbuffer_iovec *vecs, int n_vecs_avail,
    struct evbuffer_chain ***chainp, int exact)
{
    struct evbuffer_chain *chain;
    struct evbuffer_chain **firstchainp;
    size_t so_far;
    int i;
    ASSERT_EVBUFFER_LOCKED(buf);

    if (howmuch < 0)
        return -1;

    so_far = 0;
    /* Let firstchain be the first chain with any space on it */
    /* 从当前有数据的位置保存 */
    firstchainp = buf->last_with_datap;
    if (CHAIN_SPACE_LEN(*firstchainp) == 0) {
        firstchainp = &(*firstchainp)->next;
    }

    chain = *firstchainp;
    for (i = 0; i < n_vecs_avail && so_far < (size_t)howmuch; ++i) {
        /* 获取当前chain可用的内存大小 */
        size_t avail = (size_t) CHAIN_SPACE_LEN(chain);
        if (avail > (howmuch - so_far) && exact) /* 当前内存足够存放 */
            avail = howmuch - so_far;
        /* vecs的基地址 */
        vecs[i].iov_base = CHAIN_SPACE_PTR(chain);
        vecs[i].iov_len = avail;
        so_far += avail;

        chain = chain->next;
    }

    *chainp = firstchainp;
    return i;
}

在接收的过程中只需要设置好对应的缓存区大小以及对应的缓存地址，然后调用readv进行报文的接收。

Libevent中零拷贝的实现
关于零拷贝技术在libevent中主要体现在用于文件发送的过程中，提供了发送文件内容的接口:

int
evbuffer_add_file(struct evbuffer *outbuf, int fd,
    ev_off_t offset, ev_off_t length)

其中outbuf主要用于输出的缓存，fd是指需要被发送的文件描述符，offset是文件的偏移量，length是指需要发送的长度，其中outbuf是在bufferevent中的output。

int
evbuffer_add_file(struct evbuffer *outbuf, int fd,
    ev_off_t offset, ev_off_t length)
{
#if defined(USE_SENDFILE) || defined(_EVENT_HAVE_MMAP)
    struct evbuffer_chain *chain;
    struct evbuffer_chain_fd *info;
#endif
#if defined(USE_SENDFILE)
    int sendfile_okay = 1;
#endif
    int ok = 1;

#if defined(USE_SENDFILE)
    if (use_sendfile) {
        EVBUFFER_LOCK(outbuf);
        sendfile_okay = outbuf->flags & EVBUFFER_FLAG_DRAINS_TO_FD;
        EVBUFFER_UNLOCK(outbuf);
    }

    if (use_sendfile && sendfile_okay) {
        /* 1K大小，分配一个chain */
        chain = evbuffer_chain_new(sizeof(struct evbuffer_chain_fd));
        if (chain == NULL) {
            event_warn("%s: out of memory", __func__);
            return (-1);
        }

     /* 设置chain的属性 */
        chain->flags |= EVBUFFER_SENDFILE | EVBUFFER_IMMUTABLE;
        chain->buffer = NULL;    /* no reading possible，在sendfile中不需要buffer */
        chain->buffer_len = length + offset;
        chain->off = length;
        chain->misalign = offset;

        info = EVBUFFER_CHAIN_EXTRA(struct evbuffer_chain_fd, chain);
        /* 避免拷贝数据，保存的是文件的fd，后面获取该fd进行发送操作 */
        info->fd = fd;

        EVBUFFER_LOCK(outbuf);
        if (outbuf->freeze_end) {
            mm_free(chain);
            ok = 0;
        } else {
            /* 添加到output吧 */
            outbuf->n_add_for_cb += length;
            evbuffer_chain_insert(outbuf, chain);
        }
    } else
#endif
#if defined(_EVENT_HAVE_MMAP)
    if (use_mmap) {
        /* 内存映射，将文件映射到内存中,采用mmap的方式减少内存拷贝 */
        void *mapped = mmap(NULL, length + offset, PROT_READ,
#ifdef MAP_NOCACHE
            MAP_NOCACHE |
#endif
#ifdef MAP_FILE
            MAP_FILE |
#endif
            MAP_PRIVATE,
            fd, 0);
        /* some mmap implementations require offset to be a multiple of
         * the page size. most users of this api, are likely to use 0
         * so mapping everything is not likely to be a problem.
         * TODO(niels): determine page size and round offset to that
         * page size to avoid mapping too much memory.
         */
        if (mapped == MAP_FAILED) {
            event_warn("%s: mmap(%d, %d, %zu) failed",
                __func__, fd, 0, (size_t)(offset + length));
            return (-1);
        }

        /* 将需要处理的报文压缩为一个chain */
        chain = evbuffer_chain_new(sizeof(struct evbuffer_chain_fd));
        if (chain == NULL) {
            event_warn("%s: out of memory", __func__);
            munmap(mapped, length);
            return (-1);
        }

        chain->flags |= EVBUFFER_MMAP | EVBUFFER_IMMUTABLE;
        chain->buffer = mapped;  //映射后的内存地址，发送报文时使用该地址
        chain->buffer_len = length + offset;
        chain->off = length + offset;

        info = EVBUFFER_CHAIN_EXTRA(struct evbuffer_chain_fd, chain);
        /* 需要处理的文件，实际上将需要拷贝的文件 */
        info->fd = fd;

        EVBUFFER_LOCK(outbuf);
        if (outbuf->freeze_end) {
            info->fd = -1;
            evbuffer_chain_free(chain);
            ok = 0;
        } else {
            outbuf->n_add_for_cb += length;

            evbuffer_chain_insert(outbuf, chain);

            /* we need to subtract whatever we don't need */
            evbuffer_drain(outbuf, offset);
        }
    } else /* 在以上几种技术都不支持的情况下，采用传统的先拷贝，然后再发送的方式 */
#endif
    { /* 普通的处理方式，先拷贝再发送 */
        /* the default implementation */
        struct evbuffer *tmp = evbuffer_new();
        ev_ssize_t read;

        if (tmp == NULL)
            return (-1);

        if (lseek(fd, offset, SEEK_SET) == -1) {
            evbuffer_free(tmp);
            return (-1);
        }

        /* we add everything to a temporary buffer, so that we
         * can abort without side effects if the read fails.
         */
        while (length) { /* 实际是将数据拷贝到evbuffer中 */
            read = evbuffer_readfile(tmp, fd, (ev_ssize_t)length);
            if (read == -1) {
                evbuffer_free(tmp);
                return (-1);
            }

            length -= read;
        }

        EVBUFFER_LOCK(outbuf);
        if (outbuf->freeze_end) {
            evbuffer_free(tmp);
            ok = 0;
        } else {
         /* 将读出来的buffer添加到输出的buffer中 */
            evbuffer_add_buffer(outbuf, tmp);
            evbuffer_free(tmp);

            /* 关闭文件操作 */
            close(fd);
        }
    }

    if (ok)
        evbuffer_invoke_callbacks(outbuf);
    EVBUFFER_UNLOCK(outbuf);

    return ok ? 0 : -1;
}

从上面的代码中可知，在支持sendfile的系统中采用了sendfile的方式，在支持mmap的系统中采用mmap的方式，而都不支持则直接进行普通的处理，先拷贝再发送。

在上述不同的三种处理中，都是创建新的buffer_chain，新的内存块，然后插入到outbuf中，但是新的内存块中的零拷贝技术的内容不一样。普通情况下都是将对应的报文内容填充到一系列的buffer_chain。而在零拷贝中chain中保存的是evbuffer_chain_fd，该结构体中实际只是包含了文件的fd。但mmap中还保存了映射内存的起始地址和长度。而sendfile中则无数据相关的处理。
关于sendfile的具体发送操作如下所示:

/* 如果支持sendfile，一种零拷贝的技术 */
#ifdef USE_SENDFILE
static inline int
evbuffer_write_sendfile(struct evbuffer *buffer, evutil_socket_t fd,
    ev_ssize_t howmuch)
{
    struct evbuffer_chain *chain = buffer->first;
    /* 采用fd的方式保存在chain的头部之后 */
    struct evbuffer_chain_fd *info =
        EVBUFFER_CHAIN_EXTRA(struct evbuffer_chain_fd, chain);

    ev_ssize_t res;
    off_t offset = chain->misalign;

    ASSERT_EVBUFFER_LOCKED(buffer);
    /* TODO(niels): implement splice */
    /* 发送数据到fd,也就是将文件中的内容发送出去 */
    res = sendfile(fd, info->fd, &offset, chain->off);
    if (res == -1 && EVUTIL_ERR_RW_RETRIABLE(errno)) {
        /* if this is EAGAIN or EINTR return 0; otherwise, -1 */
        return (0);
    }
    return (res);
}

上述的操作避免的文件到内存的拷贝，能够提高数据发送的效率。

而关于mmap的处理，加载内存之后还是需要按照常规的发送进行处理，只是减少了从文件到内存的拷贝过程。