xref: /spdk/module/sock/posix/posix.c (revision 57fd99b91e71a4baa5543e19ff83958dc99d4dac)
1 /*   SPDX-License-Identifier: BSD-3-Clause
2  *   Copyright (C) 2018 Intel Corporation. All rights reserved.
3  *   Copyright (c) 2020, 2021 Mellanox Technologies LTD. All rights reserved.
4  *   Copyright (c) 2021 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
5  */
6 
7 #include "spdk/stdinc.h"
8 
9 #if defined(__FreeBSD__)
10 #include <sys/event.h>
11 #define SPDK_KEVENT
12 #else
13 #include <sys/epoll.h>
14 #define SPDK_EPOLL
15 #endif
16 
17 #if defined(__linux__)
18 #include <linux/errqueue.h>
19 #endif
20 
21 #include "spdk/env.h"
22 #include "spdk/log.h"
23 #include "spdk/pipe.h"
24 #include "spdk/sock.h"
25 #include "spdk/util.h"
26 #include "spdk/string.h"
27 #include "spdk/net.h"
28 #include "spdk/file.h"
29 #include "spdk_internal/sock.h"
30 #include "spdk/net.h"
31 
32 #include "openssl/crypto.h"
33 #include "openssl/err.h"
34 #include "openssl/ssl.h"
35 
36 #define MAX_TMPBUF 1024
37 #define PORTNUMLEN 32
38 
39 #if defined(SO_ZEROCOPY) && defined(MSG_ZEROCOPY)
40 #define SPDK_ZEROCOPY
41 #endif
42 
43 struct spdk_posix_sock {
44 	struct spdk_sock	base;
45 	int			fd;
46 
47 	uint32_t		sendmsg_idx;
48 
49 	struct spdk_pipe	*recv_pipe;
50 	int			recv_buf_sz;
51 	bool			pipe_has_data;
52 	bool			socket_has_data;
53 	bool			zcopy;
54 
55 	int			placement_id;
56 
57 	SSL_CTX			*ctx;
58 	SSL			*ssl;
59 
60 	TAILQ_ENTRY(spdk_posix_sock)	link;
61 
62 	char			interface_name[IFNAMSIZ];
63 };
64 
65 TAILQ_HEAD(spdk_has_data_list, spdk_posix_sock);
66 
67 struct spdk_posix_sock_group_impl {
68 	struct spdk_sock_group_impl	base;
69 	int				fd;
70 	struct spdk_interrupt		*intr;
71 	struct spdk_has_data_list	socks_with_data;
72 	int				placement_id;
73 	struct spdk_pipe_group		*pipe_group;
74 };
75 
76 static struct spdk_sock_impl_opts g_posix_impl_opts = {
77 	.recv_buf_size = DEFAULT_SO_RCVBUF_SIZE,
78 	.send_buf_size = DEFAULT_SO_SNDBUF_SIZE,
79 	.enable_recv_pipe = true,
80 	.enable_quickack = false,
81 	.enable_placement_id = PLACEMENT_NONE,
82 	.enable_zerocopy_send_server = true,
83 	.enable_zerocopy_send_client = false,
84 	.zerocopy_threshold = 0,
85 	.tls_version = 0,
86 	.enable_ktls = false,
87 	.psk_key = NULL,
88 	.psk_key_size = 0,
89 	.psk_identity = NULL,
90 	.get_key = NULL,
91 	.get_key_ctx = NULL,
92 	.tls_cipher_suites = NULL
93 };
94 
95 static struct spdk_sock_impl_opts g_ssl_impl_opts = {
96 	.recv_buf_size = MIN_SO_RCVBUF_SIZE,
97 	.send_buf_size = MIN_SO_SNDBUF_SIZE,
98 	.enable_recv_pipe = true,
99 	.enable_quickack = false,
100 	.enable_placement_id = PLACEMENT_NONE,
101 	.enable_zerocopy_send_server = true,
102 	.enable_zerocopy_send_client = false,
103 	.zerocopy_threshold = 0,
104 	.tls_version = 0,
105 	.enable_ktls = false,
106 	.psk_key = NULL,
107 	.psk_identity = NULL
108 };
109 
110 static struct spdk_sock_map g_map = {
111 	.entries = STAILQ_HEAD_INITIALIZER(g_map.entries),
112 	.mtx = PTHREAD_MUTEX_INITIALIZER
113 };
114 
115 __attribute((destructor)) static void
116 posix_sock_map_cleanup(void)
117 {
118 	spdk_sock_map_cleanup(&g_map);
119 }
120 
121 #define __posix_sock(sock) (struct spdk_posix_sock *)sock
122 #define __posix_group_impl(group) (struct spdk_posix_sock_group_impl *)group
123 
124 static void
125 posix_sock_copy_impl_opts(struct spdk_sock_impl_opts *dest, const struct spdk_sock_impl_opts *src,
126 			  size_t len)
127 {
128 #define FIELD_OK(field) \
129 	offsetof(struct spdk_sock_impl_opts, field) + sizeof(src->field) <= len
130 
131 #define SET_FIELD(field) \
132 	if (FIELD_OK(field)) { \
133 		dest->field = src->field; \
134 	}
135 
136 	SET_FIELD(recv_buf_size);
137 	SET_FIELD(send_buf_size);
138 	SET_FIELD(enable_recv_pipe);
139 	SET_FIELD(enable_zerocopy_send);
140 	SET_FIELD(enable_quickack);
141 	SET_FIELD(enable_placement_id);
142 	SET_FIELD(enable_zerocopy_send_server);
143 	SET_FIELD(enable_zerocopy_send_client);
144 	SET_FIELD(zerocopy_threshold);
145 	SET_FIELD(tls_version);
146 	SET_FIELD(enable_ktls);
147 	SET_FIELD(psk_key);
148 	SET_FIELD(psk_key_size);
149 	SET_FIELD(psk_identity);
150 	SET_FIELD(get_key);
151 	SET_FIELD(get_key_ctx);
152 	SET_FIELD(tls_cipher_suites);
153 
154 #undef SET_FIELD
155 #undef FIELD_OK
156 }
157 
158 static int
159 _sock_impl_get_opts(struct spdk_sock_impl_opts *opts, struct spdk_sock_impl_opts *impl_opts,
160 		    size_t *len)
161 {
162 	if (!opts || !len) {
163 		errno = EINVAL;
164 		return -1;
165 	}
166 
167 	assert(sizeof(*opts) >= *len);
168 	memset(opts, 0, *len);
169 
170 	posix_sock_copy_impl_opts(opts, impl_opts, *len);
171 	*len = spdk_min(*len, sizeof(*impl_opts));
172 
173 	return 0;
174 }
175 
176 static int
177 posix_sock_impl_get_opts(struct spdk_sock_impl_opts *opts, size_t *len)
178 {
179 	return _sock_impl_get_opts(opts, &g_posix_impl_opts, len);
180 }
181 
182 static int
183 ssl_sock_impl_get_opts(struct spdk_sock_impl_opts *opts, size_t *len)
184 {
185 	return _sock_impl_get_opts(opts, &g_ssl_impl_opts, len);
186 }
187 
188 static int
189 _sock_impl_set_opts(const struct spdk_sock_impl_opts *opts, struct spdk_sock_impl_opts *impl_opts,
190 		    size_t len)
191 {
192 	if (!opts) {
193 		errno = EINVAL;
194 		return -1;
195 	}
196 
197 	assert(sizeof(*opts) >= len);
198 	posix_sock_copy_impl_opts(impl_opts, opts, len);
199 
200 	return 0;
201 }
202 
203 static int
204 posix_sock_impl_set_opts(const struct spdk_sock_impl_opts *opts, size_t len)
205 {
206 	return _sock_impl_set_opts(opts, &g_posix_impl_opts, len);
207 }
208 
209 static int
210 ssl_sock_impl_set_opts(const struct spdk_sock_impl_opts *opts, size_t len)
211 {
212 	return _sock_impl_set_opts(opts, &g_ssl_impl_opts, len);
213 }
214 
215 static void
216 _opts_get_impl_opts(const struct spdk_sock_opts *opts, struct spdk_sock_impl_opts *dest,
217 		    const struct spdk_sock_impl_opts *default_impl)
218 {
219 	/* Copy the default impl_opts first to cover cases when user's impl_opts is smaller */
220 	memcpy(dest, default_impl, sizeof(*dest));
221 
222 	if (opts->impl_opts != NULL) {
223 		assert(sizeof(*dest) >= opts->impl_opts_size);
224 		posix_sock_copy_impl_opts(dest, opts->impl_opts, opts->impl_opts_size);
225 	}
226 }
227 
228 static int
229 posix_sock_getaddr(struct spdk_sock *_sock, char *saddr, int slen, uint16_t *sport,
230 		   char *caddr, int clen, uint16_t *cport)
231 {
232 	struct spdk_posix_sock *sock = __posix_sock(_sock);
233 
234 	assert(sock != NULL);
235 	return spdk_net_getaddr(sock->fd, saddr, slen, sport, caddr, clen, cport);
236 }
237 
238 static const char *
239 posix_sock_get_interface_name(struct spdk_sock *_sock)
240 {
241 	struct spdk_posix_sock *sock = __posix_sock(_sock);
242 	char saddr[64];
243 	int rc;
244 
245 	rc = spdk_net_getaddr(sock->fd, saddr, sizeof(saddr), NULL, NULL, 0, NULL);
246 	if (rc != 0) {
247 		return NULL;
248 	}
249 
250 	rc = spdk_net_get_interface_name(saddr, sock->interface_name,
251 					 sizeof(sock->interface_name));
252 	if (rc != 0) {
253 		return NULL;
254 	}
255 
256 	return sock->interface_name;
257 }
258 
259 static uint32_t
260 posix_sock_get_numa_socket_id(struct spdk_sock *sock)
261 {
262 	const char *interface_name;
263 	uint32_t numa_socket_id;
264 	int rc;
265 
266 	interface_name = posix_sock_get_interface_name(sock);
267 	if (interface_name == NULL) {
268 		return SPDK_ENV_SOCKET_ID_ANY;
269 	}
270 
271 	rc = spdk_read_sysfs_attribute_uint32(&numa_socket_id,
272 					      "/sys/class/net/%s/device/numa_node", interface_name);
273 	if (rc == 0) {
274 		return numa_socket_id;
275 	} else {
276 		return SPDK_ENV_SOCKET_ID_ANY;
277 	}
278 }
279 
280 enum posix_sock_create_type {
281 	SPDK_SOCK_CREATE_LISTEN,
282 	SPDK_SOCK_CREATE_CONNECT,
283 };
284 
285 static int
286 posix_sock_alloc_pipe(struct spdk_posix_sock *sock, int sz)
287 {
288 	uint8_t *new_buf, *old_buf;
289 	struct spdk_pipe *new_pipe;
290 	struct iovec siov[2];
291 	struct iovec diov[2];
292 	int sbytes;
293 	ssize_t bytes;
294 	int rc;
295 
296 	if (sock->recv_buf_sz == sz) {
297 		return 0;
298 	}
299 
300 	/* If the new size is 0, just free the pipe */
301 	if (sz == 0) {
302 		old_buf = spdk_pipe_destroy(sock->recv_pipe);
303 		free(old_buf);
304 		sock->recv_pipe = NULL;
305 		return 0;
306 	} else if (sz < MIN_SOCK_PIPE_SIZE) {
307 		SPDK_ERRLOG("The size of the pipe must be larger than %d\n", MIN_SOCK_PIPE_SIZE);
308 		return -1;
309 	}
310 
311 	/* Round up to next 64 byte multiple */
312 	rc = posix_memalign((void **)&new_buf, 64, sz);
313 	if (rc != 0) {
314 		SPDK_ERRLOG("socket recv buf allocation failed\n");
315 		return -ENOMEM;
316 	}
317 	memset(new_buf, 0, sz);
318 
319 	new_pipe = spdk_pipe_create(new_buf, sz);
320 	if (new_pipe == NULL) {
321 		SPDK_ERRLOG("socket pipe allocation failed\n");
322 		free(new_buf);
323 		return -ENOMEM;
324 	}
325 
326 	if (sock->recv_pipe != NULL) {
327 		/* Pull all of the data out of the old pipe */
328 		sbytes = spdk_pipe_reader_get_buffer(sock->recv_pipe, sock->recv_buf_sz, siov);
329 		if (sbytes > sz) {
330 			/* Too much data to fit into the new pipe size */
331 			old_buf = spdk_pipe_destroy(new_pipe);
332 			free(old_buf);
333 			return -EINVAL;
334 		}
335 
336 		sbytes = spdk_pipe_writer_get_buffer(new_pipe, sz, diov);
337 		assert(sbytes == sz);
338 
339 		bytes = spdk_iovcpy(siov, 2, diov, 2);
340 		spdk_pipe_writer_advance(new_pipe, bytes);
341 
342 		old_buf = spdk_pipe_destroy(sock->recv_pipe);
343 		free(old_buf);
344 	}
345 
346 	sock->recv_buf_sz = sz;
347 	sock->recv_pipe = new_pipe;
348 
349 	if (sock->base.group_impl) {
350 		struct spdk_posix_sock_group_impl *group;
351 
352 		group = __posix_group_impl(sock->base.group_impl);
353 		spdk_pipe_group_add(group->pipe_group, sock->recv_pipe);
354 	}
355 
356 	return 0;
357 }
358 
359 static int
360 posix_sock_set_recvbuf(struct spdk_sock *_sock, int sz)
361 {
362 	struct spdk_posix_sock *sock = __posix_sock(_sock);
363 	int min_size;
364 	int rc;
365 
366 	assert(sock != NULL);
367 
368 	if (_sock->impl_opts.enable_recv_pipe) {
369 		rc = posix_sock_alloc_pipe(sock, sz);
370 		if (rc) {
371 			return rc;
372 		}
373 	}
374 
375 	/* Set kernel buffer size to be at least MIN_SO_RCVBUF_SIZE and
376 	 * _sock->impl_opts.recv_buf_size. */
377 	min_size = spdk_max(MIN_SO_RCVBUF_SIZE, _sock->impl_opts.recv_buf_size);
378 
379 	if (sz < min_size) {
380 		sz = min_size;
381 	}
382 
383 	rc = setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &sz, sizeof(sz));
384 	if (rc < 0) {
385 		return rc;
386 	}
387 
388 	_sock->impl_opts.recv_buf_size = sz;
389 
390 	return 0;
391 }
392 
393 static int
394 posix_sock_set_sendbuf(struct spdk_sock *_sock, int sz)
395 {
396 	struct spdk_posix_sock *sock = __posix_sock(_sock);
397 	int min_size;
398 	int rc;
399 
400 	assert(sock != NULL);
401 
402 	/* Set kernel buffer size to be at least MIN_SO_SNDBUF_SIZE and
403 	 * _sock->impl_opts.send_buf_size. */
404 	min_size = spdk_max(MIN_SO_SNDBUF_SIZE, _sock->impl_opts.send_buf_size);
405 
406 	if (sz < min_size) {
407 		sz = min_size;
408 	}
409 
410 	rc = setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz));
411 	if (rc < 0) {
412 		return rc;
413 	}
414 
415 	_sock->impl_opts.send_buf_size = sz;
416 
417 	return 0;
418 }
419 
420 static void
421 posix_sock_init(struct spdk_posix_sock *sock, bool enable_zero_copy)
422 {
423 #if defined(SPDK_ZEROCOPY) || defined(__linux__)
424 	int flag;
425 	int rc;
426 #endif
427 
428 #if defined(SPDK_ZEROCOPY)
429 	flag = 1;
430 
431 	if (enable_zero_copy) {
432 		/* Try to turn on zero copy sends */
433 		rc = setsockopt(sock->fd, SOL_SOCKET, SO_ZEROCOPY, &flag, sizeof(flag));
434 		if (rc == 0) {
435 			sock->zcopy = true;
436 		}
437 	}
438 #endif
439 
440 #if defined(__linux__)
441 	flag = 1;
442 
443 	if (sock->base.impl_opts.enable_quickack) {
444 		rc = setsockopt(sock->fd, IPPROTO_TCP, TCP_QUICKACK, &flag, sizeof(flag));
445 		if (rc != 0) {
446 			SPDK_ERRLOG("quickack was failed to set\n");
447 		}
448 	}
449 
450 	spdk_sock_get_placement_id(sock->fd, sock->base.impl_opts.enable_placement_id,
451 				   &sock->placement_id);
452 
453 	if (sock->base.impl_opts.enable_placement_id == PLACEMENT_MARK) {
454 		/* Save placement_id */
455 		spdk_sock_map_insert(&g_map, sock->placement_id, NULL);
456 	}
457 #endif
458 }
459 
460 static struct spdk_posix_sock *
461 posix_sock_alloc(int fd, struct spdk_sock_impl_opts *impl_opts, bool enable_zero_copy)
462 {
463 	struct spdk_posix_sock *sock;
464 
465 	sock = calloc(1, sizeof(*sock));
466 	if (sock == NULL) {
467 		SPDK_ERRLOG("sock allocation failed\n");
468 		return NULL;
469 	}
470 
471 	sock->fd = fd;
472 	memcpy(&sock->base.impl_opts, impl_opts, sizeof(*impl_opts));
473 	posix_sock_init(sock, enable_zero_copy);
474 
475 	return sock;
476 }
477 
478 static int
479 posix_fd_create(struct addrinfo *res, struct spdk_sock_opts *opts,
480 		struct spdk_sock_impl_opts *impl_opts)
481 {
482 	int fd;
483 	int val = 1;
484 	int rc, sz;
485 #if defined(__linux__)
486 	int to;
487 #endif
488 
489 	fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
490 	if (fd < 0) {
491 		/* error */
492 		return -1;
493 	}
494 
495 	sz = impl_opts->recv_buf_size;
496 	rc = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &sz, sizeof(sz));
497 	if (rc) {
498 		/* Not fatal */
499 	}
500 
501 	sz = impl_opts->send_buf_size;
502 	rc = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz));
503 	if (rc) {
504 		/* Not fatal */
505 	}
506 
507 	rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof val);
508 	if (rc != 0) {
509 		close(fd);
510 		/* error */
511 		return -1;
512 	}
513 	rc = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof val);
514 	if (rc != 0) {
515 		close(fd);
516 		/* error */
517 		return -1;
518 	}
519 
520 #if defined(SO_PRIORITY)
521 	if (opts->priority) {
522 		rc = setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &opts->priority, sizeof val);
523 		if (rc != 0) {
524 			close(fd);
525 			/* error */
526 			return -1;
527 		}
528 	}
529 #endif
530 
531 	if (res->ai_family == AF_INET6) {
532 		rc = setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &val, sizeof val);
533 		if (rc != 0) {
534 			close(fd);
535 			/* error */
536 			return -1;
537 		}
538 	}
539 
540 	if (opts->ack_timeout) {
541 #if defined(__linux__)
542 		to = opts->ack_timeout;
543 		rc = setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &to, sizeof(to));
544 		if (rc != 0) {
545 			close(fd);
546 			/* error */
547 			return -1;
548 		}
549 #else
550 		SPDK_WARNLOG("TCP_USER_TIMEOUT is not supported.\n");
551 #endif
552 	}
553 
554 	return fd;
555 }
556 
557 static int
558 posix_sock_psk_find_session_server_cb(SSL *ssl, const unsigned char *identity,
559 				      size_t identity_len, SSL_SESSION **sess)
560 {
561 	struct spdk_sock_impl_opts *impl_opts = SSL_get_app_data(ssl);
562 	uint8_t key[SSL_MAX_MASTER_KEY_LENGTH] = {};
563 	int keylen;
564 	int rc, i;
565 	STACK_OF(SSL_CIPHER) *ciphers;
566 	const SSL_CIPHER *cipher;
567 	const char *cipher_name;
568 	const char *user_cipher = NULL;
569 	bool found = false;
570 
571 	if (impl_opts->get_key) {
572 		rc = impl_opts->get_key(key, sizeof(key), &user_cipher, identity, impl_opts->get_key_ctx);
573 		if (rc < 0) {
574 			SPDK_ERRLOG("Unable to find PSK for identity: %s\n", identity);
575 			return 0;
576 		}
577 		keylen = rc;
578 	} else {
579 		if (impl_opts->psk_key == NULL) {
580 			SPDK_ERRLOG("PSK is not set\n");
581 			return 0;
582 		}
583 
584 		SPDK_DEBUGLOG(sock_posix, "Length of Client's PSK ID %lu\n", strlen(impl_opts->psk_identity));
585 		if (strcmp(impl_opts->psk_identity, identity) != 0) {
586 			SPDK_ERRLOG("Unknown Client's PSK ID\n");
587 			return 0;
588 		}
589 		keylen = impl_opts->psk_key_size;
590 
591 		memcpy(key, impl_opts->psk_key, keylen);
592 		user_cipher = impl_opts->tls_cipher_suites;
593 	}
594 
595 	if (user_cipher == NULL) {
596 		SPDK_ERRLOG("Cipher suite not set\n");
597 		return 0;
598 	}
599 
600 	*sess = SSL_SESSION_new();
601 	if (*sess == NULL) {
602 		SPDK_ERRLOG("Unable to allocate new SSL session\n");
603 		return 0;
604 	}
605 
606 	ciphers = SSL_get_ciphers(ssl);
607 	for (i = 0; i < sk_SSL_CIPHER_num(ciphers); i++) {
608 		cipher = sk_SSL_CIPHER_value(ciphers, i);
609 		cipher_name = SSL_CIPHER_get_name(cipher);
610 
611 		if (strcmp(user_cipher, cipher_name) == 0) {
612 			rc = SSL_SESSION_set_cipher(*sess, cipher);
613 			if (rc != 1) {
614 				SPDK_ERRLOG("Unable to set cipher: %s\n", cipher_name);
615 				goto err;
616 			}
617 			found = true;
618 			break;
619 		}
620 	}
621 	if (found == false) {
622 		SPDK_ERRLOG("No suitable cipher found\n");
623 		goto err;
624 	}
625 
626 	SPDK_DEBUGLOG(sock_posix, "Cipher selected: %s\n", cipher_name);
627 
628 	rc = SSL_SESSION_set_protocol_version(*sess, TLS1_3_VERSION);
629 	if (rc != 1) {
630 		SPDK_ERRLOG("Unable to set TLS version: %d\n", TLS1_3_VERSION);
631 		goto err;
632 	}
633 
634 	rc = SSL_SESSION_set1_master_key(*sess, key, keylen);
635 	if (rc != 1) {
636 		SPDK_ERRLOG("Unable to set PSK for session\n");
637 		goto err;
638 	}
639 
640 	return 1;
641 
642 err:
643 	SSL_SESSION_free(*sess);
644 	*sess = NULL;
645 	return 0;
646 }
647 
648 static int
649 posix_sock_psk_use_session_client_cb(SSL *ssl, const EVP_MD *md, const unsigned char **identity,
650 				     size_t *identity_len, SSL_SESSION **sess)
651 {
652 	struct spdk_sock_impl_opts *impl_opts = SSL_get_app_data(ssl);
653 	int rc, i;
654 	STACK_OF(SSL_CIPHER) *ciphers;
655 	const SSL_CIPHER *cipher;
656 	const char *cipher_name;
657 	long keylen;
658 	bool found = false;
659 
660 	if (impl_opts->psk_key == NULL) {
661 		SPDK_ERRLOG("PSK is not set\n");
662 		return 0;
663 	}
664 	if (impl_opts->psk_key_size > SSL_MAX_MASTER_KEY_LENGTH) {
665 		SPDK_ERRLOG("PSK too long\n");
666 		return 0;
667 	}
668 	keylen = impl_opts->psk_key_size;
669 
670 	if (impl_opts->tls_cipher_suites == NULL) {
671 		SPDK_ERRLOG("Cipher suite not set\n");
672 		return 0;
673 	}
674 	*sess = SSL_SESSION_new();
675 	if (*sess == NULL) {
676 		SPDK_ERRLOG("Unable to allocate new SSL session\n");
677 		return 0;
678 	}
679 
680 	ciphers = SSL_get_ciphers(ssl);
681 	for (i = 0; i < sk_SSL_CIPHER_num(ciphers); i++) {
682 		cipher = sk_SSL_CIPHER_value(ciphers, i);
683 		cipher_name = SSL_CIPHER_get_name(cipher);
684 
685 		if (strcmp(impl_opts->tls_cipher_suites, cipher_name) == 0) {
686 			rc = SSL_SESSION_set_cipher(*sess, cipher);
687 			if (rc != 1) {
688 				SPDK_ERRLOG("Unable to set cipher: %s\n", cipher_name);
689 				goto err;
690 			}
691 			found = true;
692 			break;
693 		}
694 	}
695 	if (found == false) {
696 		SPDK_ERRLOG("No suitable cipher found\n");
697 		goto err;
698 	}
699 
700 	SPDK_DEBUGLOG(sock_posix, "Cipher selected: %s\n", cipher_name);
701 
702 	rc = SSL_SESSION_set_protocol_version(*sess, TLS1_3_VERSION);
703 	if (rc != 1) {
704 		SPDK_ERRLOG("Unable to set TLS version: %d\n", TLS1_3_VERSION);
705 		goto err;
706 	}
707 
708 	rc = SSL_SESSION_set1_master_key(*sess, impl_opts->psk_key, keylen);
709 	if (rc != 1) {
710 		SPDK_ERRLOG("Unable to set PSK for session\n");
711 		goto err;
712 	}
713 
714 	*identity_len = strlen(impl_opts->psk_identity);
715 	*identity = impl_opts->psk_identity;
716 
717 	return 1;
718 
719 err:
720 	SSL_SESSION_free(*sess);
721 	*sess = NULL;
722 	return 0;
723 }
724 
725 static SSL_CTX *
726 posix_sock_create_ssl_context(const SSL_METHOD *method, struct spdk_sock_opts *opts,
727 			      struct spdk_sock_impl_opts *impl_opts)
728 {
729 	SSL_CTX *ctx;
730 	int tls_version = 0;
731 	bool ktls_enabled = false;
732 #ifdef SSL_OP_ENABLE_KTLS
733 	long options;
734 #endif
735 
736 	SSL_library_init();
737 	OpenSSL_add_all_algorithms();
738 	SSL_load_error_strings();
739 	/* Produce a SSL CTX in SSL V2 and V3 standards compliant way */
740 	ctx = SSL_CTX_new(method);
741 	if (!ctx) {
742 		SPDK_ERRLOG("SSL_CTX_new() failed, msg = %s\n", ERR_error_string(ERR_peek_last_error(), NULL));
743 		return NULL;
744 	}
745 	SPDK_DEBUGLOG(sock_posix, "SSL context created\n");
746 
747 	switch (impl_opts->tls_version) {
748 	case 0:
749 		/* auto-negotiation */
750 		break;
751 	case SPDK_TLS_VERSION_1_3:
752 		tls_version = TLS1_3_VERSION;
753 		break;
754 	default:
755 		SPDK_ERRLOG("Incorrect TLS version provided: %d\n", impl_opts->tls_version);
756 		goto err;
757 	}
758 
759 	if (tls_version) {
760 		SPDK_DEBUGLOG(sock_posix, "Hardening TLS version to '%d'='0x%X'\n", impl_opts->tls_version,
761 			      tls_version);
762 		if (!SSL_CTX_set_min_proto_version(ctx, tls_version)) {
763 			SPDK_ERRLOG("Unable to set Min TLS version to '%d'='0x%X\n", impl_opts->tls_version, tls_version);
764 			goto err;
765 		}
766 		if (!SSL_CTX_set_max_proto_version(ctx, tls_version)) {
767 			SPDK_ERRLOG("Unable to set Max TLS version to '%d'='0x%X\n", impl_opts->tls_version, tls_version);
768 			goto err;
769 		}
770 	}
771 	if (impl_opts->enable_ktls) {
772 		SPDK_DEBUGLOG(sock_posix, "Enabling kTLS offload\n");
773 #ifdef SSL_OP_ENABLE_KTLS
774 		options = SSL_CTX_set_options(ctx, SSL_OP_ENABLE_KTLS);
775 		ktls_enabled = options & SSL_OP_ENABLE_KTLS;
776 #else
777 		ktls_enabled = false;
778 #endif
779 		if (!ktls_enabled) {
780 			SPDK_ERRLOG("Unable to set kTLS offload via SSL_CTX_set_options(). Configure openssl with 'enable-ktls'\n");
781 			goto err;
782 		}
783 	}
784 
785 	/* SSL_CTX_set_ciphersuites() return 1 if the requested
786 	 * cipher suite list was configured, and 0 otherwise. */
787 	if (impl_opts->tls_cipher_suites != NULL &&
788 	    SSL_CTX_set_ciphersuites(ctx, impl_opts->tls_cipher_suites) != 1) {
789 		SPDK_ERRLOG("Unable to set TLS cipher suites for SSL'\n");
790 		goto err;
791 	}
792 
793 	return ctx;
794 
795 err:
796 	SSL_CTX_free(ctx);
797 	return NULL;
798 }
799 
800 static SSL *
801 ssl_sock_setup_connect(SSL_CTX *ctx, int fd)
802 {
803 	SSL *ssl;
804 
805 	ssl = SSL_new(ctx);
806 	if (!ssl) {
807 		SPDK_ERRLOG("SSL_new() failed, msg = %s\n", ERR_error_string(ERR_peek_last_error(), NULL));
808 		return NULL;
809 	}
810 	SSL_set_fd(ssl, fd);
811 	SSL_set_connect_state(ssl);
812 	SSL_set_psk_use_session_callback(ssl, posix_sock_psk_use_session_client_cb);
813 	SPDK_DEBUGLOG(sock_posix, "SSL object creation finished: %p\n", ssl);
814 	SPDK_DEBUGLOG(sock_posix, "%s = SSL_state_string_long(%p)\n", SSL_state_string_long(ssl), ssl);
815 	SPDK_DEBUGLOG(sock_posix, "%s = SSL_state_string_long(%p)\n", SSL_state_string_long(ssl), ssl);
816 	SPDK_DEBUGLOG(sock_posix, "Negotiated Cipher suite:%s\n",
817 		      SSL_CIPHER_get_name(SSL_get_current_cipher(ssl)));
818 	return ssl;
819 }
820 
821 static SSL *
822 ssl_sock_setup_accept(SSL_CTX *ctx, int fd)
823 {
824 	SSL *ssl;
825 
826 	ssl = SSL_new(ctx);
827 	if (!ssl) {
828 		SPDK_ERRLOG("SSL_new() failed, msg = %s\n", ERR_error_string(ERR_peek_last_error(), NULL));
829 		return NULL;
830 	}
831 	SSL_set_fd(ssl, fd);
832 	SSL_set_accept_state(ssl);
833 	SSL_set_psk_find_session_callback(ssl, posix_sock_psk_find_session_server_cb);
834 	SPDK_DEBUGLOG(sock_posix, "SSL object creation finished: %p\n", ssl);
835 	SPDK_DEBUGLOG(sock_posix, "%s = SSL_state_string_long(%p)\n", SSL_state_string_long(ssl), ssl);
836 	SPDK_DEBUGLOG(sock_posix, "%s = SSL_state_string_long(%p)\n", SSL_state_string_long(ssl), ssl);
837 	SPDK_DEBUGLOG(sock_posix, "Negotiated Cipher suite:%s\n",
838 		      SSL_CIPHER_get_name(SSL_get_current_cipher(ssl)));
839 	return ssl;
840 }
841 
842 static ssize_t
843 SSL_readv(SSL *ssl, const struct iovec *iov, int iovcnt)
844 {
845 	int i, rc = 0;
846 	ssize_t total = 0;
847 
848 	for (i = 0; i < iovcnt; i++) {
849 		rc = SSL_read(ssl, iov[i].iov_base, iov[i].iov_len);
850 
851 		if (rc > 0) {
852 			total += rc;
853 		}
854 		if (rc != (int)iov[i].iov_len) {
855 			break;
856 		}
857 	}
858 	if (total > 0) {
859 		errno = 0;
860 		return total;
861 	}
862 	switch (SSL_get_error(ssl, rc)) {
863 	case SSL_ERROR_ZERO_RETURN:
864 		errno = ENOTCONN;
865 		return 0;
866 	case SSL_ERROR_WANT_READ:
867 	case SSL_ERROR_WANT_WRITE:
868 	case SSL_ERROR_WANT_CONNECT:
869 	case SSL_ERROR_WANT_ACCEPT:
870 	case SSL_ERROR_WANT_X509_LOOKUP:
871 	case SSL_ERROR_WANT_ASYNC:
872 	case SSL_ERROR_WANT_ASYNC_JOB:
873 	case SSL_ERROR_WANT_CLIENT_HELLO_CB:
874 		errno = EAGAIN;
875 		return -1;
876 	case SSL_ERROR_SYSCALL:
877 	case SSL_ERROR_SSL:
878 		errno = ENOTCONN;
879 		return -1;
880 	default:
881 		errno = ENOTCONN;
882 		return -1;
883 	}
884 }
885 
886 static ssize_t
887 SSL_writev(SSL *ssl, struct iovec *iov, int iovcnt)
888 {
889 	int i, rc = 0;
890 	ssize_t total = 0;
891 
892 	for (i = 0; i < iovcnt; i++) {
893 		rc = SSL_write(ssl, iov[i].iov_base, iov[i].iov_len);
894 
895 		if (rc > 0) {
896 			total += rc;
897 		}
898 		if (rc != (int)iov[i].iov_len) {
899 			break;
900 		}
901 	}
902 	if (total > 0) {
903 		errno = 0;
904 		return total;
905 	}
906 	switch (SSL_get_error(ssl, rc)) {
907 	case SSL_ERROR_ZERO_RETURN:
908 		errno = ENOTCONN;
909 		return 0;
910 	case SSL_ERROR_WANT_READ:
911 	case SSL_ERROR_WANT_WRITE:
912 	case SSL_ERROR_WANT_CONNECT:
913 	case SSL_ERROR_WANT_ACCEPT:
914 	case SSL_ERROR_WANT_X509_LOOKUP:
915 	case SSL_ERROR_WANT_ASYNC:
916 	case SSL_ERROR_WANT_ASYNC_JOB:
917 	case SSL_ERROR_WANT_CLIENT_HELLO_CB:
918 		errno = EAGAIN;
919 		return -1;
920 	case SSL_ERROR_SYSCALL:
921 	case SSL_ERROR_SSL:
922 		errno = ENOTCONN;
923 		return -1;
924 	default:
925 		errno = ENOTCONN;
926 		return -1;
927 	}
928 }
929 
930 static struct spdk_sock *
931 posix_sock_create(const char *ip, int port,
932 		  enum posix_sock_create_type type,
933 		  struct spdk_sock_opts *opts,
934 		  bool enable_ssl)
935 {
936 	struct spdk_posix_sock *sock;
937 	struct spdk_sock_impl_opts impl_opts;
938 	char buf[MAX_TMPBUF];
939 	char portnum[PORTNUMLEN];
940 	char *p;
941 	const char *src_addr;
942 	uint16_t src_port;
943 	struct addrinfo hints, *res, *res0, *src_ai;
944 	int fd, flag;
945 	int rc;
946 	bool enable_zcopy_user_opts = true;
947 	bool enable_zcopy_impl_opts = true;
948 	SSL_CTX *ctx = 0;
949 	SSL *ssl = 0;
950 
951 	assert(opts != NULL);
952 	if (enable_ssl) {
953 		_opts_get_impl_opts(opts, &impl_opts, &g_ssl_impl_opts);
954 	} else {
955 		_opts_get_impl_opts(opts, &impl_opts, &g_posix_impl_opts);
956 	}
957 
958 	if (ip == NULL) {
959 		return NULL;
960 	}
961 	if (ip[0] == '[') {
962 		snprintf(buf, sizeof(buf), "%s", ip + 1);
963 		p = strchr(buf, ']');
964 		if (p != NULL) {
965 			*p = '\0';
966 		}
967 		ip = (const char *) &buf[0];
968 	}
969 
970 	snprintf(portnum, sizeof portnum, "%d", port);
971 	memset(&hints, 0, sizeof hints);
972 	hints.ai_family = PF_UNSPEC;
973 	hints.ai_socktype = SOCK_STREAM;
974 	hints.ai_flags = AI_NUMERICSERV;
975 	hints.ai_flags |= AI_PASSIVE;
976 	hints.ai_flags |= AI_NUMERICHOST;
977 	rc = getaddrinfo(ip, portnum, &hints, &res0);
978 	if (rc != 0) {
979 		SPDK_ERRLOG("getaddrinfo() failed %s (%d)\n", gai_strerror(rc), rc);
980 		return NULL;
981 	}
982 
983 	/* try listen */
984 	fd = -1;
985 	for (res = res0; res != NULL; res = res->ai_next) {
986 retry:
987 		fd = posix_fd_create(res, opts, &impl_opts);
988 		if (fd < 0) {
989 			continue;
990 		}
991 		if (type == SPDK_SOCK_CREATE_LISTEN) {
992 			rc = bind(fd, res->ai_addr, res->ai_addrlen);
993 			if (rc != 0) {
994 				SPDK_ERRLOG("bind() failed at port %d, errno = %d\n", port, errno);
995 				switch (errno) {
996 				case EINTR:
997 					/* interrupted? */
998 					close(fd);
999 					goto retry;
1000 				case EADDRNOTAVAIL:
1001 					SPDK_ERRLOG("IP address %s not available. "
1002 						    "Verify IP address in config file "
1003 						    "and make sure setup script is "
1004 						    "run before starting spdk app.\n", ip);
1005 				/* FALLTHROUGH */
1006 				default:
1007 					/* try next family */
1008 					close(fd);
1009 					fd = -1;
1010 					continue;
1011 				}
1012 			}
1013 			/* bind OK */
1014 			rc = listen(fd, 512);
1015 			if (rc != 0) {
1016 				SPDK_ERRLOG("listen() failed, errno = %d\n", errno);
1017 				close(fd);
1018 				fd = -1;
1019 				break;
1020 			}
1021 			enable_zcopy_impl_opts = impl_opts.enable_zerocopy_send_server;
1022 		} else if (type == SPDK_SOCK_CREATE_CONNECT) {
1023 			src_addr = SPDK_GET_FIELD(opts, src_addr, NULL, opts->opts_size);
1024 			src_port = SPDK_GET_FIELD(opts, src_port, 0, opts->opts_size);
1025 			if (src_addr != NULL || src_port != 0) {
1026 				snprintf(portnum, sizeof(portnum), "%"PRIu16, src_port);
1027 				memset(&hints, 0, sizeof hints);
1028 				hints.ai_family = AF_UNSPEC;
1029 				hints.ai_socktype = SOCK_STREAM;
1030 				hints.ai_flags = AI_NUMERICSERV | AI_NUMERICHOST | AI_PASSIVE;
1031 				rc = getaddrinfo(src_addr, src_port > 0 ? portnum : NULL,
1032 						 &hints, &src_ai);
1033 				if (rc != 0 || src_ai == NULL) {
1034 					SPDK_ERRLOG("getaddrinfo() failed %s (%d)\n",
1035 						    rc != 0 ? gai_strerror(rc) : "", rc);
1036 					close(fd);
1037 					fd = -1;
1038 					break;
1039 				}
1040 				rc = bind(fd, src_ai->ai_addr, src_ai->ai_addrlen);
1041 				if (rc != 0) {
1042 					SPDK_ERRLOG("bind() failed errno %d (%s:%s)\n", errno,
1043 						    src_addr ? src_addr : "", portnum);
1044 					close(fd);
1045 					fd = -1;
1046 					break;
1047 				}
1048 				freeaddrinfo(src_ai);
1049 				src_ai = NULL;
1050 			}
1051 			rc = connect(fd, res->ai_addr, res->ai_addrlen);
1052 			if (rc != 0) {
1053 				SPDK_ERRLOG("connect() failed, errno = %d\n", errno);
1054 				/* try next family */
1055 				close(fd);
1056 				fd = -1;
1057 				continue;
1058 			}
1059 			enable_zcopy_impl_opts = impl_opts.enable_zerocopy_send_client;
1060 			if (enable_ssl) {
1061 				ctx = posix_sock_create_ssl_context(TLS_client_method(), opts, &impl_opts);
1062 				if (!ctx) {
1063 					SPDK_ERRLOG("posix_sock_create_ssl_context() failed, errno = %d\n", errno);
1064 					close(fd);
1065 					fd = -1;
1066 					break;
1067 				}
1068 				ssl = ssl_sock_setup_connect(ctx, fd);
1069 				if (!ssl) {
1070 					SPDK_ERRLOG("ssl_sock_setup_connect() failed, errno = %d\n", errno);
1071 					close(fd);
1072 					fd = -1;
1073 					SSL_CTX_free(ctx);
1074 					break;
1075 				}
1076 			}
1077 		}
1078 
1079 		flag = fcntl(fd, F_GETFL);
1080 		if (fcntl(fd, F_SETFL, flag | O_NONBLOCK) < 0) {
1081 			SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%d)\n", fd, errno);
1082 			SSL_free(ssl);
1083 			SSL_CTX_free(ctx);
1084 			close(fd);
1085 			fd = -1;
1086 			break;
1087 		}
1088 		break;
1089 	}
1090 	freeaddrinfo(res0);
1091 
1092 	if (fd < 0) {
1093 		return NULL;
1094 	}
1095 
1096 	/* Only enable zero copy for non-loopback and non-ssl sockets. */
1097 	enable_zcopy_user_opts = opts->zcopy && !spdk_net_is_loopback(fd) && !enable_ssl;
1098 
1099 	sock = posix_sock_alloc(fd, &impl_opts, enable_zcopy_user_opts && enable_zcopy_impl_opts);
1100 	if (sock == NULL) {
1101 		SPDK_ERRLOG("sock allocation failed\n");
1102 		SSL_free(ssl);
1103 		SSL_CTX_free(ctx);
1104 		close(fd);
1105 		return NULL;
1106 	}
1107 
1108 	if (ctx) {
1109 		sock->ctx = ctx;
1110 	}
1111 
1112 	if (ssl) {
1113 		sock->ssl = ssl;
1114 		SSL_set_app_data(ssl, &sock->base.impl_opts);
1115 	}
1116 
1117 	return &sock->base;
1118 }
1119 
1120 static struct spdk_sock *
1121 posix_sock_listen(const char *ip, int port, struct spdk_sock_opts *opts)
1122 {
1123 	return posix_sock_create(ip, port, SPDK_SOCK_CREATE_LISTEN, opts, false);
1124 }
1125 
1126 static struct spdk_sock *
1127 posix_sock_connect(const char *ip, int port, struct spdk_sock_opts *opts)
1128 {
1129 	return posix_sock_create(ip, port, SPDK_SOCK_CREATE_CONNECT, opts, false);
1130 }
1131 
1132 static struct spdk_sock *
1133 _posix_sock_accept(struct spdk_sock *_sock, bool enable_ssl)
1134 {
1135 	struct spdk_posix_sock		*sock = __posix_sock(_sock);
1136 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(sock->base.group_impl);
1137 	struct sockaddr_storage		sa;
1138 	socklen_t			salen;
1139 	int				rc, fd;
1140 	struct spdk_posix_sock		*new_sock;
1141 	int				flag;
1142 	SSL_CTX *ctx = 0;
1143 	SSL *ssl = 0;
1144 
1145 	memset(&sa, 0, sizeof(sa));
1146 	salen = sizeof(sa);
1147 
1148 	assert(sock != NULL);
1149 
1150 	/* epoll_wait will trigger again if there is more than one request */
1151 	if (group && sock->socket_has_data) {
1152 		sock->socket_has_data = false;
1153 		TAILQ_REMOVE(&group->socks_with_data, sock, link);
1154 	}
1155 
1156 	rc = accept(sock->fd, (struct sockaddr *)&sa, &salen);
1157 
1158 	if (rc == -1) {
1159 		return NULL;
1160 	}
1161 
1162 	fd = rc;
1163 
1164 	flag = fcntl(fd, F_GETFL);
1165 	if ((!(flag & O_NONBLOCK)) && (fcntl(fd, F_SETFL, flag | O_NONBLOCK) < 0)) {
1166 		SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%d)\n", fd, errno);
1167 		close(fd);
1168 		return NULL;
1169 	}
1170 
1171 #if defined(SO_PRIORITY)
1172 	/* The priority is not inherited, so call this function again */
1173 	if (sock->base.opts.priority) {
1174 		rc = setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &sock->base.opts.priority, sizeof(int));
1175 		if (rc != 0) {
1176 			close(fd);
1177 			return NULL;
1178 		}
1179 	}
1180 #endif
1181 
1182 	/* Establish SSL connection */
1183 	if (enable_ssl) {
1184 		ctx = posix_sock_create_ssl_context(TLS_server_method(), &sock->base.opts, &sock->base.impl_opts);
1185 		if (!ctx) {
1186 			SPDK_ERRLOG("posix_sock_create_ssl_context() failed, errno = %d\n", errno);
1187 			close(fd);
1188 			return NULL;
1189 		}
1190 		ssl = ssl_sock_setup_accept(ctx, fd);
1191 		if (!ssl) {
1192 			SPDK_ERRLOG("ssl_sock_setup_accept() failed, errno = %d\n", errno);
1193 			close(fd);
1194 			SSL_CTX_free(ctx);
1195 			return NULL;
1196 		}
1197 	}
1198 
1199 	/* Inherit the zero copy feature from the listen socket */
1200 	new_sock = posix_sock_alloc(fd, &sock->base.impl_opts, sock->zcopy);
1201 	if (new_sock == NULL) {
1202 		close(fd);
1203 		SSL_free(ssl);
1204 		SSL_CTX_free(ctx);
1205 		return NULL;
1206 	}
1207 
1208 	if (ctx) {
1209 		new_sock->ctx = ctx;
1210 	}
1211 
1212 	if (ssl) {
1213 		new_sock->ssl = ssl;
1214 		SSL_set_app_data(ssl, &new_sock->base.impl_opts);
1215 	}
1216 
1217 	return &new_sock->base;
1218 }
1219 
1220 static struct spdk_sock *
1221 posix_sock_accept(struct spdk_sock *_sock)
1222 {
1223 	return _posix_sock_accept(_sock, false);
1224 }
1225 
1226 static int
1227 posix_sock_close(struct spdk_sock *_sock)
1228 {
1229 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1230 	void *pipe_buf;
1231 
1232 	assert(TAILQ_EMPTY(&_sock->pending_reqs));
1233 
1234 	if (sock->ssl != NULL) {
1235 		SSL_shutdown(sock->ssl);
1236 	}
1237 
1238 	/* If the socket fails to close, the best choice is to
1239 	 * leak the fd but continue to free the rest of the sock
1240 	 * memory. */
1241 	close(sock->fd);
1242 
1243 	SSL_free(sock->ssl);
1244 	SSL_CTX_free(sock->ctx);
1245 
1246 	pipe_buf = spdk_pipe_destroy(sock->recv_pipe);
1247 	free(pipe_buf);
1248 	free(sock);
1249 
1250 	return 0;
1251 }
1252 
1253 #ifdef SPDK_ZEROCOPY
1254 static int
1255 _sock_check_zcopy(struct spdk_sock *sock)
1256 {
1257 	struct spdk_posix_sock *psock = __posix_sock(sock);
1258 	struct msghdr msgh = {};
1259 	uint8_t buf[sizeof(struct cmsghdr) + sizeof(struct sock_extended_err)];
1260 	ssize_t rc;
1261 	struct sock_extended_err *serr;
1262 	struct cmsghdr *cm;
1263 	uint32_t idx;
1264 	struct spdk_sock_request *req, *treq;
1265 	bool found;
1266 
1267 	msgh.msg_control = buf;
1268 	msgh.msg_controllen = sizeof(buf);
1269 
1270 	while (true) {
1271 		rc = recvmsg(psock->fd, &msgh, MSG_ERRQUEUE);
1272 
1273 		if (rc < 0) {
1274 			if (errno == EWOULDBLOCK || errno == EAGAIN) {
1275 				return 0;
1276 			}
1277 
1278 			if (!TAILQ_EMPTY(&sock->pending_reqs)) {
1279 				SPDK_ERRLOG("Attempting to receive from ERRQUEUE yielded error, but pending list still has orphaned entries\n");
1280 			} else {
1281 				SPDK_WARNLOG("Recvmsg yielded an error!\n");
1282 			}
1283 			return 0;
1284 		}
1285 
1286 		cm = CMSG_FIRSTHDR(&msgh);
1287 		if (!(cm &&
1288 		      ((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) ||
1289 		       (cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR)))) {
1290 			SPDK_WARNLOG("Unexpected cmsg level or type!\n");
1291 			return 0;
1292 		}
1293 
1294 		serr = (struct sock_extended_err *)CMSG_DATA(cm);
1295 		if (serr->ee_errno != 0 || serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY) {
1296 			SPDK_WARNLOG("Unexpected extended error origin\n");
1297 			return 0;
1298 		}
1299 
1300 		/* Most of the time, the pending_reqs array is in the exact
1301 		 * order we need such that all of the requests to complete are
1302 		 * in order, in the front. It is guaranteed that all requests
1303 		 * belonging to the same sendmsg call are sequential, so once
1304 		 * we encounter one match we can stop looping as soon as a
1305 		 * non-match is found.
1306 		 */
1307 		idx = serr->ee_info;
1308 		while (true) {
1309 			found = false;
1310 			TAILQ_FOREACH_SAFE(req, &sock->pending_reqs, internal.link, treq) {
1311 				if (!req->internal.is_zcopy) {
1312 					/* This wasn't a zcopy request. It was just waiting in line to complete */
1313 					rc = spdk_sock_request_put(sock, req, 0);
1314 					if (rc < 0) {
1315 						return rc;
1316 					}
1317 				} else if (req->internal.offset == idx) {
1318 					found = true;
1319 					rc = spdk_sock_request_put(sock, req, 0);
1320 					if (rc < 0) {
1321 						return rc;
1322 					}
1323 				} else if (found) {
1324 					break;
1325 				}
1326 			}
1327 
1328 			if (idx == serr->ee_data) {
1329 				break;
1330 			}
1331 
1332 			if (idx == UINT32_MAX) {
1333 				idx = 0;
1334 			} else {
1335 				idx++;
1336 			}
1337 		}
1338 	}
1339 
1340 	return 0;
1341 }
1342 #endif
1343 
1344 static int
1345 _sock_flush(struct spdk_sock *sock)
1346 {
1347 	struct spdk_posix_sock *psock = __posix_sock(sock);
1348 	struct msghdr msg = {};
1349 	int flags;
1350 	struct iovec iovs[IOV_BATCH_SIZE];
1351 	int iovcnt;
1352 	int retval;
1353 	struct spdk_sock_request *req;
1354 	int i;
1355 	ssize_t rc, sent;
1356 	unsigned int offset;
1357 	size_t len;
1358 	bool is_zcopy = false;
1359 
1360 	/* Can't flush from within a callback or we end up with recursive calls */
1361 	if (sock->cb_cnt > 0) {
1362 		errno = EAGAIN;
1363 		return -1;
1364 	}
1365 
1366 #ifdef SPDK_ZEROCOPY
1367 	if (psock->zcopy) {
1368 		flags = MSG_ZEROCOPY | MSG_NOSIGNAL;
1369 	} else
1370 #endif
1371 	{
1372 		flags = MSG_NOSIGNAL;
1373 	}
1374 
1375 	iovcnt = spdk_sock_prep_reqs(sock, iovs, 0, NULL, &flags);
1376 	if (iovcnt == 0) {
1377 		return 0;
1378 	}
1379 
1380 #ifdef SPDK_ZEROCOPY
1381 	is_zcopy = flags & MSG_ZEROCOPY;
1382 #endif
1383 
1384 	/* Perform the vectored write */
1385 	msg.msg_iov = iovs;
1386 	msg.msg_iovlen = iovcnt;
1387 
1388 	if (psock->ssl) {
1389 		rc = SSL_writev(psock->ssl, iovs, iovcnt);
1390 	} else {
1391 		rc = sendmsg(psock->fd, &msg, flags);
1392 	}
1393 	if (rc <= 0) {
1394 		if (rc == 0 || errno == EAGAIN || errno == EWOULDBLOCK || (errno == ENOBUFS && psock->zcopy)) {
1395 			errno = EAGAIN;
1396 		}
1397 		return -1;
1398 	}
1399 
1400 	sent = rc;
1401 
1402 	if (is_zcopy) {
1403 		/* Handling overflow case, because we use psock->sendmsg_idx - 1 for the
1404 		 * req->internal.offset, so sendmsg_idx should not be zero  */
1405 		if (spdk_unlikely(psock->sendmsg_idx == UINT32_MAX)) {
1406 			psock->sendmsg_idx = 1;
1407 		} else {
1408 			psock->sendmsg_idx++;
1409 		}
1410 	}
1411 
1412 	/* Consume the requests that were actually written */
1413 	req = TAILQ_FIRST(&sock->queued_reqs);
1414 	while (req) {
1415 		offset = req->internal.offset;
1416 
1417 		/* req->internal.is_zcopy is true when the whole req or part of it is sent with zerocopy */
1418 		req->internal.is_zcopy = is_zcopy;
1419 
1420 		for (i = 0; i < req->iovcnt; i++) {
1421 			/* Advance by the offset first */
1422 			if (offset >= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len) {
1423 				offset -= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len;
1424 				continue;
1425 			}
1426 
1427 			/* Calculate the remaining length of this element */
1428 			len = SPDK_SOCK_REQUEST_IOV(req, i)->iov_len - offset;
1429 
1430 			if (len > (size_t)rc) {
1431 				/* This element was partially sent. */
1432 				req->internal.offset += rc;
1433 				return sent;
1434 			}
1435 
1436 			offset = 0;
1437 			req->internal.offset += len;
1438 			rc -= len;
1439 		}
1440 
1441 		/* Handled a full request. */
1442 		spdk_sock_request_pend(sock, req);
1443 
1444 		if (!req->internal.is_zcopy && req == TAILQ_FIRST(&sock->pending_reqs)) {
1445 			/* The sendmsg syscall above isn't currently asynchronous,
1446 			* so it's already done. */
1447 			retval = spdk_sock_request_put(sock, req, 0);
1448 			if (retval) {
1449 				break;
1450 			}
1451 		} else {
1452 			/* Re-use the offset field to hold the sendmsg call index. The
1453 			 * index is 0 based, so subtract one here because we've already
1454 			 * incremented above. */
1455 			req->internal.offset = psock->sendmsg_idx - 1;
1456 		}
1457 
1458 		if (rc == 0) {
1459 			break;
1460 		}
1461 
1462 		req = TAILQ_FIRST(&sock->queued_reqs);
1463 	}
1464 
1465 	return sent;
1466 }
1467 
1468 static int
1469 posix_sock_flush(struct spdk_sock *sock)
1470 {
1471 #ifdef SPDK_ZEROCOPY
1472 	struct spdk_posix_sock *psock = __posix_sock(sock);
1473 
1474 	if (psock->zcopy && !TAILQ_EMPTY(&sock->pending_reqs)) {
1475 		_sock_check_zcopy(sock);
1476 	}
1477 #endif
1478 
1479 	return _sock_flush(sock);
1480 }
1481 
1482 static ssize_t
1483 posix_sock_recv_from_pipe(struct spdk_posix_sock *sock, struct iovec *diov, int diovcnt)
1484 {
1485 	struct iovec siov[2];
1486 	int sbytes;
1487 	ssize_t bytes;
1488 	struct spdk_posix_sock_group_impl *group;
1489 
1490 	sbytes = spdk_pipe_reader_get_buffer(sock->recv_pipe, sock->recv_buf_sz, siov);
1491 	if (sbytes < 0) {
1492 		errno = EINVAL;
1493 		return -1;
1494 	} else if (sbytes == 0) {
1495 		errno = EAGAIN;
1496 		return -1;
1497 	}
1498 
1499 	bytes = spdk_iovcpy(siov, 2, diov, diovcnt);
1500 
1501 	if (bytes == 0) {
1502 		/* The only way this happens is if diov is 0 length */
1503 		errno = EINVAL;
1504 		return -1;
1505 	}
1506 
1507 	spdk_pipe_reader_advance(sock->recv_pipe, bytes);
1508 
1509 	/* If we drained the pipe, mark it appropriately */
1510 	if (spdk_pipe_reader_bytes_available(sock->recv_pipe) == 0) {
1511 		assert(sock->pipe_has_data == true);
1512 
1513 		group = __posix_group_impl(sock->base.group_impl);
1514 		if (group && !sock->socket_has_data) {
1515 			TAILQ_REMOVE(&group->socks_with_data, sock, link);
1516 		}
1517 
1518 		sock->pipe_has_data = false;
1519 	}
1520 
1521 	return bytes;
1522 }
1523 
1524 static inline ssize_t
1525 posix_sock_read(struct spdk_posix_sock *sock)
1526 {
1527 	struct iovec iov[2];
1528 	int bytes_avail, bytes_recvd;
1529 	struct spdk_posix_sock_group_impl *group;
1530 
1531 	bytes_avail = spdk_pipe_writer_get_buffer(sock->recv_pipe, sock->recv_buf_sz, iov);
1532 
1533 	if (bytes_avail <= 0) {
1534 		return bytes_avail;
1535 	}
1536 
1537 	if (sock->ssl) {
1538 		bytes_recvd = SSL_readv(sock->ssl, iov, 2);
1539 	} else {
1540 		bytes_recvd = readv(sock->fd, iov, 2);
1541 	}
1542 
1543 	assert(sock->pipe_has_data == false);
1544 
1545 	if (bytes_recvd <= 0) {
1546 		/* Errors count as draining the socket data */
1547 		if (sock->base.group_impl && sock->socket_has_data) {
1548 			group = __posix_group_impl(sock->base.group_impl);
1549 			TAILQ_REMOVE(&group->socks_with_data, sock, link);
1550 		}
1551 
1552 		sock->socket_has_data = false;
1553 
1554 		return bytes_recvd;
1555 	}
1556 
1557 	spdk_pipe_writer_advance(sock->recv_pipe, bytes_recvd);
1558 
1559 #if DEBUG
1560 	if (sock->base.group_impl) {
1561 		assert(sock->socket_has_data == true);
1562 	}
1563 #endif
1564 
1565 	sock->pipe_has_data = true;
1566 	if (bytes_recvd < bytes_avail) {
1567 		/* We drained the kernel socket entirely. */
1568 		sock->socket_has_data = false;
1569 	}
1570 
1571 	return bytes_recvd;
1572 }
1573 
1574 static ssize_t
1575 posix_sock_readv(struct spdk_sock *_sock, struct iovec *iov, int iovcnt)
1576 {
1577 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1578 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(sock->base.group_impl);
1579 	int rc, i;
1580 	size_t len;
1581 
1582 	if (sock->recv_pipe == NULL) {
1583 		assert(sock->pipe_has_data == false);
1584 		if (group && sock->socket_has_data) {
1585 			sock->socket_has_data = false;
1586 			TAILQ_REMOVE(&group->socks_with_data, sock, link);
1587 		}
1588 		if (sock->ssl) {
1589 			return SSL_readv(sock->ssl, iov, iovcnt);
1590 		} else {
1591 			return readv(sock->fd, iov, iovcnt);
1592 		}
1593 	}
1594 
1595 	/* If the socket is not in a group, we must assume it always has
1596 	 * data waiting for us because it is not epolled */
1597 	if (!sock->pipe_has_data && (group == NULL || sock->socket_has_data)) {
1598 		/* If the user is receiving a sufficiently large amount of data,
1599 		 * receive directly to their buffers. */
1600 		len = 0;
1601 		for (i = 0; i < iovcnt; i++) {
1602 			len += iov[i].iov_len;
1603 		}
1604 
1605 		if (len >= MIN_SOCK_PIPE_SIZE) {
1606 			/* TODO: Should this detect if kernel socket is drained? */
1607 			if (sock->ssl) {
1608 				return SSL_readv(sock->ssl, iov, iovcnt);
1609 			} else {
1610 				return readv(sock->fd, iov, iovcnt);
1611 			}
1612 		}
1613 
1614 		/* Otherwise, do a big read into our pipe */
1615 		rc = posix_sock_read(sock);
1616 		if (rc <= 0) {
1617 			return rc;
1618 		}
1619 	}
1620 
1621 	return posix_sock_recv_from_pipe(sock, iov, iovcnt);
1622 }
1623 
1624 static ssize_t
1625 posix_sock_recv(struct spdk_sock *sock, void *buf, size_t len)
1626 {
1627 	struct iovec iov[1];
1628 
1629 	iov[0].iov_base = buf;
1630 	iov[0].iov_len = len;
1631 
1632 	return posix_sock_readv(sock, iov, 1);
1633 }
1634 
1635 static ssize_t
1636 posix_sock_writev(struct spdk_sock *_sock, struct iovec *iov, int iovcnt)
1637 {
1638 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1639 	int rc;
1640 
1641 	/* In order to process a writev, we need to flush any asynchronous writes
1642 	 * first. */
1643 	rc = _sock_flush(_sock);
1644 	if (rc < 0) {
1645 		return rc;
1646 	}
1647 
1648 	if (!TAILQ_EMPTY(&_sock->queued_reqs)) {
1649 		/* We weren't able to flush all requests */
1650 		errno = EAGAIN;
1651 		return -1;
1652 	}
1653 
1654 	if (sock->ssl) {
1655 		return SSL_writev(sock->ssl, iov, iovcnt);
1656 	} else {
1657 		return writev(sock->fd, iov, iovcnt);
1658 	}
1659 }
1660 
1661 static int
1662 posix_sock_recv_next(struct spdk_sock *_sock, void **buf, void **ctx)
1663 {
1664 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1665 	struct iovec iov;
1666 	ssize_t rc;
1667 
1668 	if (sock->recv_pipe != NULL) {
1669 		errno = ENOTSUP;
1670 		return -1;
1671 	}
1672 
1673 	iov.iov_len = spdk_sock_group_get_buf(_sock->group_impl->group, &iov.iov_base, ctx);
1674 	if (iov.iov_len == 0) {
1675 		errno = ENOBUFS;
1676 		return -1;
1677 	}
1678 
1679 	rc = posix_sock_readv(_sock, &iov, 1);
1680 	if (rc <= 0) {
1681 		spdk_sock_group_provide_buf(_sock->group_impl->group, iov.iov_base, iov.iov_len, *ctx);
1682 		return rc;
1683 	}
1684 
1685 	*buf = iov.iov_base;
1686 
1687 	return rc;
1688 }
1689 
1690 static void
1691 posix_sock_writev_async(struct spdk_sock *sock, struct spdk_sock_request *req)
1692 {
1693 	int rc;
1694 
1695 	spdk_sock_request_queue(sock, req);
1696 
1697 	/* If there are a sufficient number queued, just flush them out immediately. */
1698 	if (sock->queued_iovcnt >= IOV_BATCH_SIZE) {
1699 		rc = _sock_flush(sock);
1700 		if (rc < 0 && errno != EAGAIN) {
1701 			spdk_sock_abort_requests(sock);
1702 		}
1703 	}
1704 }
1705 
1706 static int
1707 posix_sock_set_recvlowat(struct spdk_sock *_sock, int nbytes)
1708 {
1709 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1710 	int val;
1711 	int rc;
1712 
1713 	assert(sock != NULL);
1714 
1715 	val = nbytes;
1716 	rc = setsockopt(sock->fd, SOL_SOCKET, SO_RCVLOWAT, &val, sizeof val);
1717 	if (rc != 0) {
1718 		return -1;
1719 	}
1720 	return 0;
1721 }
1722 
1723 static bool
1724 posix_sock_is_ipv6(struct spdk_sock *_sock)
1725 {
1726 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1727 	struct sockaddr_storage sa;
1728 	socklen_t salen;
1729 	int rc;
1730 
1731 	assert(sock != NULL);
1732 
1733 	memset(&sa, 0, sizeof sa);
1734 	salen = sizeof sa;
1735 	rc = getsockname(sock->fd, (struct sockaddr *) &sa, &salen);
1736 	if (rc != 0) {
1737 		SPDK_ERRLOG("getsockname() failed (errno=%d)\n", errno);
1738 		return false;
1739 	}
1740 
1741 	return (sa.ss_family == AF_INET6);
1742 }
1743 
1744 static bool
1745 posix_sock_is_ipv4(struct spdk_sock *_sock)
1746 {
1747 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1748 	struct sockaddr_storage sa;
1749 	socklen_t salen;
1750 	int rc;
1751 
1752 	assert(sock != NULL);
1753 
1754 	memset(&sa, 0, sizeof sa);
1755 	salen = sizeof sa;
1756 	rc = getsockname(sock->fd, (struct sockaddr *) &sa, &salen);
1757 	if (rc != 0) {
1758 		SPDK_ERRLOG("getsockname() failed (errno=%d)\n", errno);
1759 		return false;
1760 	}
1761 
1762 	return (sa.ss_family == AF_INET);
1763 }
1764 
1765 static bool
1766 posix_sock_is_connected(struct spdk_sock *_sock)
1767 {
1768 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1769 	uint8_t byte;
1770 	int rc;
1771 
1772 	rc = recv(sock->fd, &byte, 1, MSG_PEEK);
1773 	if (rc == 0) {
1774 		return false;
1775 	}
1776 
1777 	if (rc < 0) {
1778 		if (errno == EAGAIN || errno == EWOULDBLOCK) {
1779 			return true;
1780 		}
1781 
1782 		return false;
1783 	}
1784 
1785 	return true;
1786 }
1787 
1788 static struct spdk_sock_group_impl *
1789 posix_sock_group_impl_get_optimal(struct spdk_sock *_sock, struct spdk_sock_group_impl *hint)
1790 {
1791 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1792 	struct spdk_sock_group_impl *group_impl;
1793 
1794 	if (sock->placement_id != -1) {
1795 		spdk_sock_map_lookup(&g_map, sock->placement_id, &group_impl, hint);
1796 		return group_impl;
1797 	}
1798 
1799 	return NULL;
1800 }
1801 
1802 static struct spdk_sock_group_impl *
1803 _sock_group_impl_create(uint32_t enable_placement_id)
1804 {
1805 	struct spdk_posix_sock_group_impl *group_impl;
1806 	int fd;
1807 
1808 #if defined(SPDK_EPOLL)
1809 	fd = epoll_create1(0);
1810 #elif defined(SPDK_KEVENT)
1811 	fd = kqueue();
1812 #endif
1813 	if (fd == -1) {
1814 		return NULL;
1815 	}
1816 
1817 	group_impl = calloc(1, sizeof(*group_impl));
1818 	if (group_impl == NULL) {
1819 		SPDK_ERRLOG("group_impl allocation failed\n");
1820 		close(fd);
1821 		return NULL;
1822 	}
1823 
1824 	group_impl->pipe_group = spdk_pipe_group_create();
1825 	if (group_impl->pipe_group == NULL) {
1826 		SPDK_ERRLOG("pipe_group allocation failed\n");
1827 		free(group_impl);
1828 		close(fd);
1829 		return NULL;
1830 	}
1831 
1832 	group_impl->fd = fd;
1833 	TAILQ_INIT(&group_impl->socks_with_data);
1834 	group_impl->placement_id = -1;
1835 
1836 	if (enable_placement_id == PLACEMENT_CPU) {
1837 		spdk_sock_map_insert(&g_map, spdk_env_get_current_core(), &group_impl->base);
1838 		group_impl->placement_id = spdk_env_get_current_core();
1839 	}
1840 
1841 	return &group_impl->base;
1842 }
1843 
1844 static struct spdk_sock_group_impl *
1845 posix_sock_group_impl_create(void)
1846 {
1847 	return _sock_group_impl_create(g_posix_impl_opts.enable_placement_id);
1848 }
1849 
1850 static struct spdk_sock_group_impl *
1851 ssl_sock_group_impl_create(void)
1852 {
1853 	return _sock_group_impl_create(g_ssl_impl_opts.enable_placement_id);
1854 }
1855 
1856 static void
1857 posix_sock_mark(struct spdk_posix_sock_group_impl *group, struct spdk_posix_sock *sock,
1858 		int placement_id)
1859 {
1860 #if defined(SO_MARK)
1861 	int rc;
1862 
1863 	rc = setsockopt(sock->fd, SOL_SOCKET, SO_MARK,
1864 			&placement_id, sizeof(placement_id));
1865 	if (rc != 0) {
1866 		/* Not fatal */
1867 		SPDK_ERRLOG("Error setting SO_MARK\n");
1868 		return;
1869 	}
1870 
1871 	rc = spdk_sock_map_insert(&g_map, placement_id, &group->base);
1872 	if (rc != 0) {
1873 		/* Not fatal */
1874 		SPDK_ERRLOG("Failed to insert sock group into map: %d\n", rc);
1875 		return;
1876 	}
1877 
1878 	sock->placement_id = placement_id;
1879 #endif
1880 }
1881 
1882 static void
1883 posix_sock_update_mark(struct spdk_sock_group_impl *_group, struct spdk_sock *_sock)
1884 {
1885 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
1886 
1887 	if (group->placement_id == -1) {
1888 		group->placement_id = spdk_sock_map_find_free(&g_map);
1889 
1890 		/* If a free placement id is found, update existing sockets in this group */
1891 		if (group->placement_id != -1) {
1892 			struct spdk_sock  *sock, *tmp;
1893 
1894 			TAILQ_FOREACH_SAFE(sock, &_group->socks, link, tmp) {
1895 				posix_sock_mark(group, __posix_sock(sock), group->placement_id);
1896 			}
1897 		}
1898 	}
1899 
1900 	if (group->placement_id != -1) {
1901 		/*
1902 		 * group placement id is already determined for this poll group.
1903 		 * Mark socket with group's placement id.
1904 		 */
1905 		posix_sock_mark(group, __posix_sock(_sock), group->placement_id);
1906 	}
1907 }
1908 
1909 static int
1910 posix_sock_group_impl_add_sock(struct spdk_sock_group_impl *_group, struct spdk_sock *_sock)
1911 {
1912 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
1913 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1914 	int rc;
1915 
1916 #if defined(SPDK_EPOLL)
1917 	struct epoll_event event;
1918 
1919 	memset(&event, 0, sizeof(event));
1920 	/* EPOLLERR is always on even if we don't set it, but be explicit for clarity */
1921 	event.events = EPOLLIN | EPOLLERR;
1922 	if (spdk_interrupt_mode_is_enabled()) {
1923 		event.events |= EPOLLOUT;
1924 	}
1925 
1926 	event.data.ptr = sock;
1927 
1928 	rc = epoll_ctl(group->fd, EPOLL_CTL_ADD, sock->fd, &event);
1929 #elif defined(SPDK_KEVENT)
1930 	struct kevent event;
1931 	struct timespec ts = {0};
1932 
1933 	EV_SET(&event, sock->fd, EVFILT_READ, EV_ADD, 0, 0, sock);
1934 
1935 	rc = kevent(group->fd, &event, 1, NULL, 0, &ts);
1936 #endif
1937 
1938 	if (rc != 0) {
1939 		return rc;
1940 	}
1941 
1942 	/* switched from another polling group due to scheduling */
1943 	if (spdk_unlikely(sock->recv_pipe != NULL  &&
1944 			  (spdk_pipe_reader_bytes_available(sock->recv_pipe) > 0))) {
1945 		sock->pipe_has_data = true;
1946 		sock->socket_has_data = false;
1947 		TAILQ_INSERT_TAIL(&group->socks_with_data, sock, link);
1948 	} else if (sock->recv_pipe != NULL) {
1949 		rc = spdk_pipe_group_add(group->pipe_group, sock->recv_pipe);
1950 		assert(rc == 0);
1951 	}
1952 
1953 	if (_sock->impl_opts.enable_placement_id == PLACEMENT_MARK) {
1954 		posix_sock_update_mark(_group, _sock);
1955 	} else if (sock->placement_id != -1) {
1956 		rc = spdk_sock_map_insert(&g_map, sock->placement_id, &group->base);
1957 		if (rc != 0) {
1958 			SPDK_ERRLOG("Failed to insert sock group into map: %d\n", rc);
1959 			/* Do not treat this as an error. The system will continue running. */
1960 		}
1961 	}
1962 
1963 	return rc;
1964 }
1965 
1966 static int
1967 posix_sock_group_impl_remove_sock(struct spdk_sock_group_impl *_group, struct spdk_sock *_sock)
1968 {
1969 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
1970 	struct spdk_posix_sock *sock = __posix_sock(_sock);
1971 	int rc;
1972 
1973 	if (sock->pipe_has_data || sock->socket_has_data) {
1974 		TAILQ_REMOVE(&group->socks_with_data, sock, link);
1975 		sock->pipe_has_data = false;
1976 		sock->socket_has_data = false;
1977 	} else if (sock->recv_pipe != NULL) {
1978 		rc = spdk_pipe_group_remove(group->pipe_group, sock->recv_pipe);
1979 		assert(rc == 0);
1980 	}
1981 
1982 	if (sock->placement_id != -1) {
1983 		spdk_sock_map_release(&g_map, sock->placement_id);
1984 	}
1985 
1986 #if defined(SPDK_EPOLL)
1987 	struct epoll_event event;
1988 
1989 	/* Event parameter is ignored but some old kernel version still require it. */
1990 	rc = epoll_ctl(group->fd, EPOLL_CTL_DEL, sock->fd, &event);
1991 #elif defined(SPDK_KEVENT)
1992 	struct kevent event;
1993 	struct timespec ts = {0};
1994 
1995 	EV_SET(&event, sock->fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
1996 
1997 	rc = kevent(group->fd, &event, 1, NULL, 0, &ts);
1998 	if (rc == 0 && event.flags & EV_ERROR) {
1999 		rc = -1;
2000 		errno = event.data;
2001 	}
2002 #endif
2003 
2004 	spdk_sock_abort_requests(_sock);
2005 
2006 	return rc;
2007 }
2008 
2009 static int
2010 posix_sock_group_impl_poll(struct spdk_sock_group_impl *_group, int max_events,
2011 			   struct spdk_sock **socks)
2012 {
2013 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
2014 	struct spdk_sock *sock, *tmp;
2015 	int num_events, i, rc;
2016 	struct spdk_posix_sock *psock, *ptmp;
2017 #if defined(SPDK_EPOLL)
2018 	struct epoll_event events[MAX_EVENTS_PER_POLL];
2019 #elif defined(SPDK_KEVENT)
2020 	struct kevent events[MAX_EVENTS_PER_POLL];
2021 	struct timespec ts = {0};
2022 #endif
2023 
2024 #ifdef SPDK_ZEROCOPY
2025 	/* When all of the following conditions are met
2026 	 * - non-blocking socket
2027 	 * - zero copy is enabled
2028 	 * - interrupts suppressed (i.e. busy polling)
2029 	 * - the NIC tx queue is full at the time sendmsg() is called
2030 	 * - epoll_wait determines there is an EPOLLIN event for the socket
2031 	 * then we can get into a situation where data we've sent is queued
2032 	 * up in the kernel network stack, but interrupts have been suppressed
2033 	 * because other traffic is flowing so the kernel misses the signal
2034 	 * to flush the software tx queue. If there wasn't incoming data
2035 	 * pending on the socket, then epoll_wait would have been sufficient
2036 	 * to kick off the send operation, but since there is a pending event
2037 	 * epoll_wait does not trigger the necessary operation.
2038 	 *
2039 	 * We deal with this by checking for all of the above conditions and
2040 	 * additionally looking for EPOLLIN events that were not consumed from
2041 	 * the last poll loop. We take this to mean that the upper layer is
2042 	 * unable to consume them because it is blocked waiting for resources
2043 	 * to free up, and those resources are most likely freed in response
2044 	 * to a pending asynchronous write completing.
2045 	 *
2046 	 * Additionally, sockets that have the same placement_id actually share
2047 	 * an underlying hardware queue. That means polling one of them is
2048 	 * equivalent to polling all of them. As a quick mechanism to avoid
2049 	 * making extra poll() calls, stash the last placement_id during the loop
2050 	 * and only poll if it's not the same. The overwhelmingly common case
2051 	 * is that all sockets in this list have the same placement_id because
2052 	 * SPDK is intentionally grouping sockets by that value, so even
2053 	 * though this won't stop all extra calls to poll(), it's very fast
2054 	 * and will catch all of them in practice.
2055 	 */
2056 	int last_placement_id = -1;
2057 
2058 	TAILQ_FOREACH(psock, &group->socks_with_data, link) {
2059 		if (psock->zcopy && psock->placement_id >= 0 &&
2060 		    psock->placement_id != last_placement_id) {
2061 			struct pollfd pfd = {psock->fd, POLLIN | POLLERR, 0};
2062 
2063 			poll(&pfd, 1, 0);
2064 			last_placement_id = psock->placement_id;
2065 		}
2066 	}
2067 #endif
2068 
2069 	/* This must be a TAILQ_FOREACH_SAFE because while flushing,
2070 	 * a completion callback could remove the sock from the
2071 	 * group. */
2072 	TAILQ_FOREACH_SAFE(sock, &_group->socks, link, tmp) {
2073 		rc = _sock_flush(sock);
2074 		if (rc < 0 && errno != EAGAIN) {
2075 			spdk_sock_abort_requests(sock);
2076 		}
2077 	}
2078 
2079 	assert(max_events > 0);
2080 
2081 #if defined(SPDK_EPOLL)
2082 	num_events = epoll_wait(group->fd, events, max_events, 0);
2083 #elif defined(SPDK_KEVENT)
2084 	num_events = kevent(group->fd, NULL, 0, events, max_events, &ts);
2085 #endif
2086 
2087 	if (num_events == -1) {
2088 		return -1;
2089 	} else if (num_events == 0 && !TAILQ_EMPTY(&_group->socks)) {
2090 		sock = TAILQ_FIRST(&_group->socks);
2091 		psock = __posix_sock(sock);
2092 		/* poll() is called here to busy poll the queue associated with
2093 		 * first socket in list and potentially reap incoming data.
2094 		 */
2095 		if (sock->opts.priority) {
2096 			struct pollfd pfd = {0, 0, 0};
2097 
2098 			pfd.fd = psock->fd;
2099 			pfd.events = POLLIN | POLLERR;
2100 			poll(&pfd, 1, 0);
2101 		}
2102 	}
2103 
2104 	for (i = 0; i < num_events; i++) {
2105 #if defined(SPDK_EPOLL)
2106 		sock = events[i].data.ptr;
2107 		psock = __posix_sock(sock);
2108 
2109 #ifdef SPDK_ZEROCOPY
2110 		if (events[i].events & EPOLLERR) {
2111 			rc = _sock_check_zcopy(sock);
2112 			/* If the socket was closed or removed from
2113 			 * the group in response to a send ack, don't
2114 			 * add it to the array here. */
2115 			if (rc || sock->cb_fn == NULL) {
2116 				continue;
2117 			}
2118 		}
2119 #endif
2120 		if ((events[i].events & EPOLLIN) == 0) {
2121 			continue;
2122 		}
2123 
2124 #elif defined(SPDK_KEVENT)
2125 		sock = events[i].udata;
2126 		psock = __posix_sock(sock);
2127 #endif
2128 
2129 		/* If the socket is not already in the list, add it now */
2130 		if (!psock->socket_has_data && !psock->pipe_has_data) {
2131 			TAILQ_INSERT_TAIL(&group->socks_with_data, psock, link);
2132 		}
2133 		psock->socket_has_data = true;
2134 	}
2135 
2136 	num_events = 0;
2137 
2138 	TAILQ_FOREACH_SAFE(psock, &group->socks_with_data, link, ptmp) {
2139 		if (num_events == max_events) {
2140 			break;
2141 		}
2142 
2143 		/* If the socket's cb_fn is NULL, just remove it from the
2144 		 * list and do not add it to socks array */
2145 		if (spdk_unlikely(psock->base.cb_fn == NULL)) {
2146 			psock->socket_has_data = false;
2147 			psock->pipe_has_data = false;
2148 			TAILQ_REMOVE(&group->socks_with_data, psock, link);
2149 			continue;
2150 		}
2151 
2152 		socks[num_events++] = &psock->base;
2153 	}
2154 
2155 	/* Cycle the has_data list so that each time we poll things aren't
2156 	 * in the same order. Say we have 6 sockets in the list, named as follows:
2157 	 * A B C D E F
2158 	 * And all 6 sockets had epoll events, but max_events is only 3. That means
2159 	 * psock currently points at D. We want to rearrange the list to the following:
2160 	 * D E F A B C
2161 	 *
2162 	 * The variables below are named according to this example to make it easier to
2163 	 * follow the swaps.
2164 	 */
2165 	if (psock != NULL) {
2166 		struct spdk_posix_sock *pa, *pc, *pd, *pf;
2167 
2168 		/* Capture pointers to the elements we need */
2169 		pd = psock;
2170 		pc = TAILQ_PREV(pd, spdk_has_data_list, link);
2171 		pa = TAILQ_FIRST(&group->socks_with_data);
2172 		pf = TAILQ_LAST(&group->socks_with_data, spdk_has_data_list);
2173 
2174 		/* Break the link between C and D */
2175 		pc->link.tqe_next = NULL;
2176 
2177 		/* Connect F to A */
2178 		pf->link.tqe_next = pa;
2179 		pa->link.tqe_prev = &pf->link.tqe_next;
2180 
2181 		/* Fix up the list first/last pointers */
2182 		group->socks_with_data.tqh_first = pd;
2183 		group->socks_with_data.tqh_last = &pc->link.tqe_next;
2184 
2185 		/* D is in front of the list, make tqe prev pointer point to the head of list */
2186 		pd->link.tqe_prev = &group->socks_with_data.tqh_first;
2187 	}
2188 
2189 	return num_events;
2190 }
2191 
2192 static int
2193 posix_sock_group_impl_register_interrupt(struct spdk_sock_group_impl *_group, uint32_t events,
2194 		spdk_interrupt_fn fn, void *arg, const char *name)
2195 {
2196 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
2197 
2198 	group->intr = spdk_interrupt_register_for_events(group->fd, events, fn, arg, name);
2199 
2200 	return group->intr ? 0 : -1;
2201 }
2202 
2203 static void
2204 posix_sock_group_impl_unregister_interrupt(struct spdk_sock_group_impl *_group)
2205 {
2206 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
2207 
2208 	spdk_interrupt_unregister(&group->intr);
2209 }
2210 
2211 static int
2212 _sock_group_impl_close(struct spdk_sock_group_impl *_group, uint32_t enable_placement_id)
2213 {
2214 	struct spdk_posix_sock_group_impl *group = __posix_group_impl(_group);
2215 	int rc;
2216 
2217 	if (enable_placement_id == PLACEMENT_CPU) {
2218 		spdk_sock_map_release(&g_map, spdk_env_get_current_core());
2219 	}
2220 
2221 	spdk_pipe_group_destroy(group->pipe_group);
2222 	rc = close(group->fd);
2223 	free(group);
2224 	return rc;
2225 }
2226 
2227 static int
2228 posix_sock_group_impl_close(struct spdk_sock_group_impl *_group)
2229 {
2230 	return _sock_group_impl_close(_group, g_posix_impl_opts.enable_placement_id);
2231 }
2232 
2233 static int
2234 ssl_sock_group_impl_close(struct spdk_sock_group_impl *_group)
2235 {
2236 	return _sock_group_impl_close(_group, g_ssl_impl_opts.enable_placement_id);
2237 }
2238 
2239 static struct spdk_net_impl g_posix_net_impl = {
2240 	.name		= "posix",
2241 	.getaddr	= posix_sock_getaddr,
2242 	.get_interface_name = posix_sock_get_interface_name,
2243 	.get_numa_socket_id = posix_sock_get_numa_socket_id,
2244 	.connect	= posix_sock_connect,
2245 	.listen		= posix_sock_listen,
2246 	.accept		= posix_sock_accept,
2247 	.close		= posix_sock_close,
2248 	.recv		= posix_sock_recv,
2249 	.readv		= posix_sock_readv,
2250 	.writev		= posix_sock_writev,
2251 	.recv_next	= posix_sock_recv_next,
2252 	.writev_async	= posix_sock_writev_async,
2253 	.flush		= posix_sock_flush,
2254 	.set_recvlowat	= posix_sock_set_recvlowat,
2255 	.set_recvbuf	= posix_sock_set_recvbuf,
2256 	.set_sendbuf	= posix_sock_set_sendbuf,
2257 	.is_ipv6	= posix_sock_is_ipv6,
2258 	.is_ipv4	= posix_sock_is_ipv4,
2259 	.is_connected	= posix_sock_is_connected,
2260 	.group_impl_get_optimal	= posix_sock_group_impl_get_optimal,
2261 	.group_impl_create	= posix_sock_group_impl_create,
2262 	.group_impl_add_sock	= posix_sock_group_impl_add_sock,
2263 	.group_impl_remove_sock = posix_sock_group_impl_remove_sock,
2264 	.group_impl_poll	= posix_sock_group_impl_poll,
2265 	.group_impl_register_interrupt     = posix_sock_group_impl_register_interrupt,
2266 	.group_impl_unregister_interrupt  = posix_sock_group_impl_unregister_interrupt,
2267 	.group_impl_close	= posix_sock_group_impl_close,
2268 	.get_opts	= posix_sock_impl_get_opts,
2269 	.set_opts	= posix_sock_impl_set_opts,
2270 };
2271 
2272 SPDK_NET_IMPL_REGISTER_DEFAULT(posix, &g_posix_net_impl);
2273 
2274 static struct spdk_sock *
2275 ssl_sock_listen(const char *ip, int port, struct spdk_sock_opts *opts)
2276 {
2277 	return posix_sock_create(ip, port, SPDK_SOCK_CREATE_LISTEN, opts, true);
2278 }
2279 
2280 static struct spdk_sock *
2281 ssl_sock_connect(const char *ip, int port, struct spdk_sock_opts *opts)
2282 {
2283 	return posix_sock_create(ip, port, SPDK_SOCK_CREATE_CONNECT, opts, true);
2284 }
2285 
2286 static struct spdk_sock *
2287 ssl_sock_accept(struct spdk_sock *_sock)
2288 {
2289 	return _posix_sock_accept(_sock, true);
2290 }
2291 
2292 static struct spdk_net_impl g_ssl_net_impl = {
2293 	.name		= "ssl",
2294 	.getaddr	= posix_sock_getaddr,
2295 	.get_interface_name = posix_sock_get_interface_name,
2296 	.get_numa_socket_id = posix_sock_get_numa_socket_id,
2297 	.connect	= ssl_sock_connect,
2298 	.listen		= ssl_sock_listen,
2299 	.accept		= ssl_sock_accept,
2300 	.close		= posix_sock_close,
2301 	.recv		= posix_sock_recv,
2302 	.readv		= posix_sock_readv,
2303 	.writev		= posix_sock_writev,
2304 	.recv_next	= posix_sock_recv_next,
2305 	.writev_async	= posix_sock_writev_async,
2306 	.flush		= posix_sock_flush,
2307 	.set_recvlowat	= posix_sock_set_recvlowat,
2308 	.set_recvbuf	= posix_sock_set_recvbuf,
2309 	.set_sendbuf	= posix_sock_set_sendbuf,
2310 	.is_ipv6	= posix_sock_is_ipv6,
2311 	.is_ipv4	= posix_sock_is_ipv4,
2312 	.is_connected	= posix_sock_is_connected,
2313 	.group_impl_get_optimal	= posix_sock_group_impl_get_optimal,
2314 	.group_impl_create	= ssl_sock_group_impl_create,
2315 	.group_impl_add_sock	= posix_sock_group_impl_add_sock,
2316 	.group_impl_remove_sock = posix_sock_group_impl_remove_sock,
2317 	.group_impl_poll	= posix_sock_group_impl_poll,
2318 	.group_impl_register_interrupt    = posix_sock_group_impl_register_interrupt,
2319 	.group_impl_unregister_interrupt  = posix_sock_group_impl_unregister_interrupt,
2320 	.group_impl_close	= ssl_sock_group_impl_close,
2321 	.get_opts	= ssl_sock_impl_get_opts,
2322 	.set_opts	= ssl_sock_impl_set_opts,
2323 };
2324 
2325 SPDK_NET_IMPL_REGISTER(ssl, &g_ssl_net_impl);
2326 SPDK_LOG_REGISTER_COMPONENT(sock_posix)
2327