rpc框架yar之源码解析- 协议和传输

Yar 协议头和传输源码分析

 这篇博客主要学习rpc框架yar的协议头和传输的实现, 能力有限，有些语句没有看懂，所以猜测了一部分。

yar_header_t的实现

_yar_header的定义主要在yar_protocol.h和yar_protocol.c里面，下面介绍这两个文件里的源码。
借用网上的一幅图片，请求体包括 yar_header + packager_name + yar_request_t 这三个部分，返回类似。
下面主要介绍yar_header的部分。

yar_protocol.h

// 定义yar_header_t这个结构体的内容，一共82个字节。
typedef struct _yar_header {
    unsigned int   id;                        // 4
    unsigned short version;             // 2
    unsigned int   magic_num;        // 4
    unsigned int   reserved;            // 4
    unsigned char  provider[32];    // 32
    unsigned char  token[32];        // 32
    unsigned int   body_len;          //  4
}

// 下面是声明两个函数，稍后在.c文件中说明具体方法
yar_header_t * php_yar_protocol_parse(char *payload);
void php_yar_protocol_render(yar_header_t *header, uint id, char *provider, char *token, uint body_len, uint reserved);

yar_protocol.c

// 这个函数的作用应该是给 header指向的结构体赋值
/* htonl函数介绍：将主机数转换成无符号长整型的网络字节顺序。
 *                           本函数将一个32位数从主机字节顺序转换成网络字节顺序。
 */
void php_yar_protocol_render(yar_header_t *header, uint id, char *provider, char *token, uint body_len, uint reserved) /* {{{ */ {
    header->magic_num = htonl(YAR_PROTOCOL_MAGIC_NUM);
    header->id = htonl(id);
    header->body_len = htonl(body_len);
    header->reserved = htonl(reserved);
    if (provider) {
        memcpy(header->provider, provider, strlen(provider));
    }
    if (token) {
        memcpy(header->token, token, strlen(token));
    }
    return;
} /* }}} */

// 这个函数是将payload指向的字符串，转成header结构体对应的内容 然后返回。
yar_header_t * php_yar_protocol_parse(char *payload) /* {{{ */ {
    yar_header_t *header = (yar_header_t *)payload;

    header->magic_num = ntohl(header->magic_num);

    if (header->magic_num != YAR_PROTOCOL_MAGIC_NUM) {
        header->magic_num = htonl(header->magic_num);
        return NULL;
    }

    header->id = ntohl(header->id);
    header->body_len = ntohl(header->body_len);
    header->reserved = ntohl(header->reserved);

    return header;
} /* }}} */

Q： 下面的这句 我没有看明白, 感觉直接返回就可以了。

if (header->magic_num != YAR_PROTOCOL_MAGIC_NUM) {
        header->magic_num = htonl(header->magic_num);
        return NULL;
    }

yar传输源码分析

关于yar传输方面源码分析，还是先从下面的图开始吧。感谢作图的人，比我说的直观。

传输层主要涉及到如下几个文件：
transports/curl.c、transports/socket.c 、yar_transport.h、yar_transport.c 4个文件。

yar_transport.h

yar_transport.h 主要定义两个函数和一些结构体。

// 这个应该是传输方式的方法定义，后面会在curl.c socket.c中实现下面的内容
typedef struct _yar_transport_interface {
    void *data;
    int  (*open)(struct _yar_transport_interface *self, zend_string *address, long options, char **msg);
    int  (*send)(struct _yar_transport_interface *self, struct _yar_request *request, char **msg);
    struct _yar_response * (*exec)(struct _yar_transport_interface *self, struct _yar_request *request);
    int  (*setopt)(struct _yar_transport_interface *self, long type, void *value, void *addition);
    int  (*calldata)(struct _yar_transport_interface *self, yar_call_data_t *calldata);
    void (*close)(struct _yar_transport_interface *self);
} yar_transport_interface_t;

typedef struct _yar_transport {
    const char *name;
    struct _yar_transport_interface * (*init)();
    void (*destroy)(yar_transport_interface_t *self);
    yar_transport_multi_t *multi;
} yar_transport_t;

// 下面是并行调用时的结构体定义
typedef struct _yar_transport_multi_interface {
    void *data;
    int (*add)(struct _yar_transport_multi_interface *self, yar_transport_interface_t *cp);
    int (*exec)(struct _yar_transport_multi_interface *self, yar_concurrent_client_callback *callback);
    void (*close)(struct _yar_transport_multi_interface *self);
} yar_transport_multi_interface_t;

typedef struct _yar_transport_multi {
    struct _yar_transport_multi_interface * (*init)();
} yar_transport_multi_t;

// 根据名称获得对应的传输方式，curl / sock
PHP_YAR_API const yar_transport_t * php_yar_transport_get(char *name, int nlen);

// 注册传输方式到yar_transports_list
PHP_YAR_API int php_yar_transport_register(const yar_transport_t *transport);

