From 7d38f16674f1bc7071d08f4d4dad68fc3ffad965 Mon Sep 17 00:00:00 2001 From: Pawel Wodkowski Date: Thu, 18 Oct 2018 15:53:31 +0200 Subject: [PATCH] jsonrpc client: non-blocking mode This patch enables non blocking mode in RPC client. Requests are send and received during spdk_jsonrpc_client_poll. Change-Id: I5089737b2407055d3eeddb5e2ab0946d74e43c6a Signed-off-by: Pawel Wodkowski Reviewed-on: https://review.gerrithub.io/430095 Chandler-Test-Pool: SPDK Automated Test System Reviewed-by: Darek Stojaczyk Reviewed-by: Xiaodong Liu Reviewed-by: Shuhei Matsumoto Reviewed-by: Ben Walker Tested-by: SPDK CI Jenkins --- include/spdk/jsonrpc.h | 15 ++- lib/jsonrpc/jsonrpc_client.c | 5 +- lib/jsonrpc/jsonrpc_client_tcp.c | 199 +++++++++++++++++++++--------- lib/jsonrpc/jsonrpc_internal.h | 1 + test/rpc_client/rpc_client_test.c | 10 +- 5 files changed, 164 insertions(+), 66 deletions(-) diff --git a/include/spdk/jsonrpc.h b/include/spdk/jsonrpc.h index 9ae7f942ef..bb9dda4973 100644 --- a/include/spdk/jsonrpc.h +++ b/include/spdk/jsonrpc.h @@ -261,16 +261,25 @@ int spdk_jsonrpc_client_send_request(struct spdk_jsonrpc_client *client, struct spdk_jsonrpc_client_request *req); /** - * Receive the JSON-RPC response in JSON-RPC client. + * Poll the JSON-RPC client. When any response is available use + * \c spdk_jsonrpc_client_get_response to retrieve it. * * This function is not thread safe and should only be called from one thread at * a time while no other threads are actively \c client object. * * \param client JSON-RPC client. + * \param timeout Time in miliseconds this function will block. -1 block forever, 0 don't block. * - * \return 0 on success. + * \return If no error occurred, this function returns a non-negative number indicating how + * many ready responses can be retrieved. If an error occurred, this function returns one of + * the following negated errno values: + * -ENOTCONN - not connected yet. Try again later. + * -EINVAL - response is detected to be invalid. Client connection should be terminated. + * -ENOSPC - no space to receive another response. User need to retrieve waiting responses. + * -EIO - connection terminated (or other critical error). Client connection should be terminated. + * -ENOMEM - out of memory */ -int spdk_jsonrpc_client_recv_response(struct spdk_jsonrpc_client *client); +int spdk_jsonrpc_client_poll(struct spdk_jsonrpc_client *client, int timeout); /** * Return JSON RPC response object representing next available response from client connection. diff --git a/lib/jsonrpc/jsonrpc_client.c b/lib/jsonrpc/jsonrpc_client.c index f6dd34eccb..70bd89c96f 100644 --- a/lib/jsonrpc/jsonrpc_client.c +++ b/lib/jsonrpc/jsonrpc_client.c @@ -89,7 +89,7 @@ spdk_jsonrpc_parse_response(struct spdk_jsonrpc_client *client) /* Check to see if we have received a full JSON value. */ rc = spdk_json_parse(client->recv_buf, client->recv_offset, NULL, 0, &end, 0); if (rc == SPDK_JSON_PARSE_INCOMPLETE) { - return -EAGAIN; + return 0; } SPDK_DEBUGLOG(SPDK_LOG_RPC_CLIENT, "JSON string is :\n%s\n", client->recv_buf); @@ -145,8 +145,7 @@ spdk_jsonrpc_parse_response(struct spdk_jsonrpc_client *client) } r->ready = 1; - - return 0; + return 1; err: client->resp = NULL; diff --git a/lib/jsonrpc/jsonrpc_client_tcp.c b/lib/jsonrpc/jsonrpc_client_tcp.c index 3a199a1709..d3638c6335 100644 --- a/lib/jsonrpc/jsonrpc_client_tcp.c +++ b/lib/jsonrpc/jsonrpc_client_tcp.c @@ -46,27 +46,30 @@ _spdk_jsonrpc_client_send_request(struct spdk_jsonrpc_client *client) return 0; } - /* Reset offset in request */ - request->send_offset = 0; - - while (request->send_len > 0) { + if (request->send_len > 0) { rc = send(client->sockfd, request->send_buf + request->send_offset, request->send_len, 0); - if (rc <= 0) { - if (rc < 0 && errno == EINTR) { + if (rc < 0) { + /* For EINTR we pretend that nothing was send. */ + if (errno == EINTR) { rc = 0; } else { - return rc; + rc = -errno; + SPDK_ERRLOG("poll() failed (%d): %s\n", errno, spdk_strerror(errno)); } + + return rc; } request->send_offset += rc; request->send_len -= rc; } - client->request = NULL; + if (request->send_len == 0) { + client->request = NULL; + spdk_jsonrpc_client_free_request(request); + } - spdk_jsonrpc_client_free_request(request); return 0; } @@ -93,15 +96,17 @@ recv_buf_expand(struct spdk_jsonrpc_client *client) } static int -_spdk_jsonrpc_client_poll(struct spdk_jsonrpc_client *client) +_spdk_jsonrpc_client_resp_ready_count(struct spdk_jsonrpc_client *client) { - ssize_t rc = 0; - size_t recv_avail; + return client->resp != NULL && client->resp->ready ? 1 : 0; +} - _spdk_jsonrpc_client_send_request(client); +static int +_spdk_jsonrpc_client_recv(struct spdk_jsonrpc_client *client) +{ + ssize_t rc; if (client->recv_buf == NULL) { - /* memory malloc for recv-buf */ client->recv_buf = malloc(SPDK_JSONRPC_SEND_BUF_SIZE_INIT); if (!client->recv_buf) { rc = errno; @@ -110,54 +115,119 @@ _spdk_jsonrpc_client_poll(struct spdk_jsonrpc_client *client) } client->recv_buf_size = SPDK_JSONRPC_SEND_BUF_SIZE_INIT; client->recv_offset = 0; - } - - recv_avail = client->recv_buf_size - client->recv_offset; - while (recv_avail > 0) { - rc = recv(client->sockfd, client->recv_buf + client->recv_offset, recv_avail - 1, 0); - if (rc < 0) { - if (errno == EINTR) { - continue; - } else { - return errno; - } - } else if (rc == 0) { - return -EIO; - } - - client->recv_offset += rc; - recv_avail -= rc; - - client->recv_buf[client->recv_offset] = '\0'; - - /* Check to see if we have received a full JSON value. */ - rc = spdk_jsonrpc_parse_response(client); - if (rc == 0) { - /* Successfully parsed response */ - return 0; - } else if (rc && rc != -EAGAIN) { - SPDK_ERRLOG("jsonrpc parse request failed\n"); + } else if (client->recv_offset == client->recv_buf_size - 1) { + rc = recv_buf_expand(client); + if (rc) { return rc; } + } - /* Expand receive buffer if larger one is needed */ - if (recv_avail == 1) { - rc = recv_buf_expand(client); - if (rc != 0) { - return rc; + rc = recv(client->sockfd, client->recv_buf + client->recv_offset, + client->recv_buf_size - client->recv_offset - 1, 0); + if (rc < 0) { + /* For EINTR we pretend that nothing was reveived. */ + if (errno == EINTR) { + return 0; + } else { + rc = -errno; + SPDK_ERRLOG("recv() failed (%d): %s\n", errno, spdk_strerror(errno)); + return rc; + } + } else if (rc == 0) { + return -EIO; + } + + client->recv_offset += rc; + client->recv_buf[client->recv_offset] = '\0'; + + /* Check to see if we have received a full JSON value. */ + return spdk_jsonrpc_parse_response(client); +} + +static int +_spdk_jsonrpc_client_poll(struct spdk_jsonrpc_client *client, int timeout) +{ + int rc; + struct pollfd pfd = { .fd = client->sockfd, .events = POLLIN | POLLOUT }; + + rc = poll(&pfd, 1, timeout); + if (rc == -1) { + if (errno == EINTR) { + /* For EINTR we pretend that nothing was received nor send. */ + rc = 0; + } else { + rc = -errno; + SPDK_ERRLOG("poll() failed (%d): %s\n", errno, spdk_strerror(errno)); + } + } else if (rc > 0) { + rc = 0; + + if (pfd.revents & POLLOUT) { + rc = _spdk_jsonrpc_client_send_request(client); + } + + if (rc == 0 && (pfd.revents & POLLIN)) { + rc = _spdk_jsonrpc_client_recv(client); + /* Incomplete message in buffer isn't an error. */ + if (rc == -EAGAIN) { + rc = 0; } - recv_avail = client->recv_buf_size - client->recv_offset; } } - return 0; + return rc ? rc : _spdk_jsonrpc_client_resp_ready_count(client); +} + +static int +_spdk_jsonrpc_client_poll_connecting(struct spdk_jsonrpc_client *client, int timeout) +{ + socklen_t rc_len; + int rc; + + struct pollfd pfd = { + .fd = client->sockfd, + .events = POLLOUT + }; + + rc = poll(&pfd, 1, timeout); + if (rc == 0) { + return -ENOTCONN; + } else if (rc == -1) { + if (errno != EINTR) { + SPDK_ERRLOG("poll() failed (%d): %s\n", errno, spdk_strerror(errno)); + goto err; + } + + /* We are still not connected. Caller will have to call us again. */ + return -ENOTCONN; + } else if (pfd.revents & ~POLLOUT) { + /* We only poll for POLLOUT */ + goto err; + } else if ((pfd.revents & POLLOUT) == 0) { + /* Is this even possible to get here? */ + return -ENOTCONN; + } + + rc_len = sizeof(int); + /* connection might fail so need to check SO_ERROR. */ + if (getsockopt(client->sockfd, SOL_SOCKET, SO_ERROR, &rc, &rc_len) == -1) { + goto err; + } + + if (rc == 0) { + client->connected = true; + return 0; + } + +err: + return -EIO; } static int _spdk_jsonrpc_client_connect(struct spdk_jsonrpc_client *client, int domain, int protocol, struct sockaddr *server_addr, socklen_t addrlen) { - int rc; + int rc, flags; client->sockfd = socket(domain, SOCK_STREAM, protocol); if (client->sockfd < 0) { @@ -166,13 +236,26 @@ _spdk_jsonrpc_client_connect(struct spdk_jsonrpc_client *client, int domain, int return -rc; } - rc = connect(client->sockfd, server_addr, addrlen); - if (rc != 0) { - SPDK_ERRLOG("could not connect to JSON-RPC server: %s\n", spdk_strerror(errno)); + flags = fcntl(client->sockfd, F_GETFL); + if (flags < 0 || fcntl(client->sockfd, F_SETFL, flags | O_NONBLOCK) < 0) { + rc = errno; + SPDK_ERRLOG("fcntl(): can't set nonblocking mode for socket (%d): %s\n", + errno, spdk_strerror(errno)); goto err; } - return 0; + rc = connect(client->sockfd, server_addr, addrlen); + if (rc != 0) { + rc = errno; + if (rc != EINPROGRESS) { + SPDK_ERRLOG("could not connect to JSON-RPC server: %s\n", spdk_strerror(errno)); + goto err; + } + } else { + client->connected = true; + } + + return -rc; err: close(client->sockfd); client->sockfd = -1; @@ -244,7 +327,7 @@ spdk_jsonrpc_client_connect(const char *addr, int addr_family) } err: - if (rc != 0) { + if (rc != 0 && rc != -EINPROGRESS) { free(client); client = NULL; errno = -rc; @@ -299,9 +382,13 @@ spdk_jsonrpc_client_free_request(struct spdk_jsonrpc_client_request *req) } int -spdk_jsonrpc_client_recv_response(struct spdk_jsonrpc_client *client) +spdk_jsonrpc_client_poll(struct spdk_jsonrpc_client *client, int timeout) { - return _spdk_jsonrpc_client_poll(client); + if (client->connected) { + return _spdk_jsonrpc_client_poll(client, timeout); + } else { + return _spdk_jsonrpc_client_poll_connecting(client, timeout); + } } int spdk_jsonrpc_client_send_request(struct spdk_jsonrpc_client *client, diff --git a/lib/jsonrpc/jsonrpc_internal.h b/lib/jsonrpc/jsonrpc_internal.h index 50f509bf9f..5f7b14ef37 100644 --- a/lib/jsonrpc/jsonrpc_internal.h +++ b/lib/jsonrpc/jsonrpc_internal.h @@ -116,6 +116,7 @@ struct spdk_jsonrpc_client_response_internal { struct spdk_jsonrpc_client { int sockfd; + bool connected; size_t recv_buf_size; size_t recv_offset; diff --git a/test/rpc_client/rpc_client_test.c b/test/rpc_client/rpc_client_test.c index 1814e78543..68ca1f6578 100644 --- a/test/rpc_client/rpc_client_test.c +++ b/test/rpc_client/rpc_client_test.c @@ -78,17 +78,19 @@ spdk_jsonrpc_client_check_rpc_method(struct spdk_jsonrpc_client *client, char *m spdk_jsonrpc_client_send_request(client, request); do { - rc = spdk_jsonrpc_client_recv_response(client); - } while (rc == -EAGAIN || rc == -ENOTCONN); + rc = spdk_jsonrpc_client_poll(client, 1); + } while (rc == 0 || rc == -ENOTCONN); - if (rc != 0) { + if (rc <= 0) { + SPDK_ERRLOG("Failed to get response: %d\n", rc); + rc = -1; goto out; } json_resp = spdk_jsonrpc_client_get_response(client); if (json_resp == NULL) { SPDK_ERRLOG("spdk_jsonrpc_client_get_response() failed\n"); - rc = -errno; + rc = -1; goto out; }