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