专栏首页技巅Glusterfs之rpc模块源码分析(中)之Glusterfs的rpc模块实现(2)

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

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

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

        this = THIS;//取得本线程的xlator列表

        cli_rpc_prog = &cli_prog;//设置rpc调用过程集合(许多函数)

        options = dict_new ();//新建一个字典数据结构用于存放选项信息

        ret = dict_set_str (options, “remote-host”, state->remote_host);//设置host

        if (state->remote_port)

                port = state->remote_port;

        ret = dict_set_int32 (options, “remote-port”, port);//设置端口号

        ret = dict_set_str (options, “transport.address-family”, “inet”);//设置协议族为inet

        rpc = rpc_clnt_new (options, this->ctx, this->name);//新建一个rpc客户端对象

        ret = rpc_clnt_register_notify (rpc, cli_rpc_notify, this);//注册rpc请求通知函数

        rpc_clnt_start (rpc);//开始rpc

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

        rpc = GF_CALLOC (1, sizeof (*rpc), gf_common_mt_rpcclnt_t);//为客户端rpc对象分配内存

        pthread_mutex_init (&rpc->lock, NULL);//初始化锁

        rpc->ctx = ctx;//属于哪一个ctx

//新建请求内存池

        rpc->reqpool = mem_pool_new (struct rpc_req, RPC_CLNT_DEFAULT_REQUEST_COUNT);

        //保存帧数据的内存池

rpc->saved_frames_pool = mem_pool_new (struct saved_frame,

                                              RPC_CLNT_DEFAULT_REQUEST_COUNT);

        ret = rpc_clnt_connection_init (rpc, ctx, options, name);//初始化rpc请求连接

        rpc = rpc_clnt_ref (rpc);//客户端对象引用计数加1

        INIT_LIST_HEAD (&rpc->programs);//初始化程序集的链表

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

  pthread_mutex_init (&clnt->conn.lock, NULL);//初始化锁

conn->trans = rpc_transport_load (ctx, options, name);//装载协议库并执行初始化init

        rpc_transport_ref (conn->trans);//增加传输层的引用计数

        conn->rpc_clnt = clnt;//连接对象属于哪一个客户端对象

        ret = rpc_transport_register_notify (conn->trans, rpc_clnt_notify, conn);//注册通知函数

conn->saved_frames = saved_frames_new ();//新建一个保存帧数据的对象

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

rpc_clnt_reconnect (conn->trans);

继续跟踪函数rpc_clnt_reconnect:

        pthread_mutex_lock (&conn->lock);//初始化锁

        {

                if (conn->reconnect)//如果重新链接正在进行

                        gf_timer_call_cancel (clnt->ctx, conn->reconnect);//取消正在的链接

                conn->reconnect = 0;//初始化为0

                if (conn->connected == 0) {//还没有完成链接

                        tv.tv_sec = 3;//时间三秒

//发起传输层的链接

                        ret = rpc_transport_connect (trans, conn->config.remote_port);

//设置重链接对象

                        conn->reconnect = gf_timer_call_after (clnt->ctx, tv,

                                                     rpc_clnt_reconnect, trans);

                } 

        }

        pthread_mutex_unlock (&conn->lock);//解锁

        if ((ret == -1) && (errno != EINPROGRESS) && (clnt->notifyfn)) {

//建立链接失败就通知rpc客户端对象(调用通知函数)

                clnt->notifyfn (clnt, clnt->mydata, RPC_CLNT_DISCONNECT, NULL);

        }

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

