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 条评论
登录 后参与评论

相关文章

来自专栏性能与架构

腾讯前端首屏优化案例

下面是对腾讯前端团队优化手Q一个页面案例的总结 优化目标页面:手Q群成员分布的页面 ? 左面是首屏,右面是下拉后到底部 这个页面中,可以划分成四个部份:活跃...

37110
来自专栏北京马哥教育

Shell的18条常用命令整理

? 作者 | junmail 来源 | CSDN社区 云豆贴心提醒,本文阅读时间8分钟,文末有秘密! 1、 ls: 类似于dos下的dir命令 ...

3974
来自专栏点滴积累

python基础操作以及hdfs操作

一、前言        作为一个全栈工程师,必须要熟练掌握各种语言。。。HelloWorld。最近就被“逼着”走向了python开发之路,大体实现的功能是写一...

3968
来自专栏磨磨谈

从ceph对象中提取RBD中的指定文件

之前有个想法,是不是有办法找到rbd中的文件与对象的关系,想了很久但是一直觉得文件系统比较复杂,在fs 层的东西对ceph来说是透明的,并且对象大小是4M,而文...

1072
来自专栏along的开发之旅

程序集版本号,文件版本号及发布版本号管理

在.net框架中和CLR运行时钟引用,如使用程序集的强命名时就会使用. 按照我的理解,你所使用的库如log4net,常见的有2.0和4.0的,那么你下载的库...

1351
来自专栏娱乐心理测试

duplicate symbols for architecture arm64的问题结决方法

2404
来自专栏程序员互动联盟

【专业技术】揭秘安卓浏览器如何注入javascript脚本

Android中向webview注入js代码可以通过webview.loadUrl("javascript:xxx")来实现,然后就会执行javascript后...

5124
来自专栏晓晨的专栏

IdentityServer Topics(2)- 定义资源

1083
来自专栏码农二狗

一个简单的dao设计

1347
来自专栏维C果糖

史上最简单的 MySQL 教程(四十二)「代码执行结构」

在 MySQL 中,没有for循环,仅有while循环、loop循环和repeat循环,呃,还有一种非标准的goto循环,在此我们仅介绍while循环,其基本语...

4497

扫码关注云+社区

领取腾讯云代金券