xref: /spdk/lib/jsonrpc/jsonrpc_server_tcp.c (revision 22898a91b9b6f289933db19b0175821cfb7e7820) 
1 /*-
2  *   BSD LICENSE
3  *
4  *   Copyright (c) Intel Corporation.
5  *   All rights reserved.
6  *
7  *   Redistribution and use in source and binary forms, with or without
8  *   modification, are permitted provided that the following conditions
9  *   are met:
10  *
11  *     * Redistributions of source code must retain the above copyright
12  *       notice, this list of conditions and the following disclaimer.
13  *     * Redistributions in binary form must reproduce the above copyright
14  *       notice, this list of conditions and the following disclaimer in
15  *       the documentation and/or other materials provided with the
16  *       distribution.
17  *     * Neither the name of Intel Corporation nor the names of its
18  *       contributors may be used to endorse or promote products derived
19  *       from this software without specific prior written permission.
20  *
21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32  */
33 
34 #include "jsonrpc_internal.h"
35 #include "spdk/string.h"
36 
37 struct spdk_jsonrpc_server *
38 spdk_jsonrpc_server_listen(int domain, int protocol,
39 			   struct sockaddr *listen_addr, socklen_t addrlen,
40 			   spdk_jsonrpc_handle_request_fn handle_request)
41 {
42 	struct spdk_jsonrpc_server *server;
43 	int rc, val, flag;
44 
45 	server = calloc(1, sizeof(struct spdk_jsonrpc_server));
46 	if (server == NULL) {
47 		return NULL;
48 	}
49 
50 	server->handle_request = handle_request;
51 
52 	server->sockfd = socket(domain, SOCK_STREAM, protocol);
53 	if (server->sockfd < 0) {
54 		SPDK_ERRLOG("socket() failed\n");
55 		free(server);
56 		return NULL;
57 	}
58 
59 	val = 1;
60 	setsockopt(server->sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
61 	if (protocol == IPPROTO_TCP) {
62 		setsockopt(server->sockfd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val));
63 	}
64 
65 	flag = fcntl(server->sockfd, F_GETFL);
66 	if (fcntl(server->sockfd, F_SETFL, flag | O_NONBLOCK) < 0) {
67 		SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%s)\n",
68 			    server->sockfd, spdk_strerror(errno));
69 		close(server->sockfd);
70 		free(server);
71 		return NULL;
72 	}
73 
74 	rc = bind(server->sockfd, listen_addr, addrlen);
75 	if (rc != 0) {
76 		SPDK_ERRLOG("could not bind JSON-RPC server: %s\n", spdk_strerror(errno));
77 		close(server->sockfd);
78 		free(server);
79 		return NULL;
80 	}
81 
82 	rc = listen(server->sockfd, 512);
83 	if (rc != 0) {
84 		SPDK_ERRLOG("listen() failed, errno = %d\n", errno);
85 		close(server->sockfd);
86 		free(server);
87 		return NULL;
88 	}
89 
90 	return server;
91 }
92 
93 void
94 spdk_jsonrpc_server_shutdown(struct spdk_jsonrpc_server *server)
95 {
96 	int i;
97 
98 	close(server->sockfd);
99 
100 	for (i = 0; i < server->num_conns; i++) {
101 		close(server->conns[i].sockfd);
102 	}
103 
104 	free(server);
105 }
106 
107 static void
108 spdk_jsonrpc_server_conn_close(struct spdk_jsonrpc_server_conn *conn)
109 {
110 	conn->closed = true;
111 
112 	if (conn->sockfd >= 0) {
113 		close(conn->sockfd);
114 		conn->sockfd = -1;
115 	}
116 }
117 
118 static void
119 spdk_jsonrpc_server_conn_remove(struct spdk_jsonrpc_server_conn *conn)
120 {
121 	struct spdk_jsonrpc_server *server = conn->server;
122 	int conn_idx = conn - server->conns;
123 
124 	spdk_jsonrpc_server_conn_close(conn);
125 
126 	pthread_spin_destroy(&conn->queue_lock);
127 	assert(STAILQ_EMPTY(&conn->send_queue));
128 
129 	/* Swap conn with the last entry in conns */
130 	server->conns[conn_idx] = server->conns[server->num_conns - 1];
131 	server->num_conns--;
132 }
133 
134 static int
135 spdk_jsonrpc_server_accept(struct spdk_jsonrpc_server *server)
136 {
137 	struct spdk_jsonrpc_server_conn *conn;
138 	int rc, conn_idx, flag;
139 
140 	rc = accept(server->sockfd, NULL, NULL);
141 	if (rc >= 0) {
142 		assert(server->num_conns < SPDK_JSONRPC_MAX_CONNS);
143 		conn_idx = server->num_conns;
144 		conn = &server->conns[conn_idx];
145 		conn->server = server;
146 		conn->sockfd = rc;
147 		conn->closed = false;
148 		conn->recv_len = 0;
149 		conn->outstanding_requests = 0;
150 		pthread_spin_init(&conn->queue_lock, PTHREAD_PROCESS_PRIVATE);
151 		STAILQ_INIT(&conn->send_queue);
152 		conn->send_request = NULL;
153 
154 		flag = fcntl(conn->sockfd, F_GETFL);
155 		if (fcntl(conn->sockfd, F_SETFL, flag | O_NONBLOCK) < 0) {
156 			SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%s)\n",
157 				    conn->sockfd, spdk_strerror(errno));
158 			close(conn->sockfd);
159 			return -1;
160 		}
161 
162 		server->num_conns++;
163 
164 		return 0;
165 	}
166 
167 	if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
168 		return 0;
169 	}
170 
171 	return -1;
172 }
173 
174 void
175 spdk_jsonrpc_server_handle_request(struct spdk_jsonrpc_request *request,
176 				   const struct spdk_json_val *method, const struct spdk_json_val *params)
177 {
178 	request->conn->server->handle_request(request, method, params);
179 }
180 
181 void
182 spdk_jsonrpc_server_handle_error(struct spdk_jsonrpc_request *request, int error)
183 {
184 	const char *msg;
185 
186 	switch (error) {
187 	case SPDK_JSONRPC_ERROR_PARSE_ERROR:
188 		msg = "Parse error";
189 		break;
190 
191 	case SPDK_JSONRPC_ERROR_INVALID_REQUEST:
192 		msg = "Invalid request";
193 		break;
194 
195 	case SPDK_JSONRPC_ERROR_METHOD_NOT_FOUND:
196 		msg = "Method not found";
197 		break;
198 
199 	case SPDK_JSONRPC_ERROR_INVALID_PARAMS:
200 		msg = "Invalid parameters";
201 		break;
202 
203 	case SPDK_JSONRPC_ERROR_INTERNAL_ERROR:
204 		msg = "Internal error";
205 		break;
206 
207 	default:
208 		msg = "Error";
209 		break;
210 	}
211 
212 	spdk_jsonrpc_send_error_response(request, error, msg);
213 }
214 
215 static int
216 spdk_jsonrpc_server_conn_recv(struct spdk_jsonrpc_server_conn *conn)
217 {
218 	ssize_t rc;
219 	size_t recv_avail = SPDK_JSONRPC_RECV_BUF_SIZE - conn->recv_len;
220 
221 	rc = recv(conn->sockfd, conn->recv_buf + conn->recv_len, recv_avail, 0);
222 	if (rc == -1) {
223 		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
224 			return 0;
225 		}
226 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "recv() failed: %s\n", spdk_strerror(errno));
227 		return -1;
228 	}
229 
230 	if (rc == 0) {
231 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "remote closed connection\n");
232 		return -1;
233 	}
234 
235 	conn->recv_len += rc;
236 
237 	rc = spdk_jsonrpc_parse_request(conn, conn->recv_buf, conn->recv_len);
238 	if (rc < 0) {
239 		SPDK_ERRLOG("jsonrpc parse request failed\n");
240 		return -1;
241 	}
242 
243 	if (rc > 0) {
244 		/*
245 		 * Successfully parsed a request - move any data past the end of the
246 		 * parsed request down to the beginning.
247 		 */
248 		assert((size_t)rc <= conn->recv_len);
249 		memmove(conn->recv_buf, conn->recv_buf + rc, conn->recv_len - rc);
250 		conn->recv_len -= rc;
251 	}
252 
253 	return 0;
254 }
255 
256 void
257 spdk_jsonrpc_server_send_response(struct spdk_jsonrpc_server_conn *conn,
258 				  struct spdk_jsonrpc_request *request)
259 {
260 	/* Queue the response to be sent */
261 	pthread_spin_lock(&conn->queue_lock);
262 	STAILQ_INSERT_TAIL(&conn->send_queue, request, link);
263 	pthread_spin_unlock(&conn->queue_lock);
264 }
265 
266 static struct spdk_jsonrpc_request *
267 spdk_jsonrpc_server_dequeue_request(struct spdk_jsonrpc_server_conn *conn)
268 {
269 	struct spdk_jsonrpc_request *request = NULL;
270 
271 	pthread_spin_lock(&conn->queue_lock);
272 	request = STAILQ_FIRST(&conn->send_queue);
273 	if (request) {
274 		STAILQ_REMOVE_HEAD(&conn->send_queue, link);
275 	}
276 	pthread_spin_unlock(&conn->queue_lock);
277 	return request;
278 }
279 
280 
281 static int
282 spdk_jsonrpc_server_conn_send(struct spdk_jsonrpc_server_conn *conn)
283 {
284 	struct spdk_jsonrpc_request *request;
285 	ssize_t rc;
286 
287 more:
288 	if (conn->outstanding_requests == 0) {
289 		return 0;
290 	}
291 
292 	if (conn->send_request == NULL) {
293 		conn->send_request = spdk_jsonrpc_server_dequeue_request(conn);
294 		if (conn->send_request == NULL) {
295 			return 0;
296 		}
297 	}
298 
299 	request = conn->send_request;
300 	if (request == NULL) {
301 		/* Nothing to send right now */
302 		return 0;
303 	}
304 
305 	rc = send(conn->sockfd, request->send_buf + request->send_offset,
306 		  request->send_len, 0);
307 	if (rc < 0) {
308 		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
309 			return 0;
310 		}
311 
312 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "send() failed: %s\n", spdk_strerror(errno));
313 		return -1;
314 	}
315 
316 	request->send_offset += rc;
317 	request->send_len -= rc;
318 
319 	if (request->send_len == 0) {
320 		/*
321 		 * Full response has been sent.
322 		 * Free it and set send_request to NULL to move on to the next queued response.
323 		 */
324 		conn->send_request = NULL;
325 		spdk_jsonrpc_free_request(request);
326 		goto more;
327 	}
328 
329 	return 0;
330 }
331 
332 int
333 spdk_jsonrpc_server_poll(struct spdk_jsonrpc_server *server)
334 {
335 	int rc, i;
336 	struct spdk_jsonrpc_server_conn *conn;
337 
338 	for (i = 0; i < server->num_conns; i++) {
339 		conn = &server->conns[i];
340 
341 		if (conn->closed) {
342 			struct spdk_jsonrpc_request *request;
343 
344 			/*
345 			 * The client closed the connection, but there may still be requests
346 			 * outstanding; we have no way to cancel outstanding requests, so wait until
347 			 * each outstanding request sends a response (which will be discarded, since
348 			 * the connection is closed).
349 			 */
350 
351 			if (conn->send_request) {
352 				spdk_jsonrpc_free_request(conn->send_request);
353 				conn->send_request = NULL;
354 			}
355 
356 			while ((request = spdk_jsonrpc_server_dequeue_request(conn)) != NULL) {
357 				spdk_jsonrpc_free_request(request);
358 			}
359 
360 			if (conn->outstanding_requests == 0) {
361 				SPDK_DEBUGLOG(SPDK_LOG_RPC, "all outstanding requests completed\n");
362 				spdk_jsonrpc_server_conn_remove(conn);
363 			}
364 		}
365 	}
366 
367 	/* Check listen socket */
368 	if (server->num_conns < SPDK_JSONRPC_MAX_CONNS) {
369 		spdk_jsonrpc_server_accept(server);
370 	}
371 
372 	for (i = 0; i < server->num_conns; i++) {
373 		conn = &server->conns[i];
374 
375 		if (conn->closed) {
376 			continue;
377 		}
378 
379 		rc = spdk_jsonrpc_server_conn_send(conn);
380 		if (rc != 0) {
381 			spdk_jsonrpc_server_conn_close(conn);
382 			continue;
383 		}
384 
385 		rc = spdk_jsonrpc_server_conn_recv(conn);
386 		if (rc != 0) {
387 			spdk_jsonrpc_server_conn_close(conn);
388 			continue;
389 		}
390 	}
391 
392 	return 0;
393 }
394