twemproxy源码分析之四:处理流程ji(内容属于转载。

nc_connection.c

很赞的注释:

*                   nc_connection.[ch]
*                Connection (struct conn)
*                 +         +          +
*                 |         |          |
*                 |       Proxy        |
*                 |     nc_proxy.[ch]  |
*                 /                    *              Client                Server
*           nc_client.[ch]         nc_server.[ch]

messsage.c

*            nc_message.[ch]
*        _message (struct msg)
*            +        +            .
*            |        |            .
*            /        \            .
*         Request    Response      .../ nc_mbuf.[ch]  (mesage buffers)
*      nc_request.c  nc_response.c .../ nc_memcache.c; nc_redis.c (_message parser)

* Messages in nutcracker are manipulated by a chain of processing handlers,
* where each handler is responsible for taking the input and producing an
* output for the next handler in the chain. This mechanism of processing
* loosely conforms to the standard chain-of-responsibility design pattern

*             Client+             Proxy           Server+
*                              (nutcracker)
*                                   .
*       msg_recv {read event}       .       msg_recv {read event}
*         +                         .                         +
*         |                         .                         |
*         \                         .                         /
*         req_recv_next             .             rsp_recv_next
*           +                       .                       +
*           |                       .                       |       Rsp
*           req_recv_done           .           rsp_recv_done      <===
*             +                     .                     +
*             |                     .                     |
*    Req      \                     .                     /
*    ===>     req_filter*           .           *rsp_filter
*               +                   .                   +
*               |                   .                   |
*               \                   .                   /
*               req_forward-//  (1) . (3)  \\-rsp_forward
*                                   .
*                                   .
*       msg_send {write event}      .      msg_send {write event}
*         +                         .                         +
*         |                         .                         |
*    Rsp‘ \                         .                         /     Req‘
*   <===  rsp_send_next             .             req_send_next     ===>
*           +                       .                       +
*           |                       .                       |
*           \                       .                       /
*           rsp_send_done-//    (4) . (2)    //-req_send_done
*
*
* (1) -> (2) -> (3) -> (4) is the normal flow of transaction consisting
* of a single request response, where (1) and (2) handle request from
* client, while (3) and (4) handle the corresponding response from the
* server.

好有爱的注释!!

对应这段注释的代码:

struct conn *
conn_get(void *owner, bool client, bool redis)
{
    struct conn *conn;

    conn = _conn_get();

    conn->client = client ? 1 : 0;

    if (conn->client) {

        conn->recv = msg_recv;
        conn->recv_next = req_recv_next;
        conn->recv_done = req_recv_done;

        conn->send = msg_send;
        conn->send_next = rsp_send_next;
        conn->send_done = rsp_send_done;

        conn->close = client_close;
        conn->active = client_active;

        conn->ref = client_ref;
        conn->unref = client_unref;

        conn->enqueue_inq = NULL;
        conn->dequeue_inq = NULL;
        conn->enqueue_outq = req_client_enqueue_omsgq;
        conn->dequeue_outq = req_client_dequeue_omsgq;
    } else {

        conn->recv = msg_recv;
        conn->recv_next = rsp_recv_next;
        conn->recv_done = rsp_recv_done;

        conn->send = msg_send;
        conn->send_next = req_send_next;
        conn->send_done = req_send_done;

        conn->close = server_close;
        conn->active = server_active;

        conn->ref = server_ref;
        conn->unref = server_unref;

        conn->enqueue_inq = req_server_enqueue_imsgq;
        conn->dequeue_inq = req_server_dequeue_imsgq;
        conn->enqueue_outq = req_server_enqueue_omsgq;
        conn->dequeue_outq = req_server_dequeue_omsgq;
    }

    conn->ref(conn, owner);

    return conn;
}

一个请求被处理的流程

我们按照 messsage.c 里面的4个步骤

后面的图都采用这样的表示方法:

初始状态

考察只有一个后端的情况, 假设有2个client要发送3个请求过来

1.读取请求

此时回调函数:

conn->recv = msg_recv;
conn->recv_next = req_recv_next;
conn->recv_done = req_recv_done;

函数调用栈:

每次发生 req_recv_done(req_forward), 就会调用 req_forward()

req_forward(struct context *ctx, struct conn *c_conn, struct msg *msg)
{
    if (!msg->noreply) {
        c_conn->enqueue_outq(ctx, c_conn, msg);
    }

    //获得到后端的连接. (可能是新建, 或者从pool里面获取)
    s_conn = server_pool_conn(ctx, c_conn->owner, key, keylen);

    s_conn->enqueue_inq(ctx, s_conn, msg);
    event_add_out(ctx->evb, s_conn);
}

就有一个msg就出现在client_conn->out_q, 同时出现在server_conn->in_q

req_forward用server_pool_conn获得一个server_conn.

2.转发到后端

对于server_conn来说, 因为挂了epoll_out事件, 很快就会调用 conn->send,也就是msg_send.

此时:

conn->send = msg_send;
conn->send_next = req_send_next;
conn->send_done = req_send_done;

调用栈:

#这时, 每次发生 req_send_done, 这个msg就被放到server_conn->out_q

注意, 此时两个msg依然在client_conn->in_q里面

3.接收后端响应

因为server_conn的 epoll_in是一直开着的, 响应很快回来后, 就到了server_conn的 msg_recv 这个过程和调用栈1类似,不过两个函数钩子不一样(图中灰色框)

conn->recv = msg_recv;(和<1>一样)
conn->recv_next = rsp_recv_next;
conn->recv_done = rsp_recv_done;

这里 rsp_recv_next 的作用是, 拿到下一个要接收的msg

rsp_forward 会把 msg从 server_conn->outq 里面摘掉, 同时设置req和resp之间的一一对应关系

//establish msg <-> pmsg (response <-> request) link
pmsg->peer = msg;
msg->peer = pmsg;

上面这个代码是整个过程的精华所在

这时候, client的q_out上排队的req, 就有了对应的response

这时也会设置:

event_add_out(ctx->evb, c_conn);

每收到一个rsp, 就从server_conn的out_q摘掉, 并设置一一对应关系, 如下:

4.把响应回给client

现在每个请求的msg都有了一个对应的response msg, client_conn的out事件也挂上了, 下面这个调用栈:

最终, 一切归于沉寂, 后端连接依然在:

时间: 2024-10-25 08:19:00

twemproxy源码分析之四:处理流程ji(内容属于转载。的相关文章

HBase1.0.0源码分析之请求处理流程分析以Put操作为例(二)

HBase1.0.0源码分析之请求处理流程分析以Put操作为例(二) 1.通过mutate(put)操作,将单个put操作添加到缓冲操作中,这些缓冲操作其实就是Put的父类的一个List的集合.如下: private List<Row> writeAsyncBuffer = new LinkedList<>(); writeAsyncBuffer.add(m); 当writeAsyncBuffer满了之后或者是人为的调用backgroundFlushCommits操作促使缓冲池中的

openVswitch(OVS)源码分析之工作流程(哈希桶结构体的解释)

这篇blog是专门解决前篇openVswitch(OVS)源码分析之工作流程(哈希桶结构体的疑惑)中提到的哈希桶结构flex_array结构体成员变量含义的问题. 引用下前篇blog中分析讨论得到的flex_array结构体成员变量的含义结论: struct { int element_size; // 这是flex_array_part结构体存放的哈希头指针的大小 int total_nr_elements; // 这是全部flex_array_part结构体中的哈希头指针的总个数 int e

Spark SQL源码分析之核心流程

自从去年Spark Submit 2013 Michael Armbrust分享了他的Catalyst,到至今1年多了,Spark SQL的贡献者从几人到了几十人,而且发展速度异常迅猛,究其原因,个人认为有以下2点: 1.整合:将SQL类型的查询语言整合到 Spark 的核心RDD概念里.这样可以应用于多种任务,流处理,批处理,包括机器学习里都可以引入Sql. 2.效率:因为Shark受到hive的编程模型限制,无法再继续优化来适应Spark模型里. 前一段时间测试过Shark,并且对Spark

leveldb源码分析--插入删除流程

由于网络上对leveldb的分析文章都比较丰富,一些基础概念和模型都介绍得比较多,所以本人就不再对这些概念以专门的篇幅进行介绍,本文主要以代码流程注释的方式. 首先我们从db的插入和删除开始以对整个体系有一个感性的认识,首先看插入: Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) { WriteBatch batch; //leveldb中不管单个插入还是多个插入都是以Wri

Okhttp源码分析--基本使用流程分析

Okhttp源码分析--基本使用流程分析 一. 使用 同步请求 OkHttpClient okHttpClient=new OkHttpClient(); Request request=new Request.Builder() .get() .url("www.baidu.com") .build(); Call call =okHttpClient.newCall(request).execute(); 异步请求 OkHttpClient okHttpClient=new OkH

springmvc源码分析系列-请求处理流程

接上一篇-springmvc源码分析开头片 上一节主要说了一下springmvc与struts2的作为MVC中的C(controller)控制层的一些区别及两者在作为控制层方面的一些优缺点.今天就结合下面的一张图和上一篇中关于springmvc各个模块之间及各个模块中的类的继承关系的一张图对springmvc的请求处理流程进行一个分析.当然有了springmvc的请求处理流程我们就知道了springmvc是如何在启动的时候去加载或者去解析对应的具体控制器,以及modleAndView使干什么用的

MyBatis源码分析-MyBatis初始化流程

MyBatis 是支持定制化 SQL.存储过程以及高级映射的优秀的持久层框架.MyBatis 避免了几乎所有的 JDBC 代码和手动设置参数以及获取结果集.MyBatis 可以对配置和原生Map使用简单的 XML 或注解,将接口和 Java 的 POJOs(Plain Old Java Objects,普通的 Java对象)映射成数据库中的记录.如何新建MyBatis源码工程请点击MyBatis源码分析-IDEA新建MyBatis源码工程. MyBatis初始化的过程也就是创建Configura

Spark源码分析之四:Stage提交

各位看官,上一篇<Spark源码分析之Stage划分>详细讲述了Spark中Stage的划分,下面,我们进入第三个阶段--Stage提交. Stage提交阶段的主要目的就一个,就是将每个Stage生成一组Task,即TaskSet,其处理流程如下图所示: 与Stage划分阶段一样,我们还是从handleJobSubmitted()方法入手,在Stage划分阶段,包括最好的ResultStage和前面的若干ShuffleMapStage均已生成,那么顺理成章的下一步便是Stage的提交.在han

HBase1.0.0源码分析之请求处理流程分析以Put操作为例(一)

如下面的代码所示,是HBase Put操作的简单代码实例,关于代码中的Connection connection = ConnectionFactory.createConnection(conf),已近在前一篇博 HBase1.0.0源码分析之Client启动连接流程,中介绍了链接的相关流程以及所启动的服务信息. TableName tn = TableName.valueOf("test010"); try (Connection connection = ConnectionFa