xref: /spdk/lib/jsonrpc/jsonrpc_server_tcp.c (revision 8a0a98d35e21f282088edf28b9e8da66ec390e3a) 
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, i;
44 
45 	server = calloc(1, sizeof(struct spdk_jsonrpc_server));
46 	if (server == NULL) {
47 		return NULL;
48 	}
49 
50 	TAILQ_INIT(&server->free_conns);
51 	TAILQ_INIT(&server->conns);
52 
53 	for (i = 0; i < SPDK_JSONRPC_MAX_CONNS; i++) {
54 		TAILQ_INSERT_TAIL(&server->free_conns, &server->conns_array[i], link);
55 	}
56 
57 	server->handle_request = handle_request;
58 
59 	server->sockfd = socket(domain, SOCK_STREAM, protocol);
60 	if (server->sockfd < 0) {
61 		SPDK_ERRLOG("socket() failed\n");
62 		free(server);
63 		return NULL;
64 	}
65 
66 	val = 1;
67 	setsockopt(server->sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
68 	if (protocol == IPPROTO_TCP) {
69 		setsockopt(server->sockfd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val));
70 	}
71 
72 	flag = fcntl(server->sockfd, F_GETFL);
73 	if (fcntl(server->sockfd, F_SETFL, flag | O_NONBLOCK) < 0) {
74 		SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%s)\n",
75 			    server->sockfd, spdk_strerror(errno));
76 		close(server->sockfd);
77 		free(server);
78 		return NULL;
79 	}
80 
81 	rc = bind(server->sockfd, listen_addr, addrlen);
82 	if (rc != 0) {
83 		SPDK_ERRLOG("could not bind JSON-RPC server: %s\n", spdk_strerror(errno));
84 		close(server->sockfd);
85 		free(server);
86 		return NULL;
87 	}
88 
89 	rc = listen(server->sockfd, 512);
90 	if (rc != 0) {
91 		SPDK_ERRLOG("listen() failed, errno = %d\n", errno);
92 		close(server->sockfd);
93 		free(server);
94 		return NULL;
95 	}
96 
97 	return server;
98 }
99 
100 void
101 spdk_jsonrpc_server_shutdown(struct spdk_jsonrpc_server *server)
102 {
103 	struct spdk_jsonrpc_server_conn *conn;
104 
105 	close(server->sockfd);
106 
107 	TAILQ_FOREACH(conn, &server->conns, link) {
108 		close(conn->sockfd);
109 	}
110 
111 	free(server);
112 }
113 
114 static void
115 spdk_jsonrpc_server_conn_close(struct spdk_jsonrpc_server_conn *conn)
116 {
117 	conn->closed = true;
118 
119 	if (conn->sockfd >= 0) {
120 		close(conn->sockfd);
121 		conn->sockfd = -1;
122 	}
123 }
124 
125 static void
126 spdk_jsonrpc_server_conn_remove(struct spdk_jsonrpc_server_conn *conn)
127 {
128 	struct spdk_jsonrpc_server *server = conn->server;
129 
130 	spdk_jsonrpc_server_conn_close(conn);
131 
132 	pthread_spin_destroy(&conn->queue_lock);
133 	assert(STAILQ_EMPTY(&conn->send_queue));
134 
135 	TAILQ_REMOVE(&server->conns, conn, link);
136 	TAILQ_INSERT_HEAD(&server->free_conns, conn, link);
137 }
138 
139 static int
140 spdk_jsonrpc_server_accept(struct spdk_jsonrpc_server *server)
141 {
142 	struct spdk_jsonrpc_server_conn *conn;
143 	int rc, flag;
144 
145 	rc = accept(server->sockfd, NULL, NULL);
146 	if (rc >= 0) {
147 		conn = TAILQ_FIRST(&server->free_conns);
148 		assert(conn != NULL);
149 
150 		conn->server = server;
151 		conn->sockfd = rc;
152 		conn->closed = false;
153 		conn->recv_len = 0;
154 		conn->outstanding_requests = 0;
155 		pthread_spin_init(&conn->queue_lock, PTHREAD_PROCESS_PRIVATE);
156 		STAILQ_INIT(&conn->send_queue);
157 		conn->send_request = NULL;
158 
159 		flag = fcntl(conn->sockfd, F_GETFL);
160 		if (fcntl(conn->sockfd, F_SETFL, flag | O_NONBLOCK) < 0) {
161 			SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%s)\n",
162 				    conn->sockfd, spdk_strerror(errno));
163 			close(conn->sockfd);
164 			return -1;
165 		}
166 
167 		TAILQ_REMOVE(&server->free_conns, conn, link);
168 		TAILQ_INSERT_TAIL(&server->conns, conn, link);
169 		return 0;
170 	}
171 
172 	if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
173 		return 0;
174 	}
175 
176 	return -1;
177 }
178 
179 void
180 spdk_jsonrpc_server_handle_request(struct spdk_jsonrpc_request *request,
181 				   const struct spdk_json_val *method, const struct spdk_json_val *params)
182 {
183 	request->conn->server->handle_request(request, method, params);
184 }
185 
186 void
187 spdk_jsonrpc_server_handle_error(struct spdk_jsonrpc_request *request, int error)
188 {
189 	const char *msg;
190 
191 	switch (error) {
192 	case SPDK_JSONRPC_ERROR_PARSE_ERROR:
193 		msg = "Parse error";
194 		break;
195 
196 	case SPDK_JSONRPC_ERROR_INVALID_REQUEST:
197 		msg = "Invalid request";
198 		break;
199 
200 	case SPDK_JSONRPC_ERROR_METHOD_NOT_FOUND:
201 		msg = "Method not found";
202 		break;
203 
204 	case SPDK_JSONRPC_ERROR_INVALID_PARAMS:
205 		msg = "Invalid parameters";
206 		break;
207 
208 	case SPDK_JSONRPC_ERROR_INTERNAL_ERROR:
209 		msg = "Internal error";
210 		break;
211 
212 	default:
213 		msg = "Error";
214 		break;
215 	}
216 
217 	spdk_jsonrpc_send_error_response(request, error, msg);
218 }
219 
220 static int
221 spdk_jsonrpc_server_conn_recv(struct spdk_jsonrpc_server_conn *conn)
222 {
223 	ssize_t rc;
224 	size_t recv_avail = SPDK_JSONRPC_RECV_BUF_SIZE - conn->recv_len;
225 
226 	rc = recv(conn->sockfd, conn->recv_buf + conn->recv_len, recv_avail, 0);
227 	if (rc == -1) {
228 		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
229 			return 0;
230 		}
231 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "recv() failed: %s\n", spdk_strerror(errno));
232 		return -1;
233 	}
234 
235 	if (rc == 0) {
236 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "remote closed connection\n");
237 		return -1;
238 	}
239 
240 	conn->recv_len += rc;
241 
242 	rc = spdk_jsonrpc_parse_request(conn, conn->recv_buf, conn->recv_len);
243 	if (rc < 0) {
244 		SPDK_ERRLOG("jsonrpc parse request failed\n");
245 		return -1;
246 	}
247 
248 	if (rc > 0) {
249 		/*
250 		 * Successfully parsed a request - move any data past the end of the
251 		 * parsed request down to the beginning.
252 		 */
253 		assert((size_t)rc <= conn->recv_len);
254 		memmove(conn->recv_buf, conn->recv_buf + rc, conn->recv_len - rc);
255 		conn->recv_len -= rc;
256 	}
257 
258 	return 0;
259 }
260 
261 void
262 spdk_jsonrpc_server_send_response(struct spdk_jsonrpc_server_conn *conn,
263 				  struct spdk_jsonrpc_request *request)
264 {
265 	/* Queue the response to be sent */
266 	pthread_spin_lock(&conn->queue_lock);
267 	STAILQ_INSERT_TAIL(&conn->send_queue, request, link);
268 	pthread_spin_unlock(&conn->queue_lock);
269 }
270 
271 static struct spdk_jsonrpc_request *
272 spdk_jsonrpc_server_dequeue_request(struct spdk_jsonrpc_server_conn *conn)
273 {
274 	struct spdk_jsonrpc_request *request = NULL;
275 
276 	pthread_spin_lock(&conn->queue_lock);
277 	request = STAILQ_FIRST(&conn->send_queue);
278 	if (request) {
279 		STAILQ_REMOVE_HEAD(&conn->send_queue, link);
280 	}
281 	pthread_spin_unlock(&conn->queue_lock);
282 	return request;
283 }
284 
285 
286 static int
287 spdk_jsonrpc_server_conn_send(struct spdk_jsonrpc_server_conn *conn)
288 {
289 	struct spdk_jsonrpc_request *request;
290 	ssize_t rc;
291 
292 more:
293 	if (conn->outstanding_requests == 0) {
294 		return 0;
295 	}
296 
297 	if (conn->send_request == NULL) {
298 		conn->send_request = spdk_jsonrpc_server_dequeue_request(conn);
299 		if (conn->send_request == NULL) {
300 			return 0;
301 		}
302 	}
303 
304 	request = conn->send_request;
305 	if (request == NULL) {
306 		/* Nothing to send right now */
307 		return 0;
308 	}
309 
310 	rc = send(conn->sockfd, request->send_buf + request->send_offset,
311 		  request->send_len, 0);
312 	if (rc < 0) {
313 		if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
314 			return 0;
315 		}
316 
317 		SPDK_DEBUGLOG(SPDK_LOG_RPC, "send() failed: %s\n", spdk_strerror(errno));
318 		return -1;
319 	}
320 
321 	request->send_offset += rc;
322 	request->send_len -= rc;
323 
324 	if (request->send_len == 0) {
325 		/*
326 		 * Full response has been sent.
327 		 * Free it and set send_request to NULL to move on to the next queued response.
328 		 */
329 		conn->send_request = NULL;
330 		spdk_jsonrpc_free_request(request);
331 		goto more;
332 	}
333 
334 	return 0;
335 }
336 
337 int
338 spdk_jsonrpc_server_poll(struct spdk_jsonrpc_server *server)
339 {
340 	int rc;
341 	struct spdk_jsonrpc_server_conn *conn, *conn_tmp;
342 
343 	TAILQ_FOREACH_SAFE(conn, &server->conns, link, conn_tmp) {
344 		if (conn->closed) {
345 			struct spdk_jsonrpc_request *request;
346 
347 			/*
348 			 * The client closed the connection, but there may still be requests
349 			 * outstanding; we have no way to cancel outstanding requests, so wait until
350 			 * each outstanding request sends a response (which will be discarded, since
351 			 * the connection is closed).
352 			 */
353 
354 			if (conn->send_request) {
355 				spdk_jsonrpc_free_request(conn->send_request);
356 				conn->send_request = NULL;
357 			}
358 
359 			while ((request = spdk_jsonrpc_server_dequeue_request(conn)) != NULL) {
360 				spdk_jsonrpc_free_request(request);
361 			}
362 
363 			if (conn->outstanding_requests == 0) {
364 				SPDK_DEBUGLOG(SPDK_LOG_RPC, "all outstanding requests completed\n");
365 				spdk_jsonrpc_server_conn_remove(conn);
366 			}
367 		}
368 	}
369 
370 	/* Check listen socket */
371 	if (!TAILQ_EMPTY(&server->free_conns)) {
372 		spdk_jsonrpc_server_accept(server);
373 	}
374 
375 	TAILQ_FOREACH(conn, &server->conns, link) {
376 		if (conn->closed) {
377 			continue;
378 		}
379 
380 		rc = spdk_jsonrpc_server_conn_send(conn);
381 		if (rc != 0) {
382 			spdk_jsonrpc_server_conn_close(conn);
383 			continue;
384 		}
385 
386 		rc = spdk_jsonrpc_server_conn_recv(conn);
387 		if (rc != 0) {
388 			spdk_jsonrpc_server_conn_close(conn);
389 			continue;
390 		}
391 	}
392 
393 	return 0;
394 }
395