Nginx 源代码笔记 - HTTP 模块 - memcache

Memcache Official Memcache Protocol

• 选用 0.8.34 的内建模块代码。此模块只能用来从 memcached 读取数据。使用 $memcached_key 的值作为 key 从 memcached 中获取对应的值。

• 先复习一下 memcached 的协议 (Text Protocol)：Text lines are always teminated by \r\n. Unstructured data is also terminated by \r\n, even though \r, \n or any other 8-bit characters may also appear inside the data. Therefore, when a client retrieves data from a server, it must use the length of the data block to determine where the data block ends, and not the fact that \r\n follows the end of the data block, even though it does.

  • Data stored by memcached is identified with the help of a key. A key is a text string which should uniquely identify the data for clients that are interested in storing and retrieving it. Curreny the length limit of a key is set at 250 characters; the key must not include control characters or whitespace.

  • Storage commands - set, add, replace, append, prepend, cas. The client sends a command line, and then a data block; after that the client expects one line of response, which will indicate success or failure.

  • Retrieval commands - get, gets. The client sends a command line, which includes all the requested keys; after that for each item the server finds it sends to the client one response line with information about the item, and one data block with the item's data; the continues until the server finished with the "END" response line.

  • Other commands. The client sends one command line, and expects either one line of response, or several lines of response ending with "END" one the last line.

• 模块命名为 ngx_http_memcached_module, 配置结构体 ngx_http_memcached_loc_conf_t。 请求相关的模块上下文 ngx_http_memcached_ctx_t。

• 此模块提供了一个 CONTENT PHASE handler，并且是独占式的，即，直接修改 location 的 handler 成员。此 handler 在配置了 memcached_pass 指令后才 会被注册。独占式和非独占式 (将 handler 注册到 phase engine 中) 的区别是独 占式的 handler 作用于单独的 location，并且 Nginx 不会再调用其它 phase handler (参见 ngx_http_core_content_phase)。

  ngx_http_core_find_config_phase
      ngx_http_update_location_config
  
          if (clcf->handler) {
              r->content_handler = clcf->handler;
          }
  
  ngx_http_core_content_phase
  
      if (r->content_handler) {
          r->write_event_handler = ngx_http_request_empty_handler;
          ngx_http_finalize_request(r, r->content_handler(r));
          return NGX_OK;
      }
  
  ngx_http_memcached_pass
  
      clcf = ngx_http_conf_get_module_loc_conf(cf, ngx_http_core_module);
      clcf->handler = ngx_http_memcached_handler;
  

• upstream 机制是事件驱动框架与 HTTP 框架的综合，它既属于 HTTP 框架的一部分， 又可以处理所有基于 TCP 的应用层协议 (不限于 HTTP)。它不仅无阻塞地实现了 Nginx 与上游服务器的交互，同时又很好的解决了一个请求，多个 TCP 连接、多个读/写事件间 的复杂关系。为了帮助 Nginx 实现反向代理功能，upstream 机制提供基本的与上游交 互的功能，实现了转发上游应用层协议的响应包体到下游客户端的功能。

• 基于事件驱动架构的 upstream 机制所要访问的就是所有支持 TCP 的上游服务器。在 下游的 HTTP 协议和上游的应用协议间 (包括 HTTP 协议本身) 由 upstream 进行适配。 一般应用层协议都可以被抽像成包头和包体两部分后，包头中包含有包体的长度等信息， 通过包头，能知道该为包体分配多大的空间、采取什么样的措施处理包体或者得知什么时机 可以认为正确无误的接收了包体。upstream 也将上游的响应划分为包头、包体两部分， 包头部分必须由 HTTP 模块实现的 process_header 方法解析、处理 (根据上游数据生成 要发往下游的包头字段，告知 upstream 包体的长度等)，包体则由 upstream 不做修 改地进行转发。

