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