//得到客户端协议族相关信息

        ret = socket_client_get_remote_sockaddr (this, SA (&sockaddr), &sockaddr_len, &sa_family)

        if (port > 0)

                ((struct sockaddr_in *) (&sockaddr))->sin_port = htons (port);//端口字节序转换

        pthread_mutex_lock (&priv->lock);

        {

                memcpy (&this->peerinfo.sockaddr, &sockaddr, sockaddr_len);//赋值sockaddr信息

                this->peerinfo.sockaddr_len = sockaddr_len;//地址长度保存

                priv->sock = socket (sa_family, SOCK_STREAM, 0);//创建socket

                setsockopt (priv->sock, SOL_SOCKET, SO_RCVBUF, &priv->windowsize,

                                sizeof (priv->windowsize)) < 0) ;//设置接收的系统缓冲区

                setsockopt (priv->sock, SOL_SOCKET, SO_SNDBUF, &priv->windowsize,

                                sizeof (priv->windowsize)) < 0);//发送缓冲区设置

                if (priv->nodelay) {

                        ret = __socket_nodelay (priv->sock);//设置是否延迟发送(等待一个完整的)

                 }

                if (!priv->bio) {

                        ret = __socket_nonblock (priv->sock);//设置非阻塞

                }

                if (priv->keepalive) {

                        ret = __socket_keepalive (priv->sock, priv->keepaliveintvl,

                                                  priv->keepaliveidle);//保存链接活动

                }

                SA (&this->myinfo.sockaddr)->sa_family =

                        SA (&this->peerinfo.sockaddr)->sa_family;//保存协议族信息

                ret = client_bind (this, SA (&this->myinfo.sockaddr),//根据协议族做适当的绑定

                                   &this->myinfo.sockaddr_len, priv->sock);

                ret = connect (priv->sock, SA (&this->peerinfo.sockaddr),//发起链接请求

                               this->peerinfo.sockaddr_len);

priv->connected = 0;

                rpc_transport_ref (this);//传输对象引用加1

                priv->idx = event_register (ctx->event_pool, priv->sock,//注册epoll读写事件

                                            socket_event_handler, this, 1, 1);

        }

        pthread_mutex_unlock (&priv->lock);//解锁

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

  THIS = this->xl;//取得xlator

        priv = this->private;//取得私有数据

        pthread_mutex_lock (&priv->lock);//加锁

        {

                priv->idx = idx;//取得索引下表(对应的socket)

        }

        pthread_mutex_unlock (&priv->lock);

        if (!priv->connected) {//如果链接还没有建立先完成链接的建立

                ret = socket_connect_finish (this);//完成链接建立

        }

        if (!ret && poll_out) {//可写事件处理

                ret = socket_event_poll_out (this);

        }

        if (!ret && poll_in) {//可读事件处理

                ret = socket_event_poll_in (this);

        }

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

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

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Glusterfs之rpc模块源码分析(中)之Glusterfs的rpc模块实现(3)

    技巅
  • Glusterfs之rpc模块源码分析(中)之Glusterfs的rpc模块实现(1)

    技巅
  • 一个大型互联网公司高级技术的远程面试题目和解答过程以及源代码

    技巅
  • Glusterfs之rpc模块源码分析(中)之Glusterfs的rpc模块实现(3)

    技巅
  • 使用SSH命令行远程登录运行在CloudFoundry上的应用

    版权声明:本文为博主汪子熙原创文章,未经博主允许不得转载。 https://jerry.blog....

    Jerry Wang
  • Python生成html邮件

    最近在学习Python,这是我第一个Python脚本。没有技术含量,之前不知道zip()函数,想让2个结果对应起来,弄了好久,后来问了做php的同事,才知道。

    py3study
  • 当代码无法运行的时候,我在想什么?

    我经常被问的一句话就是:为什么代码无法运行?然后细看有些问题,真是让我哭笑不得,比如no module name pygame……

    短短的路走走停停
  • Thymeleaf语法详解

    1.2 判断字符串是否为空 Thymeleaf 内置对象 注意语法: a.调用内置对象一定要用# b.大部分的内置对象都以 s 结尾 strings、n...

    用户4919348
  • 低代码开发平台,能让 CIO 们“快活”吗?

    自从 Forrester 创造了“低代码开发平台”术语以来,每年都会对低代码开发平台领域做出报告分析,根据 Forrester 的报告,低代码开发平台市场将从2...

    用户1564362
  • 每个程序员都需要有自己的图床

    重要的事情说三遍,免费图库看似让你捡了个便宜,实际就是一个大坑。因为你不知道什么时候,免费图床就挂掉了。而你之前引用的所有图片,都变成了404或403了,如果只...

    章鱼喵

扫码关注云+社区

领取腾讯云代金券