c - libuv分配的内存缓冲区重用技术

Question

我正在将libuv用于我与广泛网络交互的应用程序，并且我担心哪种复用复用已分配内存的技术在执行libuv回调延迟的同时将既有效又安全。

在libuv用户可以看到的最基本的一层上，除了设置句柄阅读器外，还需要指定缓冲区分配回调:

UV_EXTERN int uv_read_start(uv_stream_t*, uv_alloc_cb alloc_cb, uv_read_cb read_cb);

uv_alloc_cb在哪里

typedef void (*uv_alloc_cb)(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);

但这是问题所在:每次通过句柄(例如，接收到来自uv_udp_t句柄的每个UDP数据报)发送新消息时，都会调用此分配内存的回调，并且为每个传入的UDP数据报进行新缓冲区的直接分配似乎非常不合理。明智的内存。

因此，我要求一种通用的C技术(可能在libuv回调系统引入的延迟执行上下文内)在可能的情况下重复使用具有相同分配内存的C技术。

另外，如果可能的话，我想保持Windows便携式。

笔记:

我知道这个问题:Does libuv provide any facilities to attach a buffer to a connection and re use it;它被接受的答案除了说明静态分配的缓冲区是不可用的事实之外，没有回答如何使用libuv正确地进行内存分配。特别是，它没有涵盖安全性(在同一个缓冲区上具有延迟的写回调，这可能与libuv主循环的多次迭代中的另一个读取回调调用重叠)附加到句柄的缓冲区(通过包装器结构或handle->数据上下文)。

在阅读http://nikhilm.github.io/uvbook/filesystem.html时，我注意到片段uvtee/main.c - Write to pipe下的以下短语:

We make a copy so we can free the two buffers from the two calls to write_data independently of each other. While acceptable for a demo program like this, you’ll probably want smarter memory management, like reference counted buffers or a pool of buffers in any major application.

但是我找不到任何涉及在libuv缓冲区上进行引用计数的解决方案(如何正确执行？)或libuv环境中的缓冲池的显式示例(是否有用于该库的库？)。

Answer 1

我想分享我自己解决这个问题的经验。我可以感觉到您的痛苦和困惑，但是实际上，考虑到您正在做的事情，可以考虑到您拥有的众多选择，实现一个可行的解决方案并不难。

客观的

实现一个缓冲池，该缓冲池能够执行两个操作-获取和发布。

基本池化策略:

获取从池中撤出一个缓冲区，从而有效地将可用缓冲区数减少了1；

如果没有可用的缓冲区，则会出现两个选项:

扩展池并返回一个新创建的缓冲区；或

创建并返回一个虚拟缓冲区(如下所述)。

版本将缓冲区返回到池中。

池的大小可以固定或可变。 “变量”表示最初有M个预分配的缓冲区(例如零)，并且池可以根据需要增长到N个。“固定”表示所有缓冲区在创建池时就已预先分配(M = N) 。

实现一个回调，该回调为libuv获取缓冲区。

在任何情况下(内存不足的情况除外)，都不允许无限池增长仍具有池功能。

执行

现在，让我们更详细地说明所有这一切。

池结构:

#define BUFPOOL_CAPACITY 100

typedef struct bufpool_s bufpool_t;

struct bufpool_s {
    void *bufs[BUFPOOL_CAPACITY];
    int size;
};

size是当前池的大小。

缓冲区本身是一个具有以下结构的存储块:

#define bufbase(ptr) ((bufbase_t *)((char *)(ptr) - sizeof(bufbase_t)))
#define buflen(ptr) (bufbase(ptr)->len)

typedef struct bufbase_s bufbase_t;

struct bufbase_s {
    bufpool_t *pool;
    int len;
};

len是缓冲区的长度(以字节为单位)。

新缓冲区的分配如下所示:

void *bufpool_alloc(bufpool_t *pool, int len) {
    bufbase_t *base = malloc(sizeof(bufbase_t) + len);
    if (!base) return 0;
    base->pool = pool;
    base->len = len;
    return (char *)base + sizeof(bufbase_t);
}

请注意，返回的指针指向 header 之后的下一个字节-数据区域。这允许具有缓冲区指针，就好像它们是通过对malloc的标准调用分配的一样。

解除分配则相反:

void bufpool_free(void *ptr) {
    if (!ptr) return;
    free(bufbase(ptr));
}

libuv的分配回调如下所示:

void alloc_cb(uv_handle_t *handle, size_t size, uv_buf_t *buf) {
    int len;
    void *ptr = bufpool_acquire(handle->loop->data, &len);
    *buf = uv_buf_init(ptr, len);
}

您可以在此处看到alloc_cb从循环上的用户数据指针获取缓冲池的指针。这意味着在使用缓冲池之前，应将其附加到事件循环。换句话说，您应该在创建循环时初始化池，并将其指针分配给data字段。如果您已经在该字段中保存了其他用户数据，则只需扩展您的结构即可。

