Memcache源码阅读（3）---处理用户输入

桃扇骨 2022-07-13 00:45 251阅读 0赞

我看的源码版本是1.2.4

从main函数开始，看看作者怎么初始化,然后再看作者怎么监听连接，怎么接受用户的输入，怎么处理用户的输入。

主程序

main函数里的一些操作

  1. 自定义中断处理函数
  2. 从命令行中获取用户打开memcached的指令值
  3. 设置core文件
  4. 初始化资源，统计资源，分配内存
  5. 屏蔽SIGPIPE信号，因为SIGPIPE的默认处理是中断程序，客户端断开连接后，memcached再往客户端写数据就会触发该信号，此时memcached应该要继续运行的，所以要屏蔽该信号。
  6. 监听用户的连接，作者用到libevent的事件驱动模型，一个主线程负责监听连接，将接受到的连接交给工作线程去处理。

主程序监听套接字的设置

  1. //创建一个套接字
  2. l_socket = server_socket(settings.port, 0);
  4. //对监听套接字设置事件处理，这里主要是为套接字l_socket设置事件处理函数
  5. //将l_socket设置为conn_listening状态，一直在这个状态中运行。
  6. //后面就会看到关于状态机的处理代码。
  7. listen_conn = conn_new(l_socket, conn_listening,
  8.                                  EV_READ | EV_PERSIST, 1, false, main_base);
  9. //程序会一直在这里等待,有时间到来就会调用注册的时间处理函数                                                                 
  10. event_base_loop(main_base, 0);

主线程创建一个套接字，然后对这个套接字设置一个状态，为conn_listening，这个套接字一直都处理conn_listening状态，它一直负责接收用户的连接。主线程收到客户的连接，就会将这个连接交给工作线程处理，然后主线程继续回到监听状态，等待下一个连接。

