Implementatin of a searpc transport based on named pipe

前言

经过之前的分析，已经分析完成了searpc服务器、客户端以及二者连接的基本原理。接下来对searpc-name-pipe-transport文件中的基于命名管道的searpc传输的应用进行分析。

named-pipe

管道是一个有两端的对象。一个进程向管道写入信息，而另一个进程从管道读取信息。其本质是用于进程间通信的共享内存区域，确切的说是线程间通信方法。

命名管道是全双工的，且支持网络通信。创建管道的进程称为管道服务器，连接到这个管道的进程成为管道客户端。命名管道支持多客户端连接，支持双向通信。

searpc-named-pipe-transport.h

首先定义searpc命名管道服务器端的接口，及其构造方法

struct _SearpcNamedPipeServer {
    char path[4096];
    pthread_t listener_thread;
    SearpcNamedPipe pipe_fd;
    GThreadPool *named_pipe_server_thread_pool;
};
​
typedef struct _SearpcNamedPipeServer LIBSEARPC_API SearpcNamedPipeServer;
​
LIBSEARPC_API
SearpcNamedPipeServer* searpc_create_named_pipe_server(const char *path);
​
LIBSEARPC_API
SearpcNamedPipeServer* searpc_create_named_pipe_server_with_threadpool(const char *path, int named_pipe_server_thread_pool_size);
​
LIBSEARPC_API
int searpc_named_pipe_server_start(SearpcNamedPipeServer *server);

• pthread_t listener_thread用于创建一个监听可能的连接的线程

• char path[4096]与SearpcNamedPipe pipe_fd用于存放管道服务器的路径与管道号

• GThreadPool *named_pipe_server_thread_pool是一个线程池，其作用是调用rpc函数

然后

定义searpc命名管道客户端的接口，及其构造方法

struct _SearpcNamedPipeClient {
    char path[4096];
    SearpcNamedPipe pipe_fd;
};
​
typedef struct _SearpcNamedPipeClient LIBSEARPC_API SearpcNamedPipeClient;
​
LIBSEARPC_API
SearpcNamedPipeClient* searpc_create_named_pipe_client(const char *path);
​
LIBSEARPC_API
SearpcClient * searpc_client_with_named_pipe_transport(SearpcNamedPipeClient *client, const char *service);
​
LIBSEARPC_API
int searpc_named_pipe_client_connect(SearpcNamedPipeClient *client);
​
LIBSEARPC_API
void searpc_free_client_with_pipe_transport (SearpcClient *client);

其成员属性同上

searpc-named-pipe-transport-c

创建SearpcClient

创建结构体，用于在客户端向管道中传输数据

typedef struct {
    SearpcNamedPipeClient* client;
    char *service;
} ClientTransportData;

其属性包含searpc命名管道客户端SearpcNamedPipeClient与服务名service

SearpcClient*
searpc_client_with_named_pipe_transport(SearpcNamedPipeClient *pipe_client,
                                        const char *service)
{
    SearpcClient *client= searpc_client_new();
    client->send = searpc_named_pipe_send;
​
    ClientTransportData *data = g_malloc(sizeof(ClientTransportData));
    data->client = pipe_client;
    data->service = g_strdup(service);
​
    client->arg = data;
    return client;
}
​
SearpcNamedPipeClient* searpc_create_named_pipe_client(const char *path)
{
    SearpcNamedPipeClient *client = g_malloc0(sizeof(SearpcNamedPipeClient));
    memcpy(client->path, path, strlen(path) + 1);
    return client;
}

上述两个方法用于创建一个searpc-client，但是其原理并非是与searpc-demo-client.c中相同（基于socket），而是基于named-pipe。

在此回顾SearpcClient的定义

struct _SearpcClient {
    TransportCB send;
    void *arg;
    
    AsyncTransportSend async_send;
    void *async_arg;
};

其中void *arg属性在之前的实现中为传输用socket的sockfd，因此我们得知，其类型之所以为void *是为了应对基于不同传输方式的实现。

在searpc_client_with_named_pipe_transport方法中，指定了SearpcClient的传输函数与管道客户端（等价于sockfd）；在searpc_create_named_pipe_client，方法中创建了SearpcNamedPipeClient，并指定了地址。

传输函数

然后讨论传输函数searpc_named_pipe_send。

char *searpc_named_pipe_send(void *arg, const gchar *fcall_str,
                             size_t fcall_len, size_t *ret_len)
{
    /* g_debug ("searpc_named_pipe_send is called\n"); */
    ClientTransportData *data = arg;
    SearpcNamedPipeClient *client = data->client;
​
    char *json_str = request_to_json(data->service, fcall_str, fcall_len);
    guint32 len = (guint32)strlen(json_str);
​
    if (pipe_write_n(client->pipe_fd, &len, sizeof(guint32)) < 0) {
        g_warning("failed to send rpc call: %s\n", strerror(errno));
        free (json_str);
        return NULL;
    }
​
    if (pipe_write_n(client->pipe_fd, json_str, len) < 0) {
        g_warning("failed to send rpc call: %s\n", strerror(errno));
        free (json_str);
        return NULL;
    }
​
    free (json_str);
​
    if (pipe_read_n(client->pipe_fd, &len, sizeof(guint32)) < 0) {
        g_warning("failed to read rpc response: %s\n", strerror(errno));
        return NULL;
    }
​
    char *buf = g_malloc(len);
​
    if (pipe_read_n(client->pipe_fd, buf, len) < 0) {
        g_warning("failed to read rpc response: %s\n", strerror(errno));
        g_free (buf);
        return NULL;
    }
​
    *ret_len = len;
    return buf;
}

我们已经知道了传输函数的作用是，将请求数据发送到服务端，并从服务端接收返回数据。其参数列表及各参数作用与searpc-client-demo中相同。

首先将传入的参数void *arg转化为ClientTransportData，并获取到NamedPipeClient；然后根据data->service、fcall_str、fcall_len创建json字符串

在这里让我疑惑的是，之前在searpc-demo-client中遇到的传输函数，并未指定service名。查询传输函数的声明，也和被调用rpc函数的服务名无关

typedef char *(*TransportCB)(void *arg, const gchar *fcall_str,
                             size_t fcall_len, size_t *ret_len);

而在服务器调用rpc函数的方法中，需要传入服务名，并通过它来查找rpc函数

searpc_server_call_function (const char *svc_name,
                             gchar *func, gsize len, gsize *ret_len)

经过查看searpc-demo-server中的实现，发现是在调用时由程序员指定的服务名。

在此之后调用pipe_write_n想管道客户端的pipe_fd写入上述json字符串；然后调用pipe_read_n从管道读入返回值，并写入到内存，最后设置返回值长度并返回。

request_to_json

本方法的作用是将服务名、被调用rpc函数的参数封装进json字符串中

request_to_json (const char *service, const char *fcall_str, size_t fcall_len)
{
    json_t *object = json_object ();
​
    char *temp_request = g_malloc0(fcall_len + 1);
    memcpy(temp_request, fcall_str, fcall_len);
​
    json_object_set_string_member (object, "service", service);
    json_object_set_string_member (object, "request", temp_request);
​
    g_free (temp_request);
​
    char *str = json_dumps (object, 0);
    json_decref (object);
    return str;
}

json_object_set_string_member

向json对象中设置键值对

static void json_object_set_string_member (json_t *object, const char *key, const char *value)
{
    json_object_set_new (object, key, json_string (value));
}

pipe_write_n&pipe_read_n

与earpc-demo-server中的writen与readn方法相同