yar_transport.c

这个文件主要实现上面两个函数，即注册curl和socket两种方式，然后根据name获得传输方式。

// 定义list
struct _yar_transports_list {
    unsigned int size;
    unsigned int num;
    const yar_transport_t **transports;
} yar_transports_list;

// 注册方法到list里面
PHP_YAR_API int php_yar_transport_register(const yar_transport_t *transport) /* {{{ */ {

    if (!yar_transports_list.size) {
       yar_transports_list.size = 5;
       yar_transports_list.transports = (const yar_transport_t **)malloc(sizeof(yar_transport_t *) * yar_transports_list.size);
    } else if (yar_transports_list.num == yar_transports_list.size) {
       yar_transports_list.size += 5;
       yar_transports_list.transports = (const yar_transport_t **)realloc(yar_transports_list.transports, sizeof(yar_transport_t *) * yar_transports_list.size);
    }
    yar_transports_list.transports[yar_transports_list.num] = transport;

    return yar_transports_list.num++;
} /* }}} */

// 根据name获得方式
PHP_YAR_API const yar_transport_t * php_yar_transport_get(char *name, int nlen) /* {{{ */ {
    int i = 0;
    for (;i<yar_transports_list.num;i++) {
        if (strncmp(yar_transports_list.transports[i]->name, name, nlen) == 0) {
            return yar_transports_list.transports[i];
        }
    }

    return NULL;
} /* }}} */

// 下面这两句没有看懂什么意思
le_calldata = zend_register_list_destructors_ex(php_yar_calldata_dtor, NULL, "Yar Call Data", module_number);
le_plink = zend_register_list_destructors_ex(NULL, php_yar_plink_dtor, "Yar Persistent Link", module_number);

transports/socket.c

主要是socket方式需要的方法的具体实现, 先看下面方法
用yar_transport_t 定义一个常量

const yar_transport_t yar_transport_socket = {
    "sock",
    php_yar_socket_init,
    php_yar_socket_destroy,
    NULL
}; /* }}} */

php_yar_socket_init的定义如下

yar_transport_interface_t * php_yar_socket_init() /* {{{ */ {
    yar_socket_data_t *data;
    yar_transport_interface_t *self;

    self = emalloc(sizeof(yar_transport_interface_t));
    self->data = data = ecalloc(1, sizeof(yar_socket_data_t));

    self->open      = php_yar_socket_open;
    self->send      = php_yar_socket_send;
    self->exec      = php_yar_socket_exec;
    self->setopt    = php_yar_socket_setopt;
    self->calldata  = NULL;
    self->close     = php_yar_socket_close;

    return  self;
} /* }}} */

然后我们再看yar_socket_data_t的定义, 其实主要使用php_stream这个流类型 底层的具体实现不追了。

typedef struct _yar_socket_data_t {
    char persistent;
    php_stream *stream;  // 主要
} yar_socket_data_t;

下面是php_yar_socket_open函数的实现

php_stream *stream = NULL;
...
// 这个是核心代码，应该是根据地址和配置创建一个流的对象。
stream = php_stream_xport_create(ZSTR_VAL(address), ZSTR_LEN(address), 0, STREAM_XPORT_CLIENT|STREAM_XPORT_CONNECT, persistent_key, &tv, NULL, &errstr, &err);

php_yar_socket_close, 这个函数就是关闭流，然后释放空间

if (!data->persistent && data->stream) {
        php_stream_close(data->stream);
    }

php_yar_socket_send 这个函数主要是把需要发送的内容，按照一定格式pack，然后放到stream中。
里面有几个函数没有看懂，以后在详细查找：

// 先构造header的内容
php_yar_protocol_render(&header, request->id, "Yar PHP Client", NULL, ZSTR_LEN(payload), data->persistent? YAR_PROTOCOL_PERSISTENT : 0);

// 复制到buf这个变量中
memcpy(buf, (char *)&header, sizeof(yar_header_t));

// 这个地方 我不太了解为什么要用goto
wait_io:
        if (bytes_left) {
            retval = php_select(fd+1, NULL, &rfds, NULL, &tv);

            if (retval == -1) {
                zend_string_release(payload);
                spprintf(msg, 0, "select error '%s'", strerror(errno));
                return 0;
            } else if (retval == 0) {
                zend_string_release(payload);
                spprintf(msg, 0, "select timeout %ldms reached", YAR_G(timeout));
                return 0;
            }

            if (PHP_SAFE_FD_ISSET(fd, &rfds)) {
                if ((ret = php_stream_xport_sendto(data->stream, ZSTR_VAL(payload) + bytes_sent, bytes_left, 0, NULL, 0)) > 0) {
                    bytes_left -= ret;
                    bytes_sent += ret;
                }
            }
            goto wait_io;
        }
    }

transports/curl.c

curl.c跟socket类似，依赖curl/curl.h文件里的内容，如

CURL        *cp;