多路复用

多路复用这种技术最早是在信息论领域提出的。多路复用是指在一条物理信道上传播多种信号，每种信号的频率或者频段不一样，多路复用技术将一条物理信道划分成多个频道，每个频道传播一个频段的信号。这样多个频率或者频段的信号可以在一条物理信道上进行传播。主要目的是为了有效地利用带宽。

 

多路复用通常分为频分多路复用、时分多路复用、波分多路复用、码分多址和空分多址。

 

多路复用在软件编程领域的应用

 

在软件编程领域，多路复用技术也被用到实际编程中。比如我们需要同时处理来自控制台的标准I/O输入输出，来自文件的读写，管道的读写等。比如我们需要同时处理来自多个客户端的I/O操作，比如我们需要同时处理大量来自客户端程序的I/O请求。还有比如我们需要同时处理来自多个数据流/管道的数据处理，这种应用场景需求在目前大数据领域是非常普遍，很多大数据平台技术都有应用到这种技术的。

 

I/O多路复用

 

I/O多路复用的实现

I/O多路复用的实现比较典型包括Select、Poll、以及Epoll。

 

Select、Poll、以及Epoll的实现和应用我在另外几篇文章都有详细讲解。

 

为什么叫“I/O多路复用”呢？

I/O多路复用主要提供了一种机制来同时处理来自多个地方的I/O请求。类似像Select、Poll以及Epoll这些实现，它们都是基于一种I/O事件处理机制来同时处理多个fd的I/O事件，进而基于事件驱动来处理相关fd的I/O请求。

 

kqueue

 

写道

kernel event notification mechanism

int kqueue(void);

该系统调用提供了一种当kevent内核事件发生或者条件存在时通知用户的基于一种称为过滤器（一小段内核代码）的结果的通用机制.

 

内核事件kevent通过一个(ident, filter)对来标识表示。

 

我通过man kqueue查看帮助文档时还看到，unix后面把可选的udata value也作为标识内核事件kevent的一个要素，即kevent通过一个(ident, filter, and optional udata value)元组来标识表示，这里的叫法也不一样了，这里是一个三元组，当然上面的(ident, filter)对也是一个元组，上面是一个二元组。udata value作为标识内核事件kevent的一个要素是可选的，只有kevent的flags指定了EV_UDATA_SPECIFIC，udata value才被认为是元组的一部分，即是一个三元组，udata value也作为标识内核事件kevent的一个要素。否则的话，udata value不被认为是元组的一部分，即还是一个三元组，和上面的一样。

 

kqueue的事件触发时机和linux下epoll的水平触发类似，事件注册后，只要读缓存中有数据可读，就会触发EVFILT_READ读事件，只要写缓存可写，就会触发EVFILT_WRITE写事件。

 

 

 

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>

#include <unistd.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/event.h>

typedef struct
{
  char *ptr;
  unsigned int size;
  unsigned int offset; 
} buffer_t;

#define BUFFER_INITIAL_DEFAULT_SIZE 1024

int buffer_alloc(buffer_t *buf, int size)
{
  if (! buf) 
  {
    return -1;
  }
  if (size <= 0)
  {
    size = BUFFER_INITIAL_DEFAULT_SIZE;
  }
  buf->size = size;
  buf->ptr = (char *) malloc(buf->size);
  if (! buf->ptr)
  {
    return -1;
  }
  buf->offset = 0;
  return 0;
}

