xref: /freebsd/sys/cam/ctl/ctl_ha.c (revision 2e3507c25e42292b45a5482e116d278f5515d04d)
1 /*-
2  * SPDX-License-Identifier: BSD-2-Clause
3  *
4  * Copyright (c) 2015 Alexander Motin <mav@FreeBSD.org>
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  * 1. Redistributions of source code must retain the above copyright
11  *    notice, this list of conditions and the following disclaimer,
12  *    without modification, immediately at the beginning of the file.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in the
15  *    documentation and/or other materials provided with the distribution.
16  *
17  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
18  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
19  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
20  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
21  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
22  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
26  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27  */
28 
29 #include <sys/param.h>
30 #include <sys/condvar.h>
31 #include <sys/conf.h>
32 #include <sys/eventhandler.h>
33 #include <sys/kernel.h>
34 #include <sys/kthread.h>
35 #include <sys/limits.h>
36 #include <sys/lock.h>
37 #include <sys/malloc.h>
38 #include <sys/mbuf.h>
39 #include <sys/module.h>
40 #include <sys/mutex.h>
41 #include <sys/proc.h>
42 #include <sys/queue.h>
43 #include <sys/socket.h>
44 #include <sys/socketvar.h>
45 #include <sys/sysctl.h>
46 #include <sys/systm.h>
47 #include <sys/uio.h>
48 #include <netinet/in.h>
49 #include <netinet/tcp.h>
50 #include <vm/uma.h>
51 
52 #include <cam/cam.h>
53 #include <cam/scsi/scsi_all.h>
54 #include <cam/scsi/scsi_da.h>
55 #include <cam/ctl/ctl_io.h>
56 #include <cam/ctl/ctl.h>
57 #include <cam/ctl/ctl_frontend.h>
58 #include <cam/ctl/ctl_util.h>
59 #include <cam/ctl/ctl_backend.h>
60 #include <cam/ctl/ctl_ioctl.h>
61 #include <cam/ctl/ctl_ha.h>
62 #include <cam/ctl/ctl_private.h>
63 #include <cam/ctl/ctl_debug.h>
64 #include <cam/ctl/ctl_error.h>
65 
66 struct ha_msg_wire {
67 	uint32_t	 channel;
68 	uint32_t	 length;
69 };
70 
71 struct ha_dt_msg_wire {
72 	ctl_ha_dt_cmd	command;
73 	uint32_t	size;
74 	uint8_t		*local;
75 	uint8_t		*remote;
76 };
77 
78 struct ha_softc {
79 	struct ctl_softc *ha_ctl_softc;
80 	ctl_evt_handler	 ha_handler[CTL_HA_CHAN_MAX];
81 	char		 ha_peer[128];
82 	struct sockaddr_in  ha_peer_in;
83 	struct socket	*ha_lso;
84 	struct socket	*ha_so;
85 	struct mbufq	 ha_sendq;
86 	struct mbuf	*ha_sending;
87 	struct mtx	 ha_lock;
88 	int		 ha_connect;
89 	int		 ha_listen;
90 	int		 ha_connected;
91 	int		 ha_receiving;
92 	int		 ha_wakeup;
93 	int		 ha_disconnect;
94 	int		 ha_shutdown;
95 	eventhandler_tag ha_shutdown_eh;
96 	TAILQ_HEAD(, ctl_ha_dt_req) ha_dts;
97 } ha_softc;
98 
99 static void
100 ctl_ha_conn_wake(struct ha_softc *softc)
101 {
102 
103 	mtx_lock(&softc->ha_lock);
104 	softc->ha_wakeup = 1;
105 	mtx_unlock(&softc->ha_lock);
106 	wakeup(&softc->ha_wakeup);
107 }
108 
109 static int
110 ctl_ha_lupcall(struct socket *so, void *arg, int waitflag)
111 {
112 	struct ha_softc *softc = arg;
113 
114 	ctl_ha_conn_wake(softc);
115 	return (SU_OK);
116 }
117 
118 static int
119 ctl_ha_rupcall(struct socket *so, void *arg, int waitflag)
120 {
121 	struct ha_softc *softc = arg;
122 
123 	wakeup(&softc->ha_receiving);
124 	return (SU_OK);
125 }
126 
127 static int
128 ctl_ha_supcall(struct socket *so, void *arg, int waitflag)
129 {
130 	struct ha_softc *softc = arg;
131 
132 	ctl_ha_conn_wake(softc);
133 	return (SU_OK);
134 }
135 
136 static void
137 ctl_ha_evt(struct ha_softc *softc, ctl_ha_channel ch, ctl_ha_event evt,
138     int param)
139 {
140 	int i;
141 
142 	if (ch < CTL_HA_CHAN_MAX) {
143 		if (softc->ha_handler[ch])
144 			softc->ha_handler[ch](ch, evt, param);
145 		return;
146 	}
147 	for (i = 0; i < CTL_HA_CHAN_MAX; i++) {
148 		if (softc->ha_handler[i])
149 			softc->ha_handler[i](i, evt, param);
150 	}
151 }
152 
153 static void
154 ctl_ha_close(struct ha_softc *softc)
155 {
156 	struct socket *so = softc->ha_so;
157 	int report = 0;
158 
159 	if (softc->ha_connected || softc->ha_disconnect) {
160 		softc->ha_connected = 0;
161 		mbufq_drain(&softc->ha_sendq);
162 		m_freem(softc->ha_sending);
163 		softc->ha_sending = NULL;
164 		report = 1;
165 	}
166 	if (so) {
167 		SOCKBUF_LOCK(&so->so_rcv);
168 		soupcall_clear(so, SO_RCV);
169 		while (softc->ha_receiving) {
170 			wakeup(&softc->ha_receiving);
171 			msleep(&softc->ha_receiving, SOCKBUF_MTX(&so->so_rcv),
172 			    0, "ha_rx exit", 0);
173 		}
174 		SOCKBUF_UNLOCK(&so->so_rcv);
175 		SOCKBUF_LOCK(&so->so_snd);
176 		soupcall_clear(so, SO_SND);
177 		SOCKBUF_UNLOCK(&so->so_snd);
178 		softc->ha_so = NULL;
179 		if (softc->ha_connect)
180 			pause("reconnect", hz / 2);
181 		soclose(so);
182 	}
183 	if (report) {
184 		ctl_ha_evt(softc, CTL_HA_CHAN_MAX, CTL_HA_EVT_LINK_CHANGE,
185 		    (softc->ha_connect || softc->ha_listen) ?
186 		    CTL_HA_LINK_UNKNOWN : CTL_HA_LINK_OFFLINE);
187 	}
188 }
189 
190 static void
191 ctl_ha_lclose(struct ha_softc *softc)
192 {
193 
194 	if (softc->ha_lso) {
195 		if (SOLISTENING(softc->ha_lso)) {
196 			SOLISTEN_LOCK(softc->ha_lso);
197 			solisten_upcall_set(softc->ha_lso, NULL, NULL);
198 			SOLISTEN_UNLOCK(softc->ha_lso);
199 		}
200 		soclose(softc->ha_lso);
201 		softc->ha_lso = NULL;
202 	}
203 }
204 
205 static void
206 ctl_ha_rx_thread(void *arg)
207 {
208 	struct ha_softc *softc = arg;
209 	struct socket *so = softc->ha_so;
210 	struct ha_msg_wire wire_hdr;
211 	struct uio uio;
212 	struct iovec iov;
213 	int error, flags, next;
214 
215 	bzero(&wire_hdr, sizeof(wire_hdr));
216 	while (1) {
217 		if (wire_hdr.length > 0)
218 			next = wire_hdr.length;
219 		else
220 			next = sizeof(wire_hdr);
221 		SOCKBUF_LOCK(&so->so_rcv);
222 		while (sbavail(&so->so_rcv) < next || softc->ha_disconnect) {
223 			if (softc->ha_connected == 0 || softc->ha_disconnect ||
224 			    so->so_error ||
225 			    (so->so_rcv.sb_state & SBS_CANTRCVMORE)) {
226 				goto errout;
227 			}
228 			so->so_rcv.sb_lowat = next;
229 			msleep(&softc->ha_receiving, SOCKBUF_MTX(&so->so_rcv),
230 			    0, "-", 0);
231 		}
232 		SOCKBUF_UNLOCK(&so->so_rcv);
233 
234 		if (wire_hdr.length == 0) {
235 			iov.iov_base = &wire_hdr;
236 			iov.iov_len = sizeof(wire_hdr);
237 			uio.uio_iov = &iov;
238 			uio.uio_iovcnt = 1;
239 			uio.uio_rw = UIO_READ;
240 			uio.uio_segflg = UIO_SYSSPACE;
241 			uio.uio_td = curthread;
242 			uio.uio_resid = sizeof(wire_hdr);
243 			flags = MSG_DONTWAIT;
244 			error = soreceive(softc->ha_so, NULL, &uio, NULL,
245 			    NULL, &flags);
246 			if (error != 0) {
247 				printf("%s: header receive error %d\n",
248 				    __func__, error);
249 				SOCKBUF_LOCK(&so->so_rcv);
250 				goto errout;
251 			}
252 		} else {
253 			ctl_ha_evt(softc, wire_hdr.channel,
254 			    CTL_HA_EVT_MSG_RECV, wire_hdr.length);
255 			wire_hdr.length = 0;
256 		}
257 	}
258 
259 errout:
260 	softc->ha_receiving = 0;
261 	wakeup(&softc->ha_receiving);
262 	SOCKBUF_UNLOCK(&so->so_rcv);
263 	ctl_ha_conn_wake(softc);
264 	kthread_exit();
265 }
266 
267 static void
268 ctl_ha_send(struct ha_softc *softc)
269 {
270 	struct socket *so = softc->ha_so;
271 	int error;
272 
273 	while (1) {
274 		if (softc->ha_sending == NULL) {
275 			mtx_lock(&softc->ha_lock);
276 			softc->ha_sending = mbufq_dequeue(&softc->ha_sendq);
277 			mtx_unlock(&softc->ha_lock);
278 			if (softc->ha_sending == NULL) {
279 				so->so_snd.sb_lowat = so->so_snd.sb_hiwat + 1;
280 				break;
281 			}
282 		}
283 		SOCKBUF_LOCK(&so->so_snd);
284 		if (sbspace(&so->so_snd) < softc->ha_sending->m_pkthdr.len) {
285 			so->so_snd.sb_lowat = softc->ha_sending->m_pkthdr.len;
286 			SOCKBUF_UNLOCK(&so->so_snd);
287 			break;
288 		}
289 		SOCKBUF_UNLOCK(&so->so_snd);
290 		error = sosend(softc->ha_so, NULL, NULL, softc->ha_sending,
291 		    NULL, MSG_DONTWAIT, curthread);
292 		softc->ha_sending = NULL;
293 		if (error != 0) {
294 			printf("%s: sosend() error %d\n", __func__, error);
295 			return;
296 		}
297 	}
298 }
299 
300 static void
301 ctl_ha_sock_setup(struct ha_softc *softc)
302 {
303 	struct sockopt opt;
304 	struct socket *so = softc->ha_so;
305 	int error, val;
306 
307 	val = 1024 * 1024;
308 	error = soreserve(so, val, val);
309 	if (error)
310 		printf("%s: soreserve failed %d\n", __func__, error);
311 
312 	SOCKBUF_LOCK(&so->so_rcv);
313 	so->so_rcv.sb_lowat = sizeof(struct ha_msg_wire);
314 	soupcall_set(so, SO_RCV, ctl_ha_rupcall, softc);
315 	SOCKBUF_UNLOCK(&so->so_rcv);
316 	SOCKBUF_LOCK(&so->so_snd);
317 	so->so_snd.sb_lowat = sizeof(struct ha_msg_wire);
318 	soupcall_set(so, SO_SND, ctl_ha_supcall, softc);
319 	SOCKBUF_UNLOCK(&so->so_snd);
320 
321 	bzero(&opt, sizeof(struct sockopt));
322 	opt.sopt_dir = SOPT_SET;
323 	opt.sopt_level = SOL_SOCKET;
324 	opt.sopt_name = SO_KEEPALIVE;
325 	opt.sopt_val = &val;
326 	opt.sopt_valsize = sizeof(val);
327 	val = 1;
328 	error = sosetopt(so, &opt);
329 	if (error)
330 		printf("%s: KEEPALIVE setting failed %d\n", __func__, error);
331 
332 	opt.sopt_level = IPPROTO_TCP;
333 	opt.sopt_name = TCP_NODELAY;
334 	val = 1;
335 	error = sosetopt(so, &opt);
336 	if (error)
337 		printf("%s: NODELAY setting failed %d\n", __func__, error);
338 
339 	opt.sopt_name = TCP_KEEPINIT;
340 	val = 3;
341 	error = sosetopt(so, &opt);
342 	if (error)
343 		printf("%s: KEEPINIT setting failed %d\n", __func__, error);
344 
345 	opt.sopt_name = TCP_KEEPIDLE;
346 	val = 1;
347 	error = sosetopt(so, &opt);
348 	if (error)
349 		printf("%s: KEEPIDLE setting failed %d\n", __func__, error);
350 
351 	opt.sopt_name = TCP_KEEPINTVL;
352 	val = 1;
353 	error = sosetopt(so, &opt);
354 	if (error)
355 		printf("%s: KEEPINTVL setting failed %d\n", __func__, error);
356 
357 	opt.sopt_name = TCP_KEEPCNT;
358 	val = 5;
359 	error = sosetopt(so, &opt);
360 	if (error)
361 		printf("%s: KEEPCNT setting failed %d\n", __func__, error);
362 }
363 
364 static int
365 ctl_ha_connect(struct ha_softc *softc)
366 {
367 	struct thread *td = curthread;
368 	struct sockaddr_in sa;
369 	struct socket *so;
370 	int error;
371 
372 	/* Create the socket */
373 	error = socreate(PF_INET, &so, SOCK_STREAM,
374 	    IPPROTO_TCP, td->td_ucred, td);
375 	if (error != 0) {
376 		printf("%s: socreate() error %d\n", __func__, error);
377 		return (error);
378 	}
379 	softc->ha_so = so;
380 	ctl_ha_sock_setup(softc);
381 
382 	memcpy(&sa, &softc->ha_peer_in, sizeof(sa));
383 	error = soconnect(so, (struct sockaddr *)&sa, td);
384 	if (error != 0) {
385 		if (bootverbose)
386 			printf("%s: soconnect() error %d\n", __func__, error);
387 		goto out;
388 	}
389 	return (0);
390 
391 out:
392 	ctl_ha_close(softc);
393 	return (error);
394 }
395 
396 static int
397 ctl_ha_accept(struct ha_softc *softc)
398 {
399 	struct socket *lso, *so;
400 	struct sockaddr_in sin = { .sin_len = sizeof(sin) };
401 	int error;
402 
403 	lso = softc->ha_lso;
404 	SOLISTEN_LOCK(lso);
405 	error = solisten_dequeue(lso, &so, 0);
406 	if (error == EWOULDBLOCK)
407 		return (error);
408 	if (error) {
409 		printf("%s: socket error %d\n", __func__, error);
410 		goto out;
411 	}
412 
413 	error = soaccept(so, (struct sockaddr *)&sin);
414 	if (error != 0) {
415 		printf("%s: soaccept() error %d\n", __func__, error);
416 		goto out;
417 	}
418 	softc->ha_so = so;
419 	ctl_ha_sock_setup(softc);
420 	return (0);
421 
422 out:
423 	ctl_ha_lclose(softc);
424 	return (error);
425 }
426 
427 static int
428 ctl_ha_listen(struct ha_softc *softc)
429 {
430 	struct thread *td = curthread;
431 	struct sockaddr_in sa;
432 	struct sockopt opt;
433 	int error, val;
434 
435 	/* Create the socket */
436 	if (softc->ha_lso == NULL) {
437 		error = socreate(PF_INET, &softc->ha_lso, SOCK_STREAM,
438 		    IPPROTO_TCP, td->td_ucred, td);
439 		if (error != 0) {
440 			printf("%s: socreate() error %d\n", __func__, error);
441 			return (error);
442 		}
443 		bzero(&opt, sizeof(struct sockopt));
444 		opt.sopt_dir = SOPT_SET;
445 		opt.sopt_level = SOL_SOCKET;
446 		opt.sopt_name = SO_REUSEADDR;
447 		opt.sopt_val = &val;
448 		opt.sopt_valsize = sizeof(val);
449 		val = 1;
450 		error = sosetopt(softc->ha_lso, &opt);
451 		if (error) {
452 			printf("%s: REUSEADDR setting failed %d\n",
453 			    __func__, error);
454 		}
455 		bzero(&opt, sizeof(struct sockopt));
456 		opt.sopt_dir = SOPT_SET;
457 		opt.sopt_level = SOL_SOCKET;
458 		opt.sopt_name = SO_REUSEPORT;
459 		opt.sopt_val = &val;
460 		opt.sopt_valsize = sizeof(val);
461 		val = 1;
462 		error = sosetopt(softc->ha_lso, &opt);
463 		if (error) {
464 			printf("%s: REUSEPORT setting failed %d\n",
465 			    __func__, error);
466 		}
467 	}
468 
469 	memcpy(&sa, &softc->ha_peer_in, sizeof(sa));
470 	error = sobind(softc->ha_lso, (struct sockaddr *)&sa, td);
471 	if (error != 0) {
472 		printf("%s: sobind() error %d\n", __func__, error);
473 		goto out;
474 	}
475 	error = solisten(softc->ha_lso, 1, td);
476 	if (error != 0) {
477 		printf("%s: solisten() error %d\n", __func__, error);
478 		goto out;
479 	}
480 	SOLISTEN_LOCK(softc->ha_lso);
481 	softc->ha_lso->so_state |= SS_NBIO;
482 	solisten_upcall_set(softc->ha_lso, ctl_ha_lupcall, softc);
483 	SOLISTEN_UNLOCK(softc->ha_lso);
484 	return (0);
485 
486 out:
487 	ctl_ha_lclose(softc);
488 	return (error);
489 }
490 
491 static void
492 ctl_ha_conn_thread(void *arg)
493 {
494 	struct ha_softc *softc = arg;
495 	int error;
496 
497 	while (1) {
498 		if (softc->ha_disconnect || softc->ha_shutdown) {
499 			ctl_ha_close(softc);
500 			if (softc->ha_disconnect == 2 || softc->ha_shutdown)
501 				ctl_ha_lclose(softc);
502 			softc->ha_disconnect = 0;
503 			if (softc->ha_shutdown)
504 				break;
505 		} else if (softc->ha_so != NULL &&
506 		    (softc->ha_so->so_error ||
507 		     softc->ha_so->so_rcv.sb_state & SBS_CANTRCVMORE))
508 			ctl_ha_close(softc);
509 		if (softc->ha_so == NULL) {
510 			if (softc->ha_lso != NULL)
511 				ctl_ha_accept(softc);
512 			else if (softc->ha_listen)
513 				ctl_ha_listen(softc);
514 			else if (softc->ha_connect)
515 				ctl_ha_connect(softc);
516 		}
517 		if (softc->ha_so != NULL) {
518 			if (softc->ha_connected == 0 &&
519 			    softc->ha_so->so_error == 0 &&
520 			    (softc->ha_so->so_state & SS_ISCONNECTING) == 0) {
521 				softc->ha_connected = 1;
522 				ctl_ha_evt(softc, CTL_HA_CHAN_MAX,
523 				    CTL_HA_EVT_LINK_CHANGE,
524 				    CTL_HA_LINK_ONLINE);
525 				softc->ha_receiving = 1;
526 				error = kproc_kthread_add(ctl_ha_rx_thread,
527 				    softc, &softc->ha_ctl_softc->ctl_proc,
528 				    NULL, 0, 0, "ctl", "ha_rx");
529 				if (error != 0) {
530 					printf("Error creating CTL HA rx thread!\n");
531 					softc->ha_receiving = 0;
532 					softc->ha_disconnect = 1;
533 				}
534 			}
535 			ctl_ha_send(softc);
536 		}
537 		mtx_lock(&softc->ha_lock);
538 		if (softc->ha_so != NULL &&
539 		    (softc->ha_so->so_error ||
540 		     softc->ha_so->so_rcv.sb_state & SBS_CANTRCVMORE))
541 			;
542 		else if (!softc->ha_wakeup)
543 			msleep(&softc->ha_wakeup, &softc->ha_lock, 0, "-", hz);
544 		softc->ha_wakeup = 0;
545 		mtx_unlock(&softc->ha_lock);
546 	}
547 	mtx_lock(&softc->ha_lock);
548 	softc->ha_shutdown = 2;
549 	wakeup(&softc->ha_wakeup);
550 	mtx_unlock(&softc->ha_lock);
551 	kthread_exit();
552 }
553 
554 static int
555 ctl_ha_peer_sysctl(SYSCTL_HANDLER_ARGS)
556 {
557 	struct ha_softc *softc = (struct ha_softc *)arg1;
558 	struct sockaddr_in *sa;
559 	int error, b1, b2, b3, b4, p, num;
560 	char buf[128];
561 
562 	strlcpy(buf, softc->ha_peer, sizeof(buf));
563 	error = sysctl_handle_string(oidp, buf, sizeof(buf), req);
564 	if ((error != 0) || (req->newptr == NULL) ||
565 	    strncmp(buf, softc->ha_peer, sizeof(buf)) == 0)
566 		return (error);
567 
568 	sa = &softc->ha_peer_in;
569 	mtx_lock(&softc->ha_lock);
570 	if ((num = sscanf(buf, "connect %d.%d.%d.%d:%d",
571 	    &b1, &b2, &b3, &b4, &p)) >= 4) {
572 		softc->ha_connect = 1;
573 		softc->ha_listen = 0;
574 	} else if ((num = sscanf(buf, "listen %d.%d.%d.%d:%d",
575 	    &b1, &b2, &b3, &b4, &p)) >= 4) {
576 		softc->ha_connect = 0;
577 		softc->ha_listen = 1;
578 	} else {
579 		softc->ha_connect = 0;
580 		softc->ha_listen = 0;
581 		if (buf[0] != 0) {
582 			buf[0] = 0;
583 			error = EINVAL;
584 		}
585 	}
586 	strlcpy(softc->ha_peer, buf, sizeof(softc->ha_peer));
587 	if (softc->ha_connect || softc->ha_listen) {
588 		memset(sa, 0, sizeof(*sa));
589 		sa->sin_len = sizeof(struct sockaddr_in);
590 		sa->sin_family = AF_INET;
591 		sa->sin_port = htons((num >= 5) ? p : 999);
592 		sa->sin_addr.s_addr =
593 		    htonl((b1 << 24) + (b2 << 16) + (b3 << 8) + b4);
594 	}
595 	softc->ha_disconnect = 2;
596 	softc->ha_wakeup = 1;
597 	mtx_unlock(&softc->ha_lock);
598 	wakeup(&softc->ha_wakeup);
599 	return (error);
600 }
601 
602 ctl_ha_status
603 ctl_ha_msg_register(ctl_ha_channel channel, ctl_evt_handler handler)
604 {
605 	struct ha_softc *softc = &ha_softc;
606 
607 	KASSERT(channel < CTL_HA_CHAN_MAX,
608 	    ("Wrong CTL HA channel %d", channel));
609 	softc->ha_handler[channel] = handler;
610 	return (CTL_HA_STATUS_SUCCESS);
611 }
612 
613 ctl_ha_status
614 ctl_ha_msg_deregister(ctl_ha_channel channel)
615 {
616 	struct ha_softc *softc = &ha_softc;
617 
618 	KASSERT(channel < CTL_HA_CHAN_MAX,
619 	    ("Wrong CTL HA channel %d", channel));
620 	softc->ha_handler[channel] = NULL;
621 	return (CTL_HA_STATUS_SUCCESS);
622 }
623 
624 /*
625  * Receive a message of the specified size.
626  */
627 ctl_ha_status
628 ctl_ha_msg_recv(ctl_ha_channel channel, void *addr, size_t len,
629 		int wait)
630 {
631 	struct ha_softc *softc = &ha_softc;
632 	struct uio uio;
633 	struct iovec iov;
634 	int error, flags;
635 
636 	if (!softc->ha_connected)
637 		return (CTL_HA_STATUS_DISCONNECT);
638 
639 	iov.iov_base = addr;
640 	iov.iov_len = len;
641 	uio.uio_iov = &iov;
642 	uio.uio_iovcnt = 1;
643 	uio.uio_rw = UIO_READ;
644 	uio.uio_segflg = UIO_SYSSPACE;
645 	uio.uio_td = curthread;
646 	uio.uio_resid = len;
647 	flags = wait ? 0 : MSG_DONTWAIT;
648 	error = soreceive(softc->ha_so, NULL, &uio, NULL, NULL, &flags);
649 	if (error == 0)
650 		return (CTL_HA_STATUS_SUCCESS);
651 
652 	/* Consider all errors fatal for HA sanity. */
653 	mtx_lock(&softc->ha_lock);
654 	if (softc->ha_connected) {
655 		softc->ha_disconnect = 1;
656 		softc->ha_wakeup = 1;
657 		wakeup(&softc->ha_wakeup);
658 	}
659 	mtx_unlock(&softc->ha_lock);
660 	return (CTL_HA_STATUS_ERROR);
661 }
662 
663 /*
664  * Send a message of the specified size.
665  */
666 ctl_ha_status
667 ctl_ha_msg_send2(ctl_ha_channel channel, const void *addr, size_t len,
668     const void *addr2, size_t len2, int wait)
669 {
670 	struct ha_softc *softc = &ha_softc;
671 	struct mbuf *mb, *newmb;
672 	struct ha_msg_wire hdr;
673 	size_t copylen, off;
674 
675 	if (!softc->ha_connected)
676 		return (CTL_HA_STATUS_DISCONNECT);
677 
678 	newmb = m_getm2(NULL, sizeof(hdr) + len + len2, wait, MT_DATA,
679 	    M_PKTHDR);
680 	if (newmb == NULL) {
681 		/* Consider all errors fatal for HA sanity. */
682 		mtx_lock(&softc->ha_lock);
683 		if (softc->ha_connected) {
684 			softc->ha_disconnect = 1;
685 			softc->ha_wakeup = 1;
686 			wakeup(&softc->ha_wakeup);
687 		}
688 		mtx_unlock(&softc->ha_lock);
689 		printf("%s: Can't allocate mbuf chain\n", __func__);
690 		return (CTL_HA_STATUS_ERROR);
691 	}
692 	hdr.channel = channel;
693 	hdr.length = len + len2;
694 	mb = newmb;
695 	memcpy(mtodo(mb, 0), &hdr, sizeof(hdr));
696 	mb->m_len += sizeof(hdr);
697 	off = 0;
698 	for (; mb != NULL && off < len; mb = mb->m_next) {
699 		copylen = min(M_TRAILINGSPACE(mb), len - off);
700 		memcpy(mtodo(mb, mb->m_len), (const char *)addr + off, copylen);
701 		mb->m_len += copylen;
702 		off += copylen;
703 		if (off == len)
704 			break;
705 	}
706 	KASSERT(off == len, ("%s: off (%zu) != len (%zu)", __func__,
707 	    off, len));
708 	off = 0;
709 	for (; mb != NULL && off < len2; mb = mb->m_next) {
710 		copylen = min(M_TRAILINGSPACE(mb), len2 - off);
711 		memcpy(mtodo(mb, mb->m_len), (const char *)addr2 + off, copylen);
712 		mb->m_len += copylen;
713 		off += copylen;
714 	}
715 	KASSERT(off == len2, ("%s: off (%zu) != len2 (%zu)", __func__,
716 	    off, len2));
717 	newmb->m_pkthdr.len = sizeof(hdr) + len + len2;
718 
719 	mtx_lock(&softc->ha_lock);
720 	if (!softc->ha_connected) {
721 		mtx_unlock(&softc->ha_lock);
722 		m_freem(newmb);
723 		return (CTL_HA_STATUS_DISCONNECT);
724 	}
725 	mbufq_enqueue(&softc->ha_sendq, newmb);
726 	softc->ha_wakeup = 1;
727 	mtx_unlock(&softc->ha_lock);
728 	wakeup(&softc->ha_wakeup);
729 	return (CTL_HA_STATUS_SUCCESS);
730 }
731 
732 ctl_ha_status
733 ctl_ha_msg_send(ctl_ha_channel channel, const void *addr, size_t len,
734     int wait)
735 {
736 
737 	return (ctl_ha_msg_send2(channel, addr, len, NULL, 0, wait));
738 }
739 
740 ctl_ha_status
741 ctl_ha_msg_abort(ctl_ha_channel channel)
742 {
743 	struct ha_softc *softc = &ha_softc;
744 
745 	mtx_lock(&softc->ha_lock);
746 	softc->ha_disconnect = 1;
747 	softc->ha_wakeup = 1;
748 	mtx_unlock(&softc->ha_lock);
749 	wakeup(&softc->ha_wakeup);
750 	return (CTL_HA_STATUS_SUCCESS);
751 }
752 
753 /*
754  * Allocate a data transfer request structure.
755  */
756 struct ctl_ha_dt_req *
757 ctl_dt_req_alloc(void)
758 {
759 
760 	return (malloc(sizeof(struct ctl_ha_dt_req), M_CTL, M_WAITOK | M_ZERO));
761 }
762 
763 /*
764  * Free a data transfer request structure.
765  */
766 void
767 ctl_dt_req_free(struct ctl_ha_dt_req *req)
768 {
769 
770 	free(req, M_CTL);
771 }
772 
773 /*
774  * Issue a DMA request for a single buffer.
775  */
776 ctl_ha_status
777 ctl_dt_single(struct ctl_ha_dt_req *req)
778 {
779 	struct ha_softc *softc = &ha_softc;
780 	struct ha_dt_msg_wire wire_dt;
781 	ctl_ha_status status;
782 
783 	wire_dt.command = req->command;
784 	wire_dt.size = req->size;
785 	wire_dt.local = req->local;
786 	wire_dt.remote = req->remote;
787 	if (req->command == CTL_HA_DT_CMD_READ && req->callback != NULL) {
788 		mtx_lock(&softc->ha_lock);
789 		TAILQ_INSERT_TAIL(&softc->ha_dts, req, links);
790 		mtx_unlock(&softc->ha_lock);
791 		ctl_ha_msg_send(CTL_HA_CHAN_DATA, &wire_dt, sizeof(wire_dt),
792 		    M_WAITOK);
793 		return (CTL_HA_STATUS_WAIT);
794 	}
795 	if (req->command == CTL_HA_DT_CMD_READ) {
796 		status = ctl_ha_msg_send(CTL_HA_CHAN_DATA, &wire_dt,
797 		    sizeof(wire_dt), M_WAITOK);
798 	} else {
799 		status = ctl_ha_msg_send2(CTL_HA_CHAN_DATA, &wire_dt,
800 		    sizeof(wire_dt), req->local, req->size, M_WAITOK);
801 	}
802 	return (status);
803 }
804 
805 static void
806 ctl_dt_event_handler(ctl_ha_channel channel, ctl_ha_event event, int param)
807 {
808 	struct ha_softc *softc = &ha_softc;
809 	struct ctl_ha_dt_req *req;
810 	ctl_ha_status isc_status;
811 
812 	if (event == CTL_HA_EVT_MSG_RECV) {
813 		struct ha_dt_msg_wire wire_dt;
814 		uint8_t *tmp;
815 		int size;
816 
817 		size = min(sizeof(wire_dt), param);
818 		isc_status = ctl_ha_msg_recv(CTL_HA_CHAN_DATA, &wire_dt,
819 					     size, M_WAITOK);
820 		if (isc_status != CTL_HA_STATUS_SUCCESS) {
821 			printf("%s: Error receiving message: %d\n",
822 			    __func__, isc_status);
823 			return;
824 		}
825 
826 		if (wire_dt.command == CTL_HA_DT_CMD_READ) {
827 			wire_dt.command = CTL_HA_DT_CMD_WRITE;
828 			tmp = wire_dt.local;
829 			wire_dt.local = wire_dt.remote;
830 			wire_dt.remote = tmp;
831 			ctl_ha_msg_send2(CTL_HA_CHAN_DATA, &wire_dt,
832 			    sizeof(wire_dt), wire_dt.local, wire_dt.size,
833 			    M_WAITOK);
834 		} else if (wire_dt.command == CTL_HA_DT_CMD_WRITE) {
835 			isc_status = ctl_ha_msg_recv(CTL_HA_CHAN_DATA,
836 			    wire_dt.remote, wire_dt.size, M_WAITOK);
837 			mtx_lock(&softc->ha_lock);
838 			TAILQ_FOREACH(req, &softc->ha_dts, links) {
839 				if (req->local == wire_dt.remote) {
840 					TAILQ_REMOVE(&softc->ha_dts, req, links);
841 					break;
842 				}
843 			}
844 			mtx_unlock(&softc->ha_lock);
845 			if (req) {
846 				req->ret = isc_status;
847 				req->callback(req);
848 			}
849 		}
850 	} else if (event == CTL_HA_EVT_LINK_CHANGE) {
851 		CTL_DEBUG_PRINT(("%s: Link state change to %d\n", __func__,
852 		    param));
853 		if (param != CTL_HA_LINK_ONLINE) {
854 			mtx_lock(&softc->ha_lock);
855 			while ((req = TAILQ_FIRST(&softc->ha_dts)) != NULL) {
856 				TAILQ_REMOVE(&softc->ha_dts, req, links);
857 				mtx_unlock(&softc->ha_lock);
858 				req->ret = CTL_HA_STATUS_DISCONNECT;
859 				req->callback(req);
860 				mtx_lock(&softc->ha_lock);
861 			}
862 			mtx_unlock(&softc->ha_lock);
863 		}
864 	} else {
865 		printf("%s: Unknown event %d\n", __func__, event);
866 	}
867 }
868 
869 ctl_ha_status
870 ctl_ha_msg_init(struct ctl_softc *ctl_softc)
871 {
872 	struct ha_softc *softc = &ha_softc;
873 	int error;
874 
875 	softc->ha_ctl_softc = ctl_softc;
876 	mtx_init(&softc->ha_lock, "CTL HA mutex", NULL, MTX_DEF);
877 	mbufq_init(&softc->ha_sendq, INT_MAX);
878 	TAILQ_INIT(&softc->ha_dts);
879 	error = kproc_kthread_add(ctl_ha_conn_thread, softc,
880 	    &ctl_softc->ctl_proc, NULL, 0, 0, "ctl", "ha_tx");
881 	if (error != 0) {
882 		printf("error creating CTL HA connection thread!\n");
883 		mtx_destroy(&softc->ha_lock);
884 		return (CTL_HA_STATUS_ERROR);
885 	}
886 	softc->ha_shutdown_eh = EVENTHANDLER_REGISTER(shutdown_pre_sync,
887 	    ctl_ha_msg_shutdown, ctl_softc, SHUTDOWN_PRI_FIRST);
888 	SYSCTL_ADD_PROC(&ctl_softc->sysctl_ctx,
889 	    SYSCTL_CHILDREN(ctl_softc->sysctl_tree),
890 	    OID_AUTO, "ha_peer",
891 	    CTLTYPE_STRING | CTLFLAG_RWTUN | CTLFLAG_MPSAFE,
892 	    softc, 0, ctl_ha_peer_sysctl, "A", "HA peer connection method");
893 
894 	if (ctl_ha_msg_register(CTL_HA_CHAN_DATA, ctl_dt_event_handler)
895 	    != CTL_HA_STATUS_SUCCESS) {
896 		printf("%s: ctl_ha_msg_register failed.\n", __func__);
897 	}
898 
899 	return (CTL_HA_STATUS_SUCCESS);
900 };
901 
902 void
903 ctl_ha_msg_shutdown(struct ctl_softc *ctl_softc)
904 {
905 	struct ha_softc *softc = &ha_softc;
906 
907 	if (SCHEDULER_STOPPED())
908 		return;
909 
910 	/* Disconnect and shutdown threads. */
911 	mtx_lock(&softc->ha_lock);
912 	if (softc->ha_shutdown < 2) {
913 		softc->ha_shutdown = 1;
914 		softc->ha_wakeup = 1;
915 		wakeup(&softc->ha_wakeup);
916 		while (softc->ha_shutdown < 2) {
917 			msleep(&softc->ha_wakeup, &softc->ha_lock, 0,
918 			    "shutdown", hz);
919 		}
920 	}
921 	mtx_unlock(&softc->ha_lock);
922 };
923 
924 ctl_ha_status
925 ctl_ha_msg_destroy(struct ctl_softc *ctl_softc)
926 {
927 	struct ha_softc *softc = &ha_softc;
928 
929 	if (softc->ha_shutdown_eh != NULL) {
930 		EVENTHANDLER_DEREGISTER(shutdown_pre_sync,
931 		    softc->ha_shutdown_eh);
932 		softc->ha_shutdown_eh = NULL;
933 	}
934 
935 	ctl_ha_msg_shutdown(ctl_softc);	/* Just in case. */
936 
937 	if (ctl_ha_msg_deregister(CTL_HA_CHAN_DATA) != CTL_HA_STATUS_SUCCESS)
938 		printf("%s: ctl_ha_msg_deregister failed.\n", __func__);
939 
940 	mtx_destroy(&softc->ha_lock);
941 	return (CTL_HA_STATUS_SUCCESS);
942 };
943