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;原文地址：http://blog.51cto.com/9681602/2160503
时间： 2024-10-12 21:29:03