int main(int argc, char **argv)
{
  int kq_fd;

  struct kevent event, *eventlist;
  int nevents = 100;

  int fd;
  struct sockaddr_in sock_addr;
  int reuse = 1;
  buffer_t *r_buf;

  kq_fd = kqueue();
  if (kq_fd == -1)
  {
    printf("kqueue err=%d!\n", errno);
    return -1;
  }  

  fd = socket(AF_INET, SOCK_STREAM, 0);
  if (fd == -1)
  {
    printf("socket err=%d\n", errno);
    return -1;
  }

  bzero(&sock_addr, sizeof(struct sockaddr_in));
  sock_addr.sin_family = AF_INET;
  sock_addr.sin_addr.s_addr = htonl(INADDR_ANY);

  sock_addr.sin_port = htons(3333);
  
  if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) == -1)
  {
    printf("setsockopt err=%d, attempting to set SO_REUSEADDR option for sock=%d\n", errno, fd);
    return -1;	
  }

  if (bind(fd, (struct sockaddr *) &sock_addr, sizeof(struct sockaddr_in)) == -1)
  {
    printf("bind err=%d\n", errno);
    return -1;
  }

  if (listen(fd, 50) == -1)
  {
    printf("listen err=%d\n", errno);
    return -1;
  }

  r_buf = malloc(sizeof(buffer_t));
  buffer_alloc(r_buf, 0);

  event.ident = fd;
  event.filter = EVFILT_READ;
  event.flags = EV_ADD | EV_ENABLE;
  event.fflags = 0;
  event.data = 0;
  event.udata = r_buf;

  // just register, do not return pending events.
  // and, only resiter one event by one call, but not pass 
  // an array of event to changelist to register number of 
  // events by one call.
  if (kevent(kq_fd, &event, 1, NULL, 0, NULL) == -1)
  {
    printf("kevent err=%d, while register events.\n", errno);
    return -1;
  }

  eventlist = malloc(sizeof(struct kevent) * nevents);
  if (eventlist == NULL)
  {
    printf("alloc err!\n");
    return -1;
  }
 
  while (1)
  {
    int i;

    // just return pending events, do not register event.
    int nretevents = kevent(kq_fd, NULL, 0, eventlist, nevents, NULL);
    if (nretevents == -1)
    {
      printf("kevent err=%d\n", errno);
      return -1;
    }
    printf("kevent, return %d\n", nretevents);
    for (i = 0; i < nretevents; i++)
    {
      struct kevent *e = &eventlist[i];
      buffer_t *r_buf = e->udata;
      
      printf("%d: ident=%lu, filter=%d, flags=%u, fflags=%u\n", 
        i, e->ident, e->filter, e->flags, e->fflags);
      if (e->filter == EVFILT_READ)
      {
        if (e->ident == fd)
        {
          // accept
          int fd_get;
          struct sockaddr_in sockaddr;
          char *ip;

          socklen_t sockaddr_len = sizeof(struct sockaddr_in);
          fd_get = accept(fd, (struct sockaddr *) &sockaddr, &sockaddr_len);
          if (fd_get == -1)
          {
            printf("accept err=%d\n", errno);
          }
          else
          {
            buffer_t *r_buf;

            ip = inet_ntoa(sockaddr.sin_addr);
            printf("accept, ip=%s\n", ip);


            r_buf = malloc(sizeof(buffer_t));
            if (! r_buf)
            {
              printf("alloc err!\n");
              return -1;	
            }
            if (buffer_alloc(r_buf, 0))
            {
              printf("buffer alloc err!\n");
              return -1;	
            }
            
            event.ident = fd_get;
            event.udata = r_buf;

            // just register, do not return pending events.
            // and, only resiter one event by one call, but not pass 
            // an array of event to changelist to register number of 
            // events by one call.
            if (kevent(kq_fd, &event, 1, NULL, 0, NULL) == -1)
            {
              printf("kevent err=%d, while register events.\n", errno);
              return -1;
            }
          }
        }
        else
        {
          // read
          
          if (r_buf->offset >= r_buf->size)
          {
            int size = r_buf->size + BUFFER_INITIAL_DEFAULT_SIZE;
            r_buf->ptr = realloc(r_buf->ptr, size);
            if (! r_buf->ptr) 
            {
              printf("alloc err!\n");
            }
            else
            {
              r_buf->size = size;
            }
          }
          if (r_buf->offset < r_buf->size)
          {
            ssize_t nbytes;
            nbytes = read(e->ident, r_buf->ptr + r_buf->offset, r_buf->size - r_buf->offset);
            if (nbytes == -1)
            {
              printf("read err=%d\n", errno);
            }
            r_buf->offset += nbytes;
            printf("buffer: read=%.*s\n", r_buf->offset, r_buf->ptr);

            printf("event: ident=%lu, filter=%d, flags=%u, fflags=%u, data=%ld, udata=%p\n", 
              event.ident, event.filter, event.flags, event.fflags, event.data, event.udata);

            //*
            //event.filter = EVFILT_WRITE;
            //event.flags = EV_ADD | EV_ONESHOT | EV_ENABLE;
            
            event.ident = e->ident;
            event.filter = EVFILT_WRITE;
            event.flags = EV_ADD | EV_ONESHOT | EV_ENABLE;
            event.fflags = 0;
            event.data = 0;
            event.udata = r_buf;
            
            if (kevent(kq_fd, &event, 1, NULL, 0, NULL) == -1)
            {
              printf("kevent err=%d, while register events.\n", errno);
              return -1;
            }
            //*/
          }
        }
      }
      else if (e->filter == EVFILT_WRITE)
      {
        // write
        
        write(e->ident, "Ok\n", 3);
        
        printf("event: ident=%lu, filter=%d, flags=%u, fflags=%u, data=%ld, udata=%p\n", 
          event.ident, event.filter, event.flags, event.fflags, event.data, event.udata);
        //*
        //event.ident = e->ident;
        //event.filter = EVFILT_READ;
        //event.flags = EV_ADD | EV_ENABLE;
        
        event.ident = e->ident;
        event.filter = EVFILT_READ;
        event.flags = EV_ADD | EV_ENABLE;
        event.fflags = 0;
        event.data = 0;
        event.udata = r_buf;
        if (kevent(kq_fd, &event, 1, NULL, 0, NULL) == -1)
        {
          printf("kevent err=%d, while register events.\n", errno);
          return -1;
        }
        //*/
      }
    }
  }
  return 0;
}