Glusterfs之rpc模块源码分析(中）之Glusterfs的rpc模块实现（2）

第二节、rpc客户端实现原理及代码分析

rpc客户端主要发起一个rpc请求，执行完rpc请求以后就退出rpc，下面分析客户端rpc请求建立的整个过程。Rpc客户端请求建立的第一步是执行cli_rpc_init函数，主要实现代码如下：

 1 this = THIS;//取得本线程的xlator列表  2  3 cli_rpc_prog = &cli_prog;//设置rpc调用过程集合（许多函数）  4  5 options = dict_new ();//新建一个字典数据结构用于存放选项信息  6  7 ret = dict_set_str (options, "remote-host", state->remote_host);//设置host  8  9 if (state->remote_port) 10 11 port = state->remote_port; 12 13 ret = dict_set_int32 (options, "remote-port", port);//设置端口号 14 15 ret = dict_set_str (options, "transport.address-family", "inet");//设置协议族为inet 16 17 rpc = rpc_clnt_new (options, this->ctx, this->name);//新建一个rpc客户端对象 18 19 ret = rpc_clnt_register_notify (rpc, cli_rpc_notify, this);//注册rpc请求通知函数 20 21 rpc_clnt_start (rpc);//开始rpc

 

这段代码其实是glusterfs客户端程序启动时建立rpc请求的初始化过程函数，真正独立开始建立一个rpc请求的过程是从函数rpc_clnt_new开始的，下面就分析这个函数的功能，先看主要代码：

 1 rpc = GF_CALLOC (1, sizeof (*rpc), gf_common_mt_rpcclnt_t);//为客户端rpc对象分配内存  2  3 pthread_mutex_init (&rpc->lock, NULL);//初始化锁  4  5 rpc->ctx = ctx;//属于哪一个ctx  6  7 //新建请求内存池  8  9 rpc->reqpool = mem_pool_new (struct rpc_req, RPC_CLNT_DEFAULT_REQUEST_COUNT); 10 11 //保存帧数据的内存池 12 13 rpc->saved_frames_pool = mem_pool_new (struct saved_frame, 14 15  RPC_CLNT_DEFAULT_REQUEST_COUNT); 16 17 ret = rpc_clnt_connection_init (rpc, ctx, options, name);//初始化rpc请求连接 18 19 rpc = rpc_clnt_ref (rpc);//客户端对象引用计数加1 20 21 INIT_LIST_HEAD (&rpc->programs);//初始化程序集的链表

 

rpc客户端发送请求也需要装载相应的协议库，它主要使用协议库里面的初始化函数和链接函数，上面代码中rpc_clnt_connection_init函数主要完成协议库装载功能和链接选项的一些初始化功能，具体实现的主要代码如下：

 1 pthread_mutex_init (&clnt->conn.lock, NULL);//初始化锁  2  3 conn->trans = rpc_transport_load (ctx, options, name);//装载协议库并执行初始化init  4  5 rpc_transport_ref (conn->trans);//增加传输层的引用计数  6  7 conn->rpc_clnt = clnt;//连接对象属于哪一个客户端对象  8  9 ret = rpc_transport_register_notify (conn->trans, rpc_clnt_notify, conn);//注册通知函数 10 11 conn->saved_frames = saved_frames_new ();//新建一个保存帧数据的对象

 

通过上面代码执行以后基本的初始化工作已经完毕，下一步就是建立于rpc服务的链接，此功能在函数rpc_clnt_start中实现，也就是在所有初始化工作完成以后调用此函数发起链接，主要代码如下：

1 rpc_clnt_reconnect (conn->trans);

 

继续跟踪函数rpc_clnt_reconnect：

  

 1 pthread_mutex_lock (&conn->lock);//初始化锁  2  3  {  4  5 if (conn->reconnect)//如果重新链接正在进行  6  7 gf_timer_call_cancel (clnt->ctx, conn->reconnect);//取消正在的链接  8  9 conn->reconnect = 0;//初始化为0 10 11 12 13 if (conn->connected == 0) {//还没有完成链接 14 15 tv.tv_sec = 3;//时间三秒 16 17 //发起传输层的链接 18 19 ret = rpc_transport_connect (trans, conn->config.remote_port); 20 21 //设置重链接对象 22 23 conn->reconnect = gf_timer_call_after (clnt->ctx, tv, 24 25  rpc_clnt_reconnect, trans); 26 27  } 28 29  } 30 31 pthread_mutex_unlock (&conn->lock);//解锁 32 33 if ((ret == -1) && (errno != EINPROGRESS) && (clnt->notifyfn)) { 34 35 //建立链接失败就通知rpc客户端对象（调用通知函数） 36 37 clnt->notifyfn (clnt, clnt->mydata, RPC_CLNT_DISCONNECT, NULL); 38 39 }

 

真正的链接是在具体的协议中的链接函数中执行，下面以tcp为例，看看它的链接函数的实现，主要代码如下：

 1 //得到客户端协议族相关信息  2  3 ret = socket_client_get_remote_sockaddr (this, SA (&sockaddr), &sockaddr_len, &sa_family)  4  5 if (port > 0)  6  7 ((struct sockaddr_in *) (&sockaddr))->sin_port = htons (port);//端口字节序转换  8  9 pthread_mutex_lock (&priv->lock); 10 11  { 12 13 memcpy (&this->peerinfo.sockaddr, &sockaddr, sockaddr_len);//赋值sockaddr信息 14 15 this->peerinfo.sockaddr_len = sockaddr_len;//地址长度保存 16 17 priv->sock = socket (sa_family, SOCK_STREAM, 0);//创建socket 18 19 setsockopt (priv->sock, SOL_SOCKET, SO_RCVBUF, &priv->windowsize, 20 21 sizeof (priv->windowsize)) < 0) ;//设置接收的系统缓冲区 22 23 setsockopt (priv->sock, SOL_SOCKET, SO_SNDBUF, &priv->windowsize, 24 25 sizeof (priv->windowsize)) < 0);//发送缓冲区设置 26 27 if (priv->nodelay) { 28 29 ret = __socket_nodelay (priv->sock);//设置是否延迟发送（等待一个完整的） 30 31  } 32 33 if (!priv->bio) { 34 35 ret = __socket_nonblock (priv->sock);//设置非阻塞 36 37  } 38 39 if (priv->keepalive) { 40 41 ret = __socket_keepalive (priv->sock, priv->keepaliveintvl, 42 43 priv->keepaliveidle);//保存链接活动 44 45  } 46 47 SA (&this->myinfo.sockaddr)->sa_family = 48 49 SA (&this->peerinfo.sockaddr)->sa_family;//保存协议族信息 50 51 ret = client_bind (this, SA (&this->myinfo.sockaddr),//根据协议族做适当的绑定 52 53 &this->myinfo.sockaddr_len, priv->sock); 54 55 ret = connect (priv->sock, SA (&this->peerinfo.sockaddr),//发起链接请求 56 57 this->peerinfo.sockaddr_len); 58 59 priv->connected = 0; 60 61 rpc_transport_ref (this);//传输对象引用加1 62 63 priv->idx = event_register (ctx->event_pool, priv->sock,//注册epoll读写事件 64 65 socket_event_handler, this, 1, 1); 66 67  } 68 69 pthread_mutex_unlock (&priv->lock);//解锁

 

整个链接过程的建立需要注意一点的就是会根据协议族的类型做适当的绑定，当发起链接以后就开始注册各种读写事件。处理这些事件发生的函数是socket_event_handler，主要代码如下：

 

 1 THIS = this->xl;//取得xlator  2  3 priv = this->private;//取得私有数据  4  5 pthread_mutex_lock (&priv->lock);//加锁  6  7  {  8  9 priv->idx = idx;//取得索引下表（对应的socket） 10 11  } 12 13 pthread_mutex_unlock (&priv->lock); 14 15 if (!priv->connected) {//如果链接还没有建立先完成链接的建立 16 17 ret = socket_connect_finish (this);//完成链接建立 18 19  } 20 21 if (!ret && poll_out) {//可写事件处理 22 23 ret = socket_event_poll_out (this); 24 25  } 26 27 if (!ret && poll_in) {//可读事件处理 28 29 ret = socket_event_poll_in (this); 30 31 }

 

到此客户端rpc请求过程完全建立，当真正的发送一个rpc请求的时候就会响应相应的epoll的写事件，把包装好的数据帧发送到rpc服务器端，rpc客户端也会通过可读事件来接收rpc服务器端的响应信息。

总结：同rpc服务器端一眼，rpc客户端的建立也是比较复杂的过程，还是通过流程图加描述来展示一下整个rpc客户端的初始化过程，图如下：