虚拟缓冲区是伪缓冲区，这意味着它不是起源于池中，但仍具有完整的功能。虚拟缓冲区的目的是使整个事情在池饥饿的罕见情况下(即当所有缓冲区都已获得并且需要另一个缓冲区时)工作。根据我的研究，在所有现代OS上，非常快速地分配了大约8Kb的小内存块-非常适合虚拟缓冲区的大小。

#define DUMMY_BUF_SIZE 8000

void *bufpool_dummy() {
    return bufpool_alloc(0, DUMMY_BUF_SIZE);
}

获取操作:

void *bufpool_acquire(bufpool_t *pool, int *len) {
    void *buf = bufpool_dequeue(pool);
    if (!buf) buf = bufpool_dummy();
    *len = buf ? buflen(buf) : 0;
    return buf;
}

释放操作:

void bufpool_release(void *ptr) {
    bufbase_t *base;
    if (!ptr) return;
    base = bufbase(ptr);
    if (base->pool) bufpool_enqueue(base->pool, ptr);
    else free(base);
}

此处有两个功能-bufpool_enqueue和bufpool_dequeue。基本上，它们执行池的所有工作。

在我的情况下，上述缓冲区之上有一个O(1)缓冲区索引队列，这使我可以更有效地跟踪池的状态，从而非常快速地获取缓冲区的索引。不必像我一样走极端，因为池的最大大小是有限的，因此任何数组搜索在时间上也将是恒定的。

在最简单的情况下，您可以在bufs结构中的整个bufpool_s数组中将这些函数实现为纯线性搜索器。例如，如果获取缓冲区，则搜索第一个非NULL点，保存指针并将NULL放入该点。下次释放缓冲区时，您将搜索第一个NULL点并将其指针保存在其中。

池内部结构如下:

#define BUF_SIZE 64000

void *bufpool_grow(bufpool_t *pool) {
    int idx = pool->size;
    void *buf;
    if (idx == BUFPOOL_CAPACITY) return 0;
    buf = bufpool_alloc(pool, BUF_SIZE);
    if (!buf) return 0;
    pool->bufs[idx] = 0;
    pool->size = idx + 1;
    return buf;
}

void bufpool_enqueue(bufpool_t *pool, void *ptr) {
    int idx;
    for (idx = 0; idx < pool->size; ++idx) {
        if (!pool->bufs[idx]) break;
    }
    assert(idx < pool->size);
    pool->bufs[idx] = ptr;
}

void *bufpool_dequeue(bufpool_t *pool) {
    int idx;
    void *ptr;
    for (idx = 0; idx < pool->size; ++idx) {
        ptr = pool->bufs[idx];
        if (ptr) {
            pool->bufs[idx] = 0;
            return ptr;
        }
    }
    return bufpool_grow(pool);
}

正常的缓冲区大小为64000字节，因为我希望它可以舒适地容纳在带有其 header 的64Kb块中。

最后，初始化和取消初始化例程:

void bufpool_init(bufpool_t *pool) {
    pool->size = 0;
}

void bufpool_done(bufpool_t *pool) {
    int idx;
    for (idx = 0; idx < pool->size; ++idx) bufpool_free(pool->bufs[idx]);
}

请注意，出于说明目的，此实现已简化。这里没有池缩减策略，而在现实世界中，很有可能需要这样做。

用法

您现在应该可以编写您的libuv回调:

void read_cb(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
    /* ... */
    bufpool_release(buf->base); /* Release the buffer */
}

循环初始化:

uv_loop_t *loop = malloc(sizeof(*loop));
bufpool_t *pool = malloc(sizeof(*pool));
uv_loop_init(loop);
bufpool_init(pool);
loop->data = pool;

手术:

uv_tcp_t *tcp = malloc(sizeof(*tcp));
uv_tcp_init(tcp);
/* ... */
uv_read_start((uv_handle_t *)tcp, alloc_cb, read_cb);

更新(2016年8月2日)

在根据请求的大小获取缓冲区时仅使用自适应策略也是一个好主意，并且仅在请求大量数据(例如，所有读取和长时间写入)时才返回缓冲的缓冲区。对于其他情况(例如大多数写入)，请返回虚拟缓冲区。这将有助于避免浪费缓冲池，同时保持可接受的分配速度。例如:

void alloc_cb(uv_handle_t *handle, size_t size, uv_buf_t *buf) {
    int len = size; /* Requested buffer size */
    void *ptr = bufpool_acquire(handle->loop->data, &len);
    *buf = uv_buf_init(ptr, len);
}

void *bufpool_acquire(bufpool_t *pool, int *len) {
    int size = *len;
    if (size > DUMMY_BUF_SIZE) {
        buf = bufpool_dequeue(pool);
        if (buf) {
            if (size > BUF_SIZE) *len = BUF_SIZE;
            return buf;
        }
        size = DUMMY_BUF_SIZE;
    }
    buf = bufpool_alloc(0, size);
    *len = buf ? size : 0;
    return buf;
}

P.S.无需使用此代码段的buflen和bufpool_dummy。