深圳做网站哪个公司好,zencart网站管理,360搜索网站提交入口,网站开发模版下载一、Linux下的网络通信模型
在网络开发的过程中#xff0c;需要处理好几个问题。首先是通信的内核支持问题#xff1b;其次是通信的模型问题#xff1b;最后是框架问题。这些问题在闭源的OS如Windows上#xff0c;基本上不算什么大问题#xff08;因为只能用人家的API需要处理好几个问题。首先是通信的内核支持问题其次是通信的模型问题最后是框架问题。这些问题在闭源的OS如Windows上基本上不算什么大问题因为只能用人家的API。但在开源的OS上典型的就是Linux上就是一个重要的问题。 在开源的系统上如果内核原生支持一种IO通信那么效率一般来说会比不支持要高很多。网络通信有一个问题它有点和算法在本质上有些相通即选用哪种模型会导致它的通信效率差距非常大。举一个不恰当的例子用一个只有两个槽的插座和十个槽的插座能同时支持的用电接口个数完全不同。而在前面两个相同而应用上由于不同人对整个应用的理解和设计架构不同导致其对应用支持的效率又有不同。同时可能一种网络通信模型无法满足一些应用场景所以框架一般会封装多种网络通信模型这也是网络通信框架存在的意义。 但是有的时候模型和框架往往混在一起所以同一种技术在网络和书籍上的叫法均有不同大家自己明白怎么回事儿即可不要在这些细节上纠结。 看过Redis源的都清楚在Redis中对类linux的网络通信模型支持的有select,poll,epoll,kqueue。当然可能不同的版本有所不同这些细节就不再讨论。 一般来说网络通信属于IO通信而在IO通信中常见的几种通信模型有 1、阻塞IOBlocking IO 2、非阻塞IO(NonBlocking IO) 3、IO多路复用IO Multiplexing 4、信号驱动Signal Driven IO 5、异步IOAsynchonous IO 这几种模型在《Unix网络编程》中都有总结。需要说明的是这是从模型角度来抽象出来的所以在其实现的过程中是结合了内核支持和模型特点及应用支持几方面的来说明的。以前一般认为在Linux中对异步支持不如Windows好。 但开源技术有一个好处缺啥补啥。所以在2019年的Linux内核5.1版本中出现了一种新的IO框架也就是io_uring。它是由block IO维护者Jens Axboe开发的一个用来支持异步IO通信的框架。
二、IO框架uring
io_uring为什么叫IO框架因为其一开始主要是给存储搞的。网上的测评那是和SPDK可以一较高低如果真那样io_uring可比SPDK要强大因为intel毕竟是一家国际大公司再加从底层直接支持。 但是网络也属于IO啊所以后来也就支持了网络IO。刚刚提到了在Linux原生支持的IO通信中大多数都是同步的即使有一些框架实现了异步非阻塞仍然做的不是很好AIO在不少的情况下仍然会出现非异步的情况。但是io_uring的出现可以说是一种革命性的突破它和eBPF是内核的两个重大的技术里程碑。 一个优秀的异步IO框架一般具有以下几个特点 1、异步非阻塞 异步和非阻塞往往是和IO通信分不开的而异步和非阻塞又往往出现在一起。在IO通信中非阻塞是实现异步的一个前提。而异步非阻塞往往是一种IO通信框架是否高效的基础。 2、减少或删除中间环节 容易理解的是同步阻塞通信因为等待的原因对中间环节通信事件通知、数据拷贝次数、系统调用次数等敏感性不强。但是异步非阻塞往往意味着海量的IO操作所以中间环节增加任何一个步骤付出的代价往往是难以忍受的。这也是AIO被诟病的地方。 3、扩展性要强 一个IO框架不能只适应一种IO情况。比如磁盘、网络等都可以支持。 4、应用方便快捷 这是一个好的框架必备的特点。再优秀的东西一旦变得复杂就会慢慢失去它的优秀性。复杂就意味着高成本和后期维护的困难。 从Linux的异步框架发展来看在io_uring之前出现过一些异步框架但是对上述问题的解决都不是太好所以一直无法融入内核之中直到io_ring的出现。 而io_uring基本实现了上述几个特点。 首先io_uring通过使用共享内存技术大幅减少了内核层和应用层数据的交互拷贝也减少了内核层和用户层的调用次数。其次数据隔离、交互减少就意味着内核和用户层可以自行其事安排自己的线程工作状态时不用等待对方状态结果这也是异步通信的一个重要前提。为了进一步实现异步io_uring还实现了一个无锁环形队列通过操作其实现用户态和内核态的IO非阻塞交互。
通过上述的说明基本可以了解io_uring的工程机制做为一种框架需要说明一下io_uring操作的流程 1、应用程序提交IO操作请求到提交队列 2、SQ内核线程读取IO操作 3、SQ内核线程发起IO请求 4、SQ内核线程将结果写回完成队列 5、应用程序在完成队列读到IO结果
io_uring通过共享内存来减少交互在io_uring中主要有四种队列即 1、提交队列Submission Queue, SQ一整块连续的内存空间存储的环形队列存放将执行 I/O 操作的数据指向SQE数组的索引 2、完成队列Completion Queue, CQ一整块连续的内存空间存储的环形队列存放 I/O 操作完成后的结果 3、提交队列项数组Submission Queue EntrySQE 提交队列中的每项是SQ中的数据项 4、CQE - Completion Queue Entry完成队列项这是储存在CQ中的数据项
io_uring 在创建时有两个选项对应着 io_uring 处理任务的不同方式 1、IORING_SETUP_IOPOLLio_uring 会使用轮询的方式执行所有的操作 2、IORING_SETUP_SQPOLLio_uring 会创建一个内核线程专门用来处理用户提交的任务 这两个选项的设置会对之后的交互产生如下影响 1、都不设置即以io_uring_enter 提交任务内核线程任务无需 syscall 2、设置IORING_SETUP_IOPOLL以io_uring_enter 提交任务和处理任务。 3、设置IORING_SETUP_SQPOLL直接提交处理任务不需要syscall。内核线程自动控制处理并在一定条件下休眠然后可使用io_uring_enter唤醒。
当用户设置了 IORING_SETUP_SQPOLL选项SQPOLL模式创建 io_uring 时内核将会创建一个名为 io_uring-sq 的内核线程即SQ线程SQ线程从提交队列读取IO操作并且发起IO请求。 当IO请求完成以后SQ线程将会把IO操作的结果写回到完成队列这样应用程序就可以从中得到IO操作的结果。
三、应用例程
直接使用io_uring有些复杂开发者在io_uring的基础上又封装了一个库即liburing,这样普通开发者可以直接使用它。看一下其UDP通信的例程
#include stdio.h
#include unistd.h
#include errno.h
#include sys/mman.h
#include stdlib.h
#include string.h
#include netinet/udp.h
#include arpa/inet.h#include liburing.h#define QD 64
#define BUF_SHIFT 12 /* 4k */
#define CQES (QD * 16)
#define BUFFERS CQES
#define CONTROLLEN 0struct sendmsg_ctx {struct msghdr msg;struct iovec iov;
};struct ctx {struct io_uring ring;struct io_uring_buf_ring *buf_ring;unsigned char *buffer_base;struct msghdr msg;int buf_shift;int af;bool verbose;struct sendmsg_ctx send[BUFFERS];size_t buf_ring_size;
};static size_t buffer_size(struct ctx *ctx)
{return 1U ctx-buf_shift;
}static unsigned char *get_buffer(struct ctx *ctx, int idx)
{return ctx-buffer_base (idx ctx-buf_shift);
}static int setup_buffer_pool(struct ctx *ctx)
{int ret, i;void *mapped;struct io_uring_buf_reg reg { .ring_addr 0,.ring_entries BUFFERS,.bgid 0 };ctx-buf_ring_size (sizeof(struct io_uring_buf) buffer_size(ctx)) * BUFFERS;mapped mmap(NULL, ctx-buf_ring_size, PROT_READ | PROT_WRITE,MAP_ANONYMOUS | MAP_PRIVATE, 0, 0);if (mapped MAP_FAILED) {fprintf(stderr, buf_ring mmap: %s\n, strerror(errno));return -1;}ctx-buf_ring (struct io_uring_buf_ring *)mapped;io_uring_buf_ring_init(ctx-buf_ring);reg (struct io_uring_buf_reg) {.ring_addr (unsigned long)ctx-buf_ring,.ring_entries BUFFERS,.bgid 0};ctx-buffer_base (unsigned char *)ctx-buf_ring sizeof(struct io_uring_buf) * BUFFERS;ret io_uring_register_buf_ring(ctx-ring, reg, 0);if (ret) {fprintf(stderr, buf_ring init failed: %s\nNB This requires a kernel version 6.0\n,strerror(-ret));return ret;}for (i 0; i BUFFERS; i) {io_uring_buf_ring_add(ctx-buf_ring, get_buffer(ctx, i), buffer_size(ctx), i,io_uring_buf_ring_mask(BUFFERS), i);}io_uring_buf_ring_advance(ctx-buf_ring, BUFFERS);return 0;
}static int setup_context(struct ctx *ctx)
{struct io_uring_params params;int ret;memset(params, 0, sizeof(params));params.cq_entries QD * 8;params.flags IORING_SETUP_SUBMIT_ALL | IORING_SETUP_COOP_TASKRUN |IORING_SETUP_CQSIZE;ret io_uring_queue_init_params(QD, ctx-ring, params);if (ret 0) {fprintf(stderr, queue_init failed: %s\nNB: This requires a kernel version 6.0\n,strerror(-ret));return ret;}ret setup_buffer_pool(ctx);if (ret)io_uring_queue_exit(ctx-ring);memset(ctx-msg, 0, sizeof(ctx-msg));ctx-msg.msg_namelen sizeof(struct sockaddr_storage);ctx-msg.msg_controllen CONTROLLEN;return ret;
}static int setup_sock(int af, int port)
{int ret;int fd;uint16_t nport port 0 ? 0 : htons(port);fd socket(af, SOCK_DGRAM, 0);if (fd 0) {fprintf(stderr, sock_init: %s\n, strerror(errno));return -1;}if (af AF_INET6) {struct sockaddr_in6 addr6 {.sin6_family af,.sin6_port nport,.sin6_addr IN6ADDR_ANY_INIT};ret bind(fd, (struct sockaddr *) addr6, sizeof(addr6));} else {struct sockaddr_in addr {.sin_family af,.sin_port nport,.sin_addr { INADDR_ANY }};ret bind(fd, (struct sockaddr *) addr, sizeof(addr));}if (ret) {fprintf(stderr, sock_bind: %s\n, strerror(errno));close(fd);return -1;}if (port 0) {int port;struct sockaddr_storage s;socklen_t sz sizeof(s);if (getsockname(fd, (struct sockaddr *)s, sz)) {fprintf(stderr, getsockname failed\n);close(fd);return -1;}port ntohs(((struct sockaddr_in *)s)-sin_port);fprintf(stderr, port bound to %d\n, port);}return fd;
}static void cleanup_context(struct ctx *ctx)
{munmap(ctx-buf_ring, ctx-buf_ring_size);io_uring_queue_exit(ctx-ring);
}static bool get_sqe(struct ctx *ctx, struct io_uring_sqe **sqe)
{*sqe io_uring_get_sqe(ctx-ring);if (!*sqe) {io_uring_submit(ctx-ring);*sqe io_uring_get_sqe(ctx-ring);}if (!*sqe) {fprintf(stderr, cannot get sqe\n);return true;}return false;
}static int add_recv(struct ctx *ctx, int idx)
{struct io_uring_sqe *sqe;if (get_sqe(ctx, sqe))return -1;io_uring_prep_recvmsg_multishot(sqe, idx, ctx-msg, MSG_TRUNC);sqe-flags | IOSQE_FIXED_FILE;sqe-flags | IOSQE_BUFFER_SELECT;sqe-buf_group 0;io_uring_sqe_set_data64(sqe, BUFFERS 1);return 0;
}static void recycle_buffer(struct ctx *ctx, int idx)
{io_uring_buf_ring_add(ctx-buf_ring, get_buffer(ctx, idx), buffer_size(ctx), idx,io_uring_buf_ring_mask(BUFFERS), 0);io_uring_buf_ring_advance(ctx-buf_ring, 1);
}static int process_cqe_send(struct ctx *ctx, struct io_uring_cqe *cqe)
{int idx cqe-user_data;if (cqe-res 0)fprintf(stderr, bad send %s\n, strerror(-cqe-res));recycle_buffer(ctx, idx);return 0;
}static int process_cqe_recv(struct ctx *ctx, struct io_uring_cqe *cqe,int fdidx)
{int ret, idx;struct io_uring_recvmsg_out *o;struct io_uring_sqe *sqe;if (!(cqe-flags IORING_CQE_F_MORE)) {ret add_recv(ctx, fdidx);if (ret)return ret;}if (cqe-res -ENOBUFS)return 0;if (!(cqe-flags IORING_CQE_F_BUFFER) || cqe-res 0) {fprintf(stderr, recv cqe bad res %d\n, cqe-res);if (cqe-res -EFAULT || cqe-res -EINVAL)fprintf(stderr,NB: This requires a kernel version 6.0\n);return -1;}idx cqe-flags 16;o io_uring_recvmsg_validate(get_buffer(ctx, cqe-flags 16),cqe-res, ctx-msg);if (!o) {fprintf(stderr, bad recvmsg\n);return -1;}if (o-namelen ctx-msg.msg_namelen) {fprintf(stderr, truncated name\n);recycle_buffer(ctx, idx);return 0;}if (o-flags MSG_TRUNC) {unsigned int r;r io_uring_recvmsg_payload_length(o, cqe-res, ctx-msg);fprintf(stderr, truncated msg need %u received %u\n,o-payloadlen, r);recycle_buffer(ctx, idx);return 0;}if (ctx-verbose) {struct sockaddr_in *addr io_uring_recvmsg_name(o);struct sockaddr_in6 *addr6 (void *)addr;char buff[INET6_ADDRSTRLEN 1];const char *name;void *paddr;if (ctx-af AF_INET6)paddr addr6-sin6_addr;elsepaddr addr-sin_addr;name inet_ntop(ctx-af, paddr, buff, sizeof(buff));if (!name)name INVALID;fprintf(stderr, received %u bytes %d from [%s]:%d\n,io_uring_recvmsg_payload_length(o, cqe-res, ctx-msg),o-namelen, name, (int)ntohs(addr-sin_port));}if (get_sqe(ctx, sqe))return -1;ctx-send[idx].iov (struct iovec) {.iov_base io_uring_recvmsg_payload(o, ctx-msg),.iov_len io_uring_recvmsg_payload_length(o, cqe-res, ctx-msg)};ctx-send[idx].msg (struct msghdr) {.msg_namelen o-namelen,.msg_name io_uring_recvmsg_name(o),.msg_control NULL,.msg_controllen 0,.msg_iov ctx-send[idx].iov,.msg_iovlen 1};io_uring_prep_sendmsg(sqe, fdidx, ctx-send[idx].msg, 0);io_uring_sqe_set_data64(sqe, idx);sqe-flags | IOSQE_FIXED_FILE;return 0;
}
static int process_cqe(struct ctx *ctx, struct io_uring_cqe *cqe, int fdidx)
{if (cqe-user_data BUFFERS)return process_cqe_send(ctx, cqe);elsereturn process_cqe_recv(ctx, cqe, fdidx);
}int main(int argc, char *argv[])
{struct ctx ctx;int ret;int port -1;int sockfd;int opt;struct io_uring_cqe *cqes[CQES];unsigned int count, i;memset(ctx, 0, sizeof(ctx));ctx.verbose false;ctx.af AF_INET;ctx.buf_shift BUF_SHIFT;while ((opt getopt(argc, argv, 6vp:b:)) ! -1) {switch (opt) {case 6:ctx.af AF_INET6;break;case p:port atoi(optarg);break;case b:ctx.buf_shift atoi(optarg);break;case v:ctx.verbose true;break;default:fprintf(stderr, Usage: %s [-p port] [-b log2(BufferSize)] [-6] [-v]\n,argv[0]);exit(-1);}}sockfd setup_sock(ctx.af, port);if (sockfd 0)return 1;if (setup_context(ctx)) {close(sockfd);return 1;}ret io_uring_register_files(ctx.ring, sockfd, 1);if (ret) {fprintf(stderr, register files: %s\n, strerror(-ret));return -1;}ret add_recv(ctx, 0);if (ret)return 1;while (true) {ret io_uring_submit_and_wait(ctx.ring, 1);if (ret -EINTR)continue;if (ret 0) {fprintf(stderr, submit and wait failed %d\n, ret);break;}count io_uring_peek_batch_cqe(ctx.ring, cqes[0], CQES);for (i 0; i count; i) {ret process_cqe(ctx, cqes[i], 0);if (ret)goto cleanup;}io_uring_cq_advance(ctx.ring, count);}cleanup:cleanup_context(ctx);close(sockfd);return ret;
}更多的代码可以参看GITHUB中的相关代码。
四、技术说明
通过上面的分析大家是否有所明白。包括DPDK等一些高效的框架一些主要的特点基本都是通过共享内存来减少内核与应用层的交互特别是数据的拷贝通过无锁队列来实现高效的非阻塞的通信。这意味着内核以后的工作可能会尽量减少和上层应用的直接通信。换句话说内核要做一个高效的管理者和协调者而不是一个大而全的工作者。 这其实给开发者的一个借鉴在于设计和开发时要适当抽象抽象越高意味着引入的中间层越多在可能的前提下尽量减少中间环节引入新技术如无锁编程等异步工作是方向。 以Linux的IO技术发展为例可以为两大部分即内核原生支持和非原生支持DPDK。而原生支持里从同步阻塞IO开始慢慢发展到异步IO的过程又到io_uring融入内核的过程就是给予开发者借鉴的一个经典的例子。可能大多数开发者意识不到这一点也有可能在国内这个过程非常迅速甚至忽略某些中间过程。但实际上整个的方向和发展的过程的经历基本是类似的。 本文对io_uring只是一个基础的介绍说明大家有兴趣可以深入学习并引入自己的工程中去。实践出真知。
五、总结
这里没有更细节展开谈IO模型和同步异步以及阻塞非阻塞的问题这些基础的问题不明白的需要去查阅相关资料。这篇文章只是一个开始以后有机会的话会将整个网络开发进行整体的分析说明和具体应用。 技术的进步一定是向前发展的开发者们一定睁开眼睛瞭望世界不要固守一隅夜郎自大。