细节描述

  1. //为sfd添加事件处理函数。memcached使用libevent这个库来进行事件处理。
  2. conn *conn_new(const int sfd, const int init_state, const int event_flags,
  3.                 const int read_buffer_size, const bool is_udp, struct event_base *base) {
  4.     conn *c = conn_from_freelist();
  5.     ...
  6.     一些对conn的初始化
  7.     ...
  8.     //设置事件处理,时间处理函数为event_handler
  9.     event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
  10.     event_base_set(base, &c->event);
  11.     c->ev_flags = event_flags;
  12.     event_add(&c->event, 0)
  13.     return c;
  14. }
  16. //事件处理就是交给drive_machine，他的参数是一个连接，每个连接都有一个状态，与它接收到的数据，每个工作线程可以根据它现在的状态和收据做对应的操作。
  17. void event_handler(const int fd, const short which, void *arg) {
  18.     conn *c;
  19.     c = (conn *)arg;
  20.     drive_machine(c);
  21. }
  23. //所有的连接的事件处理函数都是它，它会根据连接的状态和数据来作对应的处理
  24. static void drive_machine(conn *c) {
  25.     while (!stop) {
  26.         switch(c->state) {
  27.             case conn_listening:
  28.                     //监听进程一直都在执行这里的代码
  29.                     //接收连接，然后分发给工作线程，注意，交给工作线程时设置初始状态为conn_read
  30.                     sfd = accept(c->sfd, &addr, &addrlen)) 
  31.                     dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST,
  32.                                      DATA_BUFFER_SIZE, false);
  33.                     break;
  34.             case conn_read:
  35.                 //接收命令行传过来的参数，就是第二篇文章讲的第一行命令，
  36.                 //如set a 0 0 4
  37.                 //try_read_command判断接收的消息有没有一个可以操作的命令，
  38.                 //如果可以操作，就会跳到对应的状态，否则继续在这个状态，继续接收数据。
  39.                 //这里接收数据是通过try_read_network这个函数来接收数据的。
  40.                 if (try_read_command(c) != 0) {
  41.                     continue;
  42.                 }
  43.                 //这里是用来接收用户传来的信息
  44.                 if ((c->udp ? try_read_udp(c) : try_read_network(c)) != 0) {
  45.                     continue;
  46.                 }
  47.                 stop = true;
  48.                 break;
  50.             case conn_nread:
  51.                 //接收用户的数据块
  52.                 //如set a 0 0 4后的 1234
  53.                 if (c->rlbytes == 0) {
  54.                     complete_nread(c);
  55.                     break;
  56.                  }
  57.                  res = read(c->sfd, c->ritem, c->rlbytes);
  58.                  if (res > 0) {
  59.                     break;
  60.                 }
  61.                 if (res == 0) { /* end of stream */
  62.                     conn_set_state(c, conn_closing);
  63.                     break;
  64.                 }
  65.                 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
  66.                     stop = true;
  67.                     break;
  68.                 }
  69.                 /* otherwise we have a real error, on which we close the connection */
  70.                 conn_set_state(c, conn_closing);
  71.                 break;
  72.             //这些后面再讲
  73.             case conn_swallow:
  74.             case conn_write:
  75.             case conn_mwrite:
  76.             case conn_closing:
  77.         }
  78.     }
  80.     return;
  81. }
  83. //因为一个客户端与memcached是一个tcp长连接，这个长连接一般会保持到用户主动退出或者网络断开，
  84. //还有这个套接字是异步套接字，我觉得理解这个概念很重要，用户输入完一个命令之后可以从这个while循环中返回。
  85. //因为是异步套接字，所以这个函数在用户不需要断开连接也是可以返回的,
  86. //然后一会再检查能不能构成一个可以执行的命令，
  87. //如果可以就执行，不行就返回0，等待数据。
  88. static int try_read_network(conn *c) {
  89.     int gotdata = 0;
  90.     int res;
  92.     while (1) {
  93.         if (c->rbytes >= c->rsize) {
  94.             char *new_rbuf = realloc(c->rbuf, c->rsize * 2);
  95.             c->rcurr = c->rbuf = new_rbuf;
  96.             c->rsize *= 2;
  97.         }
  99.         res = read(c->sfd, c->rbuf + c->rbytes, c->rsize - c->rbytes);
  100.         if (res > 0) {
  101.             continue;
  102.         }
  103.         if (res == 0) {
  104.             /* connection closed */
  105.             conn_set_state(c, conn_closing);
  106.             return 1;
  107.         }
  108.         if (res == -1) {
  109.             if (errno == EAGAIN || errno == EWOULDBLOCK) break;
  110.             else return 0;
  111.         }
  112.     }
  113.     return gotdata;
  114. }
  116. //这个函数就是获取用户输入的第一行命令的一个预处理，将一个命令的\r\n结束改为以\0作为结束。
  117. //比如说用户输入 “get a\r\n“那么经过这里处理就变成 “get a\0\n“,为的是给后面做分割
  118. static int try_read_command(conn *c) {
  119.     char *el, *cont;
  121.     if (c->rbytes == 0)
  122.         return 0;
  123.     el = memchr(c->rcurr, '\n', c->rbytes);
  124.     if (!el)
  125.         return 0;
  126.     cont = el + 1;
  127.     if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {
  128.         el--;
  129.     }
  130.     *el = '\0';
  132.     process_command(c, c->rcurr);
  134.     return 1;
  135. }
  137. //处理用户输入,这里只会处理用户输入的第一行。
  138. //即使用户输入的是 set a 0 0 4\r\n  
  139. // 1234\r\n
  140. //第一个\r\n后的1234不是这里处理的，它会交由drive_machine的conn_nread这里来读取和处理。
  141. static void process_command(conn *c, char *command) {
  143.     token_t tokens[MAX_TOKENS];
  144.     size_t ntokens;
  145.     //用户输入的字符串分割开，tokens数组放着这些值
  146.     ntokens = tokenize_command(command, tokens, MAX_TOKENS);
  147.     if (ntokens >= 3 &&
  148.         ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) ||
  149.          (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) {
  151.         process_get_command(c, tokens, ntokens, false);
  153.     }else if (ntokens == 6 &&
  154.                ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
  155.                 (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
  156.                 (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) ||
  157.                 (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) ||
  158.                 (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) {
  160.         process_update_command(c, tokens, ntokens, comm, false);
  162.     }
  163.     ...
  164.     //处理其他的命令
  166.     return;
  167. }
  169. //分割字符串
  170. //如用户输入get a\r\n它这里会返回3，分别为get, a, a. 这是我测试的结果，原因我也不清楚。
  171. //用户输入set a 0 0 4\r\n它会返回6，分别为set, a, 0, 0, 4, 4.
  172. static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) {
  173.     char *s, *e;
  174.     size_t ntokens = 0;
  176.     assert(command != NULL && tokens != NULL && max_tokens > 1);
  178.     for (s = e = command; ntokens < max_tokens - 1; ++e) {
  179.         if (*e == ' ') {
  180.             if (s != e) {
  181.                 tokens[ntokens].value = s;
  182.                 tokens[ntokens].length = e - s;
  183.                 ntokens++;
  184.                 *e = '\0';
  185.             }
  186.             s = e + 1;
  187.         }
  188.         else if (*e == '\0') {
  189.             if (s != e) {
  190.                 tokens[ntokens].value = s;
  191.                 tokens[ntokens].length = e - s;
  192.                 ntokens++;
  193.             }
  195.             break; /* string end */
  196.         }
  197.     }
  199.     /*
  200.      * If we scanned the whole string, the terminal value pointer is null,
  201.      * otherwise it is the first unprocessed character.
  202.      */
  203.     tokens[ntokens].value =  *e == '\0' ? NULL : e;
  204.     tokens[ntokens].length = 0;
  205.     ntokens++;
  207.     return ntokens;
  208. }
  210. //对tokens进行保存，改变状态，交由drive_machine去处理
  211. static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {
  212.     char *key;
  213.     size_t nkey;
  214.     int flags;
  215.     time_t exptime;
  216.     int vlen, old_vlen;
  217.     uint64_t req_cas_id;
  218.     item *it, *old_it;
  219.     //将命令行中的各个值提取出来
  220.     key = tokens[KEY_TOKEN].value;
  221.     nkey = tokens[KEY_TOKEN].length;
  223.     flags = strtoul(tokens[2].value, NULL, 10);
  224.     exptime = strtol(tokens[3].value, NULL, 10);
  225.     vlen = strtol(tokens[4].value, NULL, 10);
  226.     //分配一个item的空间来寸数据，第二行输入的数据是存在item里面的
  227.     it = item_alloc(key, nkey, flags, realtime(exptime), vlen+2);
  229.     c->item = it;
  230.     c->ritem = ITEM_data(it);
  231.     c->rlbytes = it->nbytes;
  232.     c->item_comm = comm;
  233.     //将这个连接的状态设置为conn_nread，交由drive_machine去处理,现在可以返回到drive_machine那里看看你conn_nread要做什么了，conn_nread主要是要读取接下来的数据，接下来的数据是读到item里面去的哦。接下来如果要保存到hash桶里，也是直接用这个item的指针而已。读完数据之后会执行complete_nread.
  234.     conn_set_state(c, conn_nread);
  235. }
  237. //将数据保存到memcached
  238. static void complete_nread(conn *c) {
  239.     assert(c != NULL);
  241.     item *it = c->item;
  242.     int comm = c->item_comm;
  243.     int ret;
  245.     if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) {
  246.         out_string(c, "CLIENT_ERROR bad data chunk");
  247.     } else {
  248.       ret = store_item(it, comm);
  249.       if (ret == 1)
  250.           out_string(c, "STORED");
  251.       else if(ret == 2)
  252.           out_string(c, "EXISTS");
  253.       else if(ret == 3)
  254.           out_string(c, "NOT_FOUND");
  255.       else
  256.           out_string(c, "NOT_STORED");
  257.     }
  259.     item_remove(c->item);       /* release the c->item reference */
  260.     c->item = 0;
  261. }

这个处理输入的分析我就写到这里了。我这里只写了处理update(set, add, …)这个命令的操作作为例子，展示了从建立连接，接受请求，接受数据，到处理请求这一个流程是怎么走的。截取的源代码省略了很多安全处理的操作。

发表评论

表情：
评论列表 （有 0 条评论，251人围观）

还没有评论，来说两句吧...

相关阅读