ctl_ha.c revision 288795
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: stable/10/sys/cam/ctl/ctl_ha.c 288795 2015-10-05 10:58:41Z mav $");
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)
68struct mbufq {
69	struct mbuf *head;
70	struct mbuf *tail;
71};
72
73static void
74mbufq_init(struct mbufq *q, int limit)
75{
76
77	q->head = q->tail = NULL;
78}
79
80static void
81mbufq_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
92static struct mbuf *
93mbufq_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
107static void
108mbufq_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
119static u_int
120sbavail(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
130struct ha_msg_wire {
131	uint32_t	 channel;
132	uint32_t	 length;
133};
134
135struct ha_dt_msg_wire {
136	ctl_ha_dt_cmd	command;
137	uint32_t	size;
138	uint8_t		*local;
139	uint8_t		*remote;
140};
141
142struct 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
163static void
164ctl_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
173static int
174ctl_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
182static int
183ctl_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
191static int
192ctl_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
200static void
201ctl_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
217static void
218ctl_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
254static void
255ctl_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
267static void
268ctl_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
321errout:
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
329static void
330ctl_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
362static void
363ctl_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
426static int
427ctl_ha_connect(struct ha_softc *softc)
428{
429	struct thread *td = curthread;
430	struct socket *so;
431	int error;
432
433	/* Create the socket */
434	error = socreate(PF_INET, &so, SOCK_STREAM,
435	    IPPROTO_TCP, td->td_ucred, td);
436	if (error != 0) {
437		printf("%s: socreate() error %d\n", __func__, error);
438		return (error);
439	}
440	softc->ha_so = so;
441	ctl_ha_sock_setup(softc);
442
443	error = soconnect(so, (struct sockaddr *)&softc->ha_peer_in, td);
444	if (error != 0) {
445		printf("%s: soconnect() error %d\n", __func__, error);
446		goto out;
447	}
448	return (0);
449
450out:
451	ctl_ha_close(softc);
452	return (error);
453}
454
455static int
456ctl_ha_accept(struct ha_softc *softc)
457{
458	struct socket *so;
459	struct sockaddr *sap;
460	int error;
461
462	ACCEPT_LOCK();
463	if (softc->ha_lso->so_rcv.sb_state & SBS_CANTRCVMORE)
464		softc->ha_lso->so_error = ECONNABORTED;
465	if (softc->ha_lso->so_error) {
466		error = softc->ha_lso->so_error;
467		softc->ha_lso->so_error = 0;
468		ACCEPT_UNLOCK();
469		printf("%s: socket error %d\n", __func__, error);
470		goto out;
471	}
472	so = TAILQ_FIRST(&softc->ha_lso->so_comp);
473	if (so == NULL) {
474		ACCEPT_UNLOCK();
475		return (EWOULDBLOCK);
476	}
477	KASSERT(!(so->so_qstate & SQ_INCOMP), ("accept1: so SQ_INCOMP"));
478	KASSERT(so->so_qstate & SQ_COMP, ("accept1: so not SQ_COMP"));
479
480	/*
481	 * Before changing the flags on the socket, we have to bump the
482	 * reference count.  Otherwise, if the protocol calls sofree(),
483	 * the socket will be released due to a zero refcount.
484	 */
485	SOCK_LOCK(so);			/* soref() and so_state update */
486	soref(so);			/* file descriptor reference */
487
488	TAILQ_REMOVE(&softc->ha_lso->so_comp, so, so_list);
489	softc->ha_lso->so_qlen--;
490	so->so_state |= SS_NBIO;
491	so->so_qstate &= ~SQ_COMP;
492	so->so_head = NULL;
493
494	SOCK_UNLOCK(so);
495	ACCEPT_UNLOCK();
496
497	sap = NULL;
498	error = soaccept(so, &sap);
499	if (error != 0) {
500		printf("%s: soaccept() error %d\n", __func__, error);
501		if (sap != NULL)
502			free(sap, M_SONAME);
503		goto out;
504	}
505	if (sap != NULL)
506		free(sap, M_SONAME);
507	softc->ha_so = so;
508	ctl_ha_sock_setup(softc);
509	return (0);
510
511out:
512	ctl_ha_lclose(softc);
513	return (error);
514}
515
516static int
517ctl_ha_listen(struct ha_softc *softc)
518{
519	struct thread *td = curthread;
520	struct sockopt opt;
521	int error, val;
522
523	/* Create the socket */
524	if (softc->ha_lso == NULL) {
525		error = socreate(PF_INET, &softc->ha_lso, SOCK_STREAM,
526		    IPPROTO_TCP, td->td_ucred, td);
527		if (error != 0) {
528			printf("%s: socreate() error %d\n", __func__, error);
529			return (error);
530		}
531		bzero(&opt, sizeof(struct sockopt));
532		opt.sopt_dir = SOPT_SET;
533		opt.sopt_level = SOL_SOCKET;
534		opt.sopt_name = SO_REUSEADDR;
535		opt.sopt_val = &val;
536		opt.sopt_valsize = sizeof(val);
537		val = 1;
538		error = sosetopt(softc->ha_lso, &opt);
539		if (error) {
540			printf("%s: REUSEADDR setting failed %d\n",
541			    __func__, error);
542		}
543		bzero(&opt, sizeof(struct sockopt));
544		opt.sopt_dir = SOPT_SET;
545		opt.sopt_level = SOL_SOCKET;
546		opt.sopt_name = SO_REUSEPORT;
547		opt.sopt_val = &val;
548		opt.sopt_valsize = sizeof(val);
549		val = 1;
550		error = sosetopt(softc->ha_lso, &opt);
551		if (error) {
552			printf("%s: REUSEPORT setting failed %d\n",
553			    __func__, error);
554		}
555		SOCKBUF_LOCK(&softc->ha_lso->so_rcv);
556		soupcall_set(softc->ha_lso, SO_RCV, ctl_ha_lupcall, softc);
557		SOCKBUF_UNLOCK(&softc->ha_lso->so_rcv);
558	}
559
560	error = sobind(softc->ha_lso, (struct sockaddr *)&softc->ha_peer_in, td);
561	if (error != 0) {
562		printf("%s: sobind() error %d\n", __func__, error);
563		goto out;
564	}
565	error = solisten(softc->ha_lso, 1, td);
566	if (error != 0) {
567		printf("%s: solisten() error %d\n", __func__, error);
568		goto out;
569	}
570	return (0);
571
572out:
573	ctl_ha_lclose(softc);
574	return (error);
575}
576
577static void
578ctl_ha_conn_thread(void *arg)
579{
580	struct ha_softc *softc = arg;
581	int error;
582
583	while (1) {
584		if (softc->ha_disconnect || softc->ha_shutdown) {
585			ctl_ha_close(softc);
586			if (softc->ha_disconnect == 2 || softc->ha_shutdown)
587				ctl_ha_lclose(softc);
588			softc->ha_disconnect = 0;
589			if (softc->ha_shutdown)
590				break;
591		} else if (softc->ha_so != NULL &&
592		    (softc->ha_so->so_error ||
593		     softc->ha_so->so_rcv.sb_state & SBS_CANTRCVMORE))
594			ctl_ha_close(softc);
595		if (softc->ha_so == NULL) {
596			if (softc->ha_lso != NULL)
597				ctl_ha_accept(softc);
598			else if (softc->ha_listen)
599				ctl_ha_listen(softc);
600			else if (softc->ha_connect)
601				ctl_ha_connect(softc);
602		}
603		if (softc->ha_so != NULL) {
604			if (softc->ha_connected == 0 &&
605			    softc->ha_so->so_error == 0 &&
606			    (softc->ha_so->so_state & SS_ISCONNECTING) == 0) {
607				softc->ha_connected = 1;
608				ctl_ha_evt(softc, CTL_HA_CHAN_MAX,
609				    CTL_HA_EVT_LINK_CHANGE,
610				    CTL_HA_LINK_ONLINE);
611				softc->ha_receiving = 1;
612				error = kproc_kthread_add(ctl_ha_rx_thread,
613				    softc, &softc->ha_ctl_softc->ctl_proc,
614				    NULL, 0, 0, "ctl", "ha_rx");
615				if (error != 0) {
616					printf("Error creating CTL HA rx thread!\n");
617					softc->ha_receiving = 0;
618					softc->ha_disconnect = 1;
619				}
620			}
621			ctl_ha_send(softc);
622		}
623		mtx_lock(&softc->ha_lock);
624		if (softc->ha_so != NULL &&
625		    (softc->ha_so->so_error ||
626		     softc->ha_so->so_rcv.sb_state & SBS_CANTRCVMORE))
627			;
628		else if (!softc->ha_wakeup)
629			msleep(&softc->ha_wakeup, &softc->ha_lock, 0, "-", hz);
630		softc->ha_wakeup = 0;
631		mtx_unlock(&softc->ha_lock);
632	}
633	mtx_lock(&softc->ha_lock);
634	softc->ha_shutdown = 2;
635	wakeup(&softc->ha_wakeup);
636	mtx_unlock(&softc->ha_lock);
637	kthread_exit();
638}
639
640static int
641ctl_ha_peer_sysctl(SYSCTL_HANDLER_ARGS)
642{
643	struct ha_softc *softc = (struct ha_softc *)arg1;
644	struct sockaddr_in *sa;
645	int error, b1, b2, b3, b4, p, num;
646	char buf[128];
647
648	strlcpy(buf, softc->ha_peer, sizeof(buf));
649	error = sysctl_handle_string(oidp, buf, sizeof(buf), req);
650	if ((error != 0) || (req->newptr == NULL) ||
651	    strncmp(buf, softc->ha_peer, sizeof(buf)) == 0)
652		return (error);
653
654	sa = &softc->ha_peer_in;
655	mtx_lock(&softc->ha_lock);
656	if ((num = sscanf(buf, "connect %d.%d.%d.%d:%d",
657	    &b1, &b2, &b3, &b4, &p)) >= 4) {
658		softc->ha_connect = 1;
659		softc->ha_listen = 0;
660	} else if ((num = sscanf(buf, "listen %d.%d.%d.%d:%d",
661	    &b1, &b2, &b3, &b4, &p)) >= 4) {
662		softc->ha_connect = 0;
663		softc->ha_listen = 1;
664	} else {
665		softc->ha_connect = 0;
666		softc->ha_listen = 0;
667		if (buf[0] != 0) {
668			buf[0] = 0;
669			error = EINVAL;
670		}
671	}
672	strlcpy(softc->ha_peer, buf, sizeof(softc->ha_peer));
673	if (softc->ha_connect || softc->ha_listen) {
674		memset(sa, 0, sizeof(*sa));
675		sa->sin_len = sizeof(struct sockaddr_in);
676		sa->sin_family = AF_INET;
677		sa->sin_port = htons((num >= 5) ? p : 999);
678		sa->sin_addr.s_addr =
679		    htonl((b1 << 24) + (b2 << 16) + (b3 << 8) + b4);
680	}
681	softc->ha_disconnect = 2;
682	softc->ha_wakeup = 1;
683	mtx_unlock(&softc->ha_lock);
684	wakeup(&softc->ha_wakeup);
685	return (error);
686}
687
688ctl_ha_status
689ctl_ha_msg_register(ctl_ha_channel channel, ctl_evt_handler handler)
690{
691	struct ha_softc *softc = &ha_softc;
692
693	KASSERT(channel < CTL_HA_CHAN_MAX,
694	    ("Wrong CTL HA channel %d", channel));
695	softc->ha_handler[channel] = handler;
696	return (CTL_HA_STATUS_SUCCESS);
697}
698
699ctl_ha_status
700ctl_ha_msg_deregister(ctl_ha_channel channel)
701{
702	struct ha_softc *softc = &ha_softc;
703
704	KASSERT(channel < CTL_HA_CHAN_MAX,
705	    ("Wrong CTL HA channel %d", channel));
706	softc->ha_handler[channel] = NULL;
707	return (CTL_HA_STATUS_SUCCESS);
708}
709
710/*
711 * Receive a message of the specified size.
712 */
713ctl_ha_status
714ctl_ha_msg_recv(ctl_ha_channel channel, void *addr, size_t len,
715		int wait)
716{
717	struct ha_softc *softc = &ha_softc;
718	struct uio uio;
719	struct iovec iov;
720	int error, flags;
721
722	if (!softc->ha_connected)
723		return (CTL_HA_STATUS_DISCONNECT);
724
725	iov.iov_base = addr;
726	iov.iov_len = len;
727	uio.uio_iov = &iov;
728	uio.uio_iovcnt = 1;
729	uio.uio_rw = UIO_READ;
730	uio.uio_segflg = UIO_SYSSPACE;
731	uio.uio_td = curthread;
732	uio.uio_resid = len;
733	flags = wait ? 0 : MSG_DONTWAIT;
734	error = soreceive(softc->ha_so, NULL, &uio, NULL, NULL, &flags);
735	if (error == 0)
736		return (CTL_HA_STATUS_SUCCESS);
737
738	/* Consider all errors fatal for HA sanity. */
739	mtx_lock(&softc->ha_lock);
740	if (softc->ha_connected) {
741		softc->ha_disconnect = 1;
742		softc->ha_wakeup = 1;
743		wakeup(&softc->ha_wakeup);
744	}
745	mtx_unlock(&softc->ha_lock);
746	return (CTL_HA_STATUS_ERROR);
747}
748
749/*
750 * Send a message of the specified size.
751 */
752ctl_ha_status
753ctl_ha_msg_send2(ctl_ha_channel channel, const void *addr, size_t len,
754    const void *addr2, size_t len2, int wait)
755{
756	struct ha_softc *softc = &ha_softc;
757	struct mbuf *mb, *newmb;
758	struct ha_msg_wire hdr;
759	size_t copylen, off;
760
761	if (!softc->ha_connected)
762		return (CTL_HA_STATUS_DISCONNECT);
763
764	newmb = m_getm2(NULL, sizeof(hdr) + len + len2, wait, MT_DATA,
765	    M_PKTHDR);
766	if (newmb == NULL) {
767		/* Consider all errors fatal for HA sanity. */
768		mtx_lock(&softc->ha_lock);
769		if (softc->ha_connected) {
770			softc->ha_disconnect = 1;
771			softc->ha_wakeup = 1;
772			wakeup(&softc->ha_wakeup);
773		}
774		mtx_unlock(&softc->ha_lock);
775		printf("%s: Can't allocate mbuf chain\n", __func__);
776		return (CTL_HA_STATUS_ERROR);
777	}
778	hdr.channel = channel;
779	hdr.length = len + len2;
780	mb = newmb;
781	memcpy(mtodo(mb, 0), &hdr, sizeof(hdr));
782	mb->m_len += sizeof(hdr);
783	off = 0;
784	for (; mb != NULL && off < len; mb = mb->m_next) {
785		copylen = min(M_TRAILINGSPACE(mb), len - off);
786		memcpy(mtodo(mb, mb->m_len), (const char *)addr + off, copylen);
787		mb->m_len += copylen;
788		off += copylen;
789		if (off == len)
790			break;
791	}
792	KASSERT(off == len, ("%s: off (%zu) != len (%zu)", __func__,
793	    off, len));
794	off = 0;
795	for (; mb != NULL && off < len2; mb = mb->m_next) {
796		copylen = min(M_TRAILINGSPACE(mb), len2 - off);
797		memcpy(mtodo(mb, mb->m_len), (const char *)addr2 + off, copylen);
798		mb->m_len += copylen;
799		off += copylen;
800	}
801	KASSERT(off == len2, ("%s: off (%zu) != len2 (%zu)", __func__,
802	    off, len2));
803	newmb->m_pkthdr.len = sizeof(hdr) + len + len2;
804
805	mtx_lock(&softc->ha_lock);
806	if (!softc->ha_connected) {
807		mtx_unlock(&softc->ha_lock);
808		m_freem(newmb);
809		return (CTL_HA_STATUS_DISCONNECT);
810	}
811	mbufq_enqueue(&softc->ha_sendq, newmb);
812	softc->ha_wakeup = 1;
813	mtx_unlock(&softc->ha_lock);
814	wakeup(&softc->ha_wakeup);
815	return (CTL_HA_STATUS_SUCCESS);
816}
817
818ctl_ha_status
819ctl_ha_msg_send(ctl_ha_channel channel, const void *addr, size_t len,
820    int wait)
821{
822
823	return (ctl_ha_msg_send2(channel, addr, len, NULL, 0, wait));
824}
825
826ctl_ha_status
827ctl_ha_msg_abort(ctl_ha_channel channel)
828{
829	struct ha_softc *softc = &ha_softc;
830
831	mtx_lock(&softc->ha_lock);
832	softc->ha_disconnect = 1;
833	softc->ha_wakeup = 1;
834	mtx_unlock(&softc->ha_lock);
835	wakeup(&softc->ha_wakeup);
836	return (CTL_HA_STATUS_SUCCESS);
837}
838
839/*
840 * Allocate a data transfer request structure.
841 */
842struct ctl_ha_dt_req *
843ctl_dt_req_alloc(void)
844{
845
846	return (malloc(sizeof(struct ctl_ha_dt_req), M_CTL, M_WAITOK | M_ZERO));
847}
848
849/*
850 * Free a data transfer request structure.
851 */
852void
853ctl_dt_req_free(struct ctl_ha_dt_req *req)
854{
855
856	free(req, M_CTL);
857}
858
859/*
860 * Issue a DMA request for a single buffer.
861 */
862ctl_ha_status
863ctl_dt_single(struct ctl_ha_dt_req *req)
864{
865	struct ha_softc *softc = &ha_softc;
866	struct ha_dt_msg_wire wire_dt;
867	ctl_ha_status status;
868
869	wire_dt.command = req->command;
870	wire_dt.size = req->size;
871	wire_dt.local = req->local;
872	wire_dt.remote = req->remote;
873	if (req->command == CTL_HA_DT_CMD_READ && req->callback != NULL) {
874		mtx_lock(&softc->ha_lock);
875		TAILQ_INSERT_TAIL(&softc->ha_dts, req, links);
876		mtx_unlock(&softc->ha_lock);
877		ctl_ha_msg_send(CTL_HA_CHAN_DATA, &wire_dt, sizeof(wire_dt),
878		    M_WAITOK);
879		return (CTL_HA_STATUS_WAIT);
880	}
881	if (req->command == CTL_HA_DT_CMD_READ) {
882		status = ctl_ha_msg_send(CTL_HA_CHAN_DATA, &wire_dt,
883		    sizeof(wire_dt), M_WAITOK);
884	} else {
885		status = ctl_ha_msg_send2(CTL_HA_CHAN_DATA, &wire_dt,
886		    sizeof(wire_dt), req->local, req->size, M_WAITOK);
887	}
888	return (status);
889}
890
891static void
892ctl_dt_event_handler(ctl_ha_channel channel, ctl_ha_event event, int param)
893{
894	struct ha_softc *softc = &ha_softc;
895	struct ctl_ha_dt_req *req;
896	ctl_ha_status isc_status;
897
898	if (event == CTL_HA_EVT_MSG_RECV) {
899		struct ha_dt_msg_wire wire_dt;
900		uint8_t *tmp;
901		int size;
902
903		size = min(sizeof(wire_dt), param);
904		isc_status = ctl_ha_msg_recv(CTL_HA_CHAN_DATA, &wire_dt,
905					     size, M_WAITOK);
906		if (isc_status != CTL_HA_STATUS_SUCCESS) {
907			printf("%s: Error receiving message: %d\n",
908			    __func__, isc_status);
909			return;
910		}
911
912		if (wire_dt.command == CTL_HA_DT_CMD_READ) {
913			wire_dt.command = CTL_HA_DT_CMD_WRITE;
914			tmp = wire_dt.local;
915			wire_dt.local = wire_dt.remote;
916			wire_dt.remote = tmp;
917			ctl_ha_msg_send2(CTL_HA_CHAN_DATA, &wire_dt,
918			    sizeof(wire_dt), wire_dt.local, wire_dt.size,
919			    M_WAITOK);
920		} else if (wire_dt.command == CTL_HA_DT_CMD_WRITE) {
921			isc_status = ctl_ha_msg_recv(CTL_HA_CHAN_DATA,
922			    wire_dt.remote, wire_dt.size, M_WAITOK);
923			mtx_lock(&softc->ha_lock);
924			TAILQ_FOREACH(req, &softc->ha_dts, links) {
925				if (req->local == wire_dt.remote) {
926					TAILQ_REMOVE(&softc->ha_dts, req, links);
927					break;
928				}
929			}
930			mtx_unlock(&softc->ha_lock);
931			if (req) {
932				req->ret = isc_status;
933				req->callback(req);
934			}
935		}
936	} else if (event == CTL_HA_EVT_LINK_CHANGE) {
937		CTL_DEBUG_PRINT(("%s: Link state change to %d\n", __func__,
938		    param));
939		if (param != CTL_HA_LINK_ONLINE) {
940			mtx_lock(&softc->ha_lock);
941			while ((req = TAILQ_FIRST(&softc->ha_dts)) != NULL) {
942				TAILQ_REMOVE(&softc->ha_dts, req, links);
943				mtx_unlock(&softc->ha_lock);
944				req->ret = CTL_HA_STATUS_DISCONNECT;
945				req->callback(req);
946				mtx_lock(&softc->ha_lock);
947			}
948			mtx_unlock(&softc->ha_lock);
949		}
950	} else {
951		printf("%s: Unknown event %d\n", __func__, event);
952	}
953}
954
955
956ctl_ha_status
957ctl_ha_msg_init(struct ctl_softc *ctl_softc)
958{
959	struct ha_softc *softc = &ha_softc;
960	int error;
961
962	softc->ha_ctl_softc = ctl_softc;
963	mtx_init(&softc->ha_lock, "CTL HA mutex", NULL, MTX_DEF);
964	mbufq_init(&softc->ha_sendq, INT_MAX);
965	TAILQ_INIT(&softc->ha_dts);
966	error = kproc_kthread_add(ctl_ha_conn_thread, softc,
967	    &ctl_softc->ctl_proc, NULL, 0, 0, "ctl", "ha_tx");
968	if (error != 0) {
969		printf("error creating CTL HA connection thread!\n");
970		mtx_destroy(&softc->ha_lock);
971		return (CTL_HA_STATUS_ERROR);
972	}
973	softc->ha_shutdown_eh = EVENTHANDLER_REGISTER(shutdown_pre_sync,
974	    ctl_ha_msg_shutdown, ctl_softc, SHUTDOWN_PRI_FIRST);
975	SYSCTL_ADD_PROC(&ctl_softc->sysctl_ctx,
976	    SYSCTL_CHILDREN(ctl_softc->sysctl_tree),
977	    OID_AUTO, "ha_peer", CTLTYPE_STRING | CTLFLAG_RWTUN,
978	    softc, 0, ctl_ha_peer_sysctl, "A", "HA peer connection method");
979
980	if (ctl_ha_msg_register(CTL_HA_CHAN_DATA, ctl_dt_event_handler)
981	    != CTL_HA_STATUS_SUCCESS) {
982		printf("%s: ctl_ha_msg_register failed.\n", __func__);
983	}
984
985	return (CTL_HA_STATUS_SUCCESS);
986};
987
988void
989ctl_ha_msg_shutdown(struct ctl_softc *ctl_softc)
990{
991	struct ha_softc *softc = &ha_softc;
992
993	/* Disconnect and shutdown threads. */
994	mtx_lock(&softc->ha_lock);
995	if (softc->ha_shutdown < 2) {
996		softc->ha_shutdown = 1;
997		softc->ha_wakeup = 1;
998		wakeup(&softc->ha_wakeup);
999		while (softc->ha_shutdown < 2) {
1000			msleep(&softc->ha_wakeup, &softc->ha_lock, 0,
1001			    "shutdown", hz);
1002		}
1003	}
1004	mtx_unlock(&softc->ha_lock);
1005};
1006
1007ctl_ha_status
1008ctl_ha_msg_destroy(struct ctl_softc *ctl_softc)
1009{
1010	struct ha_softc *softc = &ha_softc;
1011
1012	if (softc->ha_shutdown_eh != NULL) {
1013		EVENTHANDLER_DEREGISTER(shutdown_pre_sync,
1014		    softc->ha_shutdown_eh);
1015		softc->ha_shutdown_eh = NULL;
1016	}
1017
1018	ctl_ha_msg_shutdown(ctl_softc);	/* Just in case. */
1019
1020	if (ctl_ha_msg_deregister(CTL_HA_CHAN_DATA) != CTL_HA_STATUS_SUCCESS)
1021		printf("%s: ctl_ha_msg_deregister failed.\n", __func__);
1022
1023	mtx_destroy(&softc->ha_lock);
1024	return (CTL_HA_STATUS_SUCCESS);
1025};
1026