• 在分析 upstream 的代码时，我们知道，每一个使用 upstream 和后端打交道的模块 都发定义一个 ngx_http_upstream_conf_t 变量，用于存储给 upstream 模块提供的配 置参数。其中，ngx_http_upstream_conf_t::upstream 用于存储后端服务的配置信息， 由 ngx_http_upstream_add 函数初始化，这个成员可以指向新创建的后端配置，也可以 使用由 upstream {} 设定的后端配置。

  • memcached_module 提供的指令，主要功能就是对 ngx_http_upstream_conf 进 行设置。

    { ngx_string("memcached_sent_timeout"), NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1, ngx_conf_set_msec_slot, NGX_HTTP_LOC_CONF_OFFSET, offsetof(ngx_http_memcached_loc_conf_t, upstream.send_timeout), NULL }

  • memcached_pass 除了设置 upstream.upstream 外，还设置了所在 location 的 handler 成员为 ngx_http_memcached_handler。同时，还定义了查询使用的 key 的变量 $memcached_key。

    static char * ngx_http_memcached_pass(ngx_conf_t cf, ngx_command_t cmd, void conf) { ngx_http_memcached_loc_conf_t mlcf = conf; ... if (mlcf->upstream.upstream) { return "is duplicate"; }

    value = cf->args->elts;
    
    ngx_memzero(&u, sizeof(ngx_ur_t));
    
    u.url = value[1];
    u.no_resolve = 1;
    
    mlcf->upstream.upstream = ngx_http_upstream_add(cf, &u, 0);
    ...
    clcf->handler = ngx_http_memcached_handler;
    ...
    mlcf->index = ngx_http_get_variable_index(cf, &ngx_http_memcached_key);
    ...
    return NGX_CONF_OK;
    

    }

  • ngx_http_upstream_srv_conf_t - 一个 upstream 的上游服务器配置。Nginx 生成这个结构体后，将它存储于 ngx_http_upstream_main_conf_t::upstreams 数组 中，同时也能够通过 ngx_http_upstream_conf_t.upstream 引用到它。

  • ngx_http_upstream_t - 请求相关的 upstream 信息维护结构，它包含 upstream 的配置，处理 upstream 响应的回调函数和数据缓存等等信息。

• 请求匹配到配置了 memcached_pass 的 location 后，调用其 handler 即 ngx_http_memcached_handler 函数，配置回调函数，开始启动 upstream 机制，并发 起 upstream 请求。

  • 发起请求

    static ngx_int_t ngx_http_memcached_handler(ngx_http_request_t *r) { ... rc = ngx_http_discard_request_body(r); ... if (ngx_http_upstream_create(r) != NGX_OK) { return NGX_HTTP_INTERNAL_SERVER_ERROR; }

    u = r->upstream;
    
    u->schema.len = sizeof("memcached://") - 1;
    u->schema.data = (u_char *) "memcached://";
    
    u->output.tag = (ngx_buf_tag_t) &ngx_http_memcached_module;
    
    mlcf = ngx_http_get_module_loc_conf(r, ngx_http_memcached_module);
    
    u->conf = &mlcf->upstream;
    
    u->create_request = ngx_http_memcached_create_request;
    u->reinit_request = ngx_http_memcached_reinit_request;
    u->process_header = ngx_http_memcached_process_header;
    u->abort_request = ngx_http_memcached_abort_request;
    u->finalize_request = ngx_http_memcached_finalize_request;
    
    ctx = ngx_palloc(r->pool, sizeof(ngx_http_memecached_ctx_t));
    ...
    ngx_http_set_ctx(r, ctx, ngx_http_memcached_module);
    
    u->input_filter_init = ngx_http_memcached_filter_init;
    u->input_filter = ngx_http_memcached_filter;
    u->input_filter_ctx = ctx;
    
    r->main->count++; /* 下游事件结束了，但又有了对上游的请求事件 */
    
    ngx_http_upstream_init(r);
    
    return NGX_DONE;
    

    }

  • 在向后端的连接开始前，upstream 调用上面设定的 create_request 模块创建 要发往上游服务器的请求。这里的请求是应用层协议相关的，比如此模块需要构造合法 的 memcached get 请求：

    ---

    get \r\n

    static ngx_int_t ngx_http_memcached_create_request(ngx_http_request_t r) { ... ngx_http_variable_value_t vv; ... vv = ngx_http_get_indexed_variable(r, mlcf->index); ... escape = 2 * ngx_escape_uri(NULL, vv->data, vv->len, NGX_ESCAPE_MEMCACHED);

    len = sizeof("get ") - 1 + vv->len + escapse + sizeof(CRLF) - 1;
    
    b = ngx_create_temp_buf(r->pool, len);
    ...
    cl = ngx_alloc_chain_link(r->pool);
    ...
    cl->buf = b;
    cl->next = NULL;
    
    r->upstream->request_bufs = cl;
    
    *b->last++ = 'g'; *b->last++ = 'e'; *b->last++ = 't'; *b->last++ = ' ';
    ...
    if (escape == 0) {
        b->last = ngx_copy(b->last, vv->data, vv->len);
    
    } else {
        b->last = (u_char *) ngx_escape_uri(b->last, vv->data, vv->len,
                                            NGX_ESCAPE_MEMCACHED);
    }
    ...
    *b->last++ = CR; *b->last++ = LF;
    
    return NGX_OK;
    

    }

  • 收到上游服务器响应时，upstream 将调用 process_header 处理包头部分，随 后原封不动发送包体部分。此模块将上游响应的数据信息部分作为包头处理：

    ---

    VALUE []\r\n \r\n END\r\n

    ---

    static ngx_int_t ngx_http_memcached_process_header(ngx_http_request_t *r) { ... u = r->upstream;

    for (p = u->buffer.pos; p < u->buffer.last; p++) {
        if (*p == LF) {
            goto found;
        }
    }
    
    return NGX_AGAIN;
    

    found:

    *p = '\0';
    
    line.len = p - u->buffer.pos - 1;
    line.data = u->buffer.pos;
    
    p = u->buffer.pos;
    
    if (ngx_strncmp(p, "VALUE ", sizeof("VALUE ") - 1) == 0) {
    
        p += sizeof("VALUE ") - 1;
    
        if (ngx_strncmp(p, ctx->key.data, ctx->key.len) != 0) {
            ...
            return NGX_HTTP_UPSTREAM_INVALID_HEADER;
        }
    
        p += ctx->key.len;
    
        if (*p++ != ' ') {
            goto no_valid;
        }
    
        /* skip flags */
    
        while (*p) {
            if (*p++ == ' ') {
                goto length;
            }
        }
    
        goto no_valid;
    
    length:
    
        len = p;
    
        while (*p && *p++ != CR) { /* void */ }
    
        r->headers_out.content_length_n = ngx_atoof(len, p - len - 1);
        ...
        u->headers_in.status_n = 200;
        ...
        u->buffer.pos = p + 1;
    
        return NGX_OK;
    }
    
    if (ngx_strcmp(p, "END\x0d") == 0) {
        ...
        u->headers_in.status_n = 404;
        return NGX_OK;
    }
    ...
    

    }

• 另外的几个回调函数就不作详细分析了，只将它们的主要功能列举如下：

  • ngx_http_memcached_filter_init - 模块处理包体前的初始化方法。
  • ngx_http_memcached_filter - 模块实现的处理上游包体的方法。

• LAST

  • ngx_http_upstream_conf_t - upstream 各种参数和运行方式。

  • ngx_http_upstream_srv_conf_t - 配置文件中解析而来的后端服务器信息。

  • ngx_http_upstream_t - 每个请求都有此类型变量，保存 upstream 配置和后端 服务器连接、信息等，及用于接收上游数据的缓存区 和模块回调函数。

  • ngx_http_upstream_main_conf_t - 配置文件中的所有 srv_conf 配置所在地， 内置响应包头的 hash 表。

  • ngx_http_memcached_ctx_t - 在请求中保存模块上下文用的结构体。

  • ngx_http_memcached_loc_conf_t - 每个模块都要有 upstream_conf，每个 upstream_conf 都对应一个 upstream_srv_conf_t；

Comments

不要轻轻地离开我，请留下点什么...

Comments

comments powered by Disqus

---