Zookeeper C Client分析

gggggds

　　每个zookeeper API必须有一个zhandle。当初始化一个zhandle时（zookeeper_init）首先初始化zhandle的相应字段然后创建两个线程：do_io，do_completion；并且等待这两个线程初始化完成后才返回，这两个线程也要等待彼此初始化完成后，才提供服务（notify_thread_ready）。
　　1 IO线程/do_io
　　显然这个线程用于处理io请求（使用Poll多路复用方式），这里的io请求就是与zookeeper server的网络io处理（zh->fd）。除了这个io外，该线程还关注一个管道的读取端的io（adaptor_threads->self_pipe[0]），该io的唯一作用就是唤醒该线程，即其它线程通过向adaptor_threads->self_pipe[1]写一个字符来唤醒该线程（wakeup_io_thread）。对于第一种io，需要确定所关注的event，以及等待的时间（zookeeper_interest）；然后进入poll等待事件可用；当poll返回后先确定哪些事件可用了；清空adaptor_threads->self_pipe[0]；最后调用zookeeper_process处理可用事件，如对于同步请求的响应则直接在这个线程里把结果保存到应用的buf，并唤醒应用，对于异步请求则把sent_requests的completion对象放入completions_to_process队列让completion线程来处理。
　　2 completion线程/do_completion
　　该线程负责从completions_to_process取出completion来处理，这个处理包括对completion回调及watch回调的处理。处理的函数为process_completions。
　　3 API
　　Zookeeper的API可以分成三类，如我们以get为例：zoo_get、zoo_wget、zoo_awget。其中’w’表示是否自己指定watch回调函数，如果指定的话需要传入watch 回调函数及context参数，’a’表示异步方式，需要指定completion回调函数及它的data参数，并且这个这个回调函数由completion线程调用（非a表示同步方式，即等待请求从服务返回后才返回给应用，这期间应用阻塞等待）。对于内部实现来说get及wget都是通过awget来实现的。首先get->wget：如果get需要watch则使用默认zh->watcher再调用wget；wget->awget：传递SYNCHRONOUS_MARKER的completion回调函数再调用awget，并且传入的data为应用的buff，（在zookeeper_process函数里可以看到如果completion函数是SYNCHRONOUS_MARKER，则直接在io线程里处理，并将服务响应的结果放到completion的data里），然后wget调用wait_sync_completion等待io线程处理完成（从这里可见对于wget方式其实内部也是使用的异步方法来实现的，但对于客户进程来说是不一样的，因为同步方式客户会被阻塞而异步方式客户在请求发送完成后就返回，不等待请求响应，客户进程在它自己的业务逻辑里选择什么时候等待请求的响应结果，而且异步方式的回调函数由completion线程处理，而同步方式则由io线程处理（实际上没有回调函数））。
　　最后我们看一下awget是怎么实现？它首先序列化请求报文；然后构造watch的对象（create_watcher_registration：包括path,watch,wathc_ctx,checker=data_result_checker），然后将这个watch对象及completion，data，h.xid一起封装为_completion_list对象（create_completion_entry），并把这个对象放到sent_requests（发送的xid必须与接收到xid一致）队尾（add_completion）；最后再将刚才序列化的报文放入to_send列队，然后唤醒io线程出发送这个报文，注意这里并没有把watch放到hashtable里（这个行为在activateWatcher完成）。
　　4 Zookeeper网络协议相关
　　不管是什么watch对于client与server之间它只需要一个标志位，具体的watch回调函数保存在client的hashtable，服务器只会在当被标志为需要watch的node发生改变时把这个改变的类型告诉client，client再根据这个类型及node找到它的所有watch回调函数，并执行；
　　client会对每个报文指定一个xid，服务器在响应这个报文的时候把这个xid原样返回，并且预定义了小于0的几种特殊报文（如：WATCHER_EVENT_XID，AUTH，PING，SET_WATCHES_XID，对于WATCHER_EVENT_XID报文又包含type类型以让client确定watch回调函数保存的hasttable[node,exist,child]，然后再通过path hash出最终的回调函数）；其它的就是正常数据报文，这种报文又通过type标志它的请求行为（如GETDATA_OP、EXISTS_OP、CREATE_OP等），服务根据这个标志来返回相应的数据给client，而client则根据completion_type（等待结果的类型，保存在completion对象里）进行处理。
　　Client的可能状态顺序：ZOO_CONNECTING_STATE—prime req-->ZOO_ASSOCIATING_STATE—prime resp-->ZOO_CONNECTED_STATE（ZOO_EXPIRED_SESSION_STATE:client_id不同）--authfailed-->ZOO_AUTH_FAILED_STATE。其中红色表示报文，prime req表示客户端向服务端发送，prime resp、auth failed表示服务器发送给客户端。
　　5 主要函数
　　Zookeeper_interest：
　　该函数首先判断该io是否已经建立了连接，如果没有则向某个zookeep server发起连接，这时可能进入ZOO_CONNECTING_STATE状态，连接成功后则向server发起第一个prime报文（prime_connection，这里直接走send不再经过poll）进入ZOO_ASSOCIATING_STATE状态；如果现在连接已经建立起来则先计算它距接收到下一个报文的时间（可接受的时间），如果这个时间小于0说明已经超时了（calculate_recv_to当前至上一次接收到报文的时间超过2/3*recv_timeout）；如果没有超时则再计算发送ping报文的时间，如果已经到了该发送ping（距离上次ping超过1/3*recv_timeout），则向zh->to_send放入一个ping报文；最后确定的polltimeout为上面的recv_to与send_to的最小时间，即在这两个的最小时间里要么接收到报文，要么发送ping报文。更新next_deadline为这个timeout时间之后。标志关心该io的事件为：ZOOKEEPER_READ（服务器对ping的响应），如果zh->to_send有则也关注ZOOKEEPER_WRITE事件。
　　zookeeper_process：
　　该函数首先检查事件（如果是处于ZOO_CONNECTING_STATE状态并收到ZOOKEEPER_WRITE事件，则发送prime报文，进入ZOO_ASSOCIATING_STATE状态；如果是ZOOKEEPER_WRITE事件，并且to_send缓存里有需要发送的报文则把to_send里的所有报文发送完，如果block的话则直接返回（flush_send_queue，该函数还可以设置一个timeout等待时间）；如果是ZOOKEEPER_READ事件则将网络报文放到zh->input_buffer缓存，如果这个报文是对prime的响应则初始zh->client_id信息，用于下次重连，此时状态进入ZOO_CONNECTED_STATE，并向server发送所有的AUTH报文及WATCH报文（这些watch信息从active_node_watchers、exist、child里获得），最后构造一个ZOO_SESSION_EVENTwatch的completion放入completions_to_process，作为对ZOO_SESSION_EVENT（连接成功）事件的响应（queue_session_event，连接真正建立的watch回调，即连接建立成功后要自己构造一个watch，这个watch的回调函数就是zh->watcher，它关注的path为””，然后直接调用process_completions来处理这种回调）；如果接收的报文不是对prime的响应，则直接把网络报文放入to_process队列）；接下来处理to_process里接收到的所有报文，解析报文，并判断该报文的xid，如果是WATCHER_EVENT_XID，即有watch事件发生了，则创建一个WATCHER_EVENT_XIDwatch的completion，并且这个watch的回调函数由服务器指定的type（如：ZOO_SESSION_EVENT，CREATED_EVENT_DEF等）及path决定（collectWatchers），然后把这个completion放入completions_to_process队列让completions线程处理；如果接收的报文是对SET_WATCHES_XID的响应则不处理；如果是对AUTH_XID的响应则直接调用auth_completion_func来处理，注意这里不是发给completions_to_process，而是自己直接处理验证结果，如果验证失败则进入ZOO_AUTH_FAILED_STATE状态，并退出；如果接收到的是对PING的响应，则更新last_recv时间就可以了；对于其它的报文（如对get的响应等），这里首先判断响应的报文是否是当前等待的报文，如果不是则直接丢弃，此时就可以把它的watch回调放入到相应的hashtable里（activateWatcher，即只有在请求被响应并且check成功后才会把它的相应watch回调函数加入相应的hashtable），然后判断之前的请求是同步还是异步，如果是异步（使用a*的接口）则把这个请求的competion放入completions_to_process；否则则在该线程内进行处理（使用非a*的接口没有completion回调函数），这里的处理就是根据completion_type解析报文，并将解析的结果放入sync_completion里，最后notify_sync_completion通知客户端已经从服务器上接收到响应报文，api可以返回了。
　　process_completions
该函数为completions线程的处理函数，它也被其它函数直接调用如queue_session_event，即连接成功后调用默认的zh->watcher回调。它从completions_to_process队列里取出需要处理的completions，并进行处理。首先它判断接收到的报文是否是WATCHER_EVENT报文，如果是的话则调用这个watch的所有回调函数（deliverWatchers）进行执行；否则就是其它响应报文，则根据completion_type调用相应的completion回调函数进行处理。