sbin/hastd/primary.c

74 struct hio {
110 	TAILQ_ENTRY(hio)	*hio_next;
119 static TAILQ_HEAD(, hio) hio_free_list;
128 static TAILQ_HEAD(, hio) *hio_send_list;
138 static TAILQ_HEAD(, hio) *hio_recv_list;
147 static TAILQ_HEAD(, hio) hio_done_list;
182 #define	QUEUE_INSERT1(hio, name, ncomp)	do {				\
186 	TAILQ_INSERT_TAIL(&hio_##name##_list[(ncomp)], (hio),		\
191 #define	QUEUE_INSERT2(hio, name)	do {				\
195 	TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_##name##_next);\
199 #define	QUEUE_TAKE1(hio, name, ncomp, timeout)	do {			\
204 	while (((hio) = TAILQ_FIRST(&hio_##name##_list[(ncomp)])) == NULL && !_last) { \
210 	if (hio != NULL) {						\
213 		TAILQ_REMOVE(&hio_##name##_list[(ncomp)], (hio),	\
218 #define	QUEUE_TAKE2(hio, name)	do {					\
220 	while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) {	\
226 	TAILQ_REMOVE(&hio_##name##_list, (hio), hio_##name##_next);	\
230 #define ISFULLSYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_FULLSYNC)
231 #define ISMEMSYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_MEMSYNC)
232 #define ISASYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_ASYNC)
234 #define	SYNCREQ(hio)		do {					\
235 	(hio)->hio_ggio.gctl_unit = -1;					\
236 	(hio)->hio_ggio.gctl_seq = 1;					\
238 #define	ISSYNCREQ(hio)		((hio)->hio_ggio.gctl_unit == -1)
239 #define	SYNCREQDONE(hio)	do { (hio)->hio_ggio.gctl_unit = -2; } while (0)
240 #define	ISSYNCREQDONE(hio)	((hio)->hio_ggio.gctl_unit == -2)
242 #define ISMEMSYNCWRITE(hio)	(ISMEMSYNC(hio) &&			\
243 	    (hio)->hio_ggio.gctl_cmd == BIO_WRITE && !ISSYNCREQ(hio))
380 	struct hio *hio;
472 		hio = malloc(sizeof(*hio));
473 		if (hio == NULL) {
475 			    "Unable to allocate %zu bytes of memory for hio request.",
476 			    sizeof(*hio));
478 		refcnt_init(&hio->hio_countdown, 0);
479 		hio->hio_errors = malloc(sizeof(hio->hio_errors[0]) * ncomps);
480 		if (hio->hio_errors == NULL) {
482 			    "Unable allocate %zu bytes of memory for hio errors.",
483 			    sizeof(hio->hio_errors[0]) * ncomps);
485 		hio->hio_next = malloc(sizeof(hio->hio_next[0]) * ncomps);
486 		if (hio->hio_next == NULL) {
489 			    sizeof(hio->hio_next[0]) * ncomps);
491 		hio->hio_ggio.gctl_version = G_GATE_VERSION;
492 		hio->hio_ggio.gctl_data = malloc(MAXPHYS);
493 		if (hio->hio_ggio.gctl_data == NULL) {
498 		hio->hio_ggio.gctl_length = MAXPHYS;
499 		hio->hio_ggio.gctl_error = 0;
500 		TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_free_next);
1180 write_complete(struct hast_resource *res, struct hio *hio)
1185 	PJDLOG_ASSERT(!hio->hio_done);
1187 	ggio = &hio->hio_ggio;
1209 	hio->hio_done = true;
1228 	struct hio *hio;
1234 		QUEUE_TAKE2(hio, free);
1235 		pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio);
1236 		ggio = &hio->hio_ggio;
1240 		hio->hio_done = false;
1241 		hio->hio_replication = res->hr_replication;
1244 		    hio);
1259 				    hio);
1279 			hio->hio_errors[ii] = EINVAL;
1282 		    hio);
1357 			if (ISMEMSYNC(hio)) {
1358 				hio->hio_memsyncacked = false;
1359 				refcnt_init(&hio->hio_writecount, ncomps);
1370 		    "ggate_recv: (%p) Moving request to the send queues.", hio);
1371 		refcnt_init(&hio->hio_countdown, ncomps);
1373 			QUEUE_INSERT1(hio, send, ii);
1388 	struct hio *hio;
1399 		QUEUE_TAKE1(hio, send, ncomp, 0);
1400 		pjdlog_debug(2, "local_send: (%p) Got request.", hio);
1401 		ggio = &hio->hio_ggio;
1408 				hio->hio_errors[ncomp] = 0;
1409 			else if (!ISSYNCREQ(hio)) {
1422 				QUEUE_INSERT1(hio, send, rncomp);
1431 				hio->hio_errors[ncomp] = errno;
1436 				hio->hio_errors[ncomp] = EIO;
1441 				hio->hio_errors[ncomp] = 0;
1442 				if (ISASYNC(hio)) {
1444 					write_complete(res, hio);
1453 				hio->hio_errors[ncomp] = errno;
1458 				hio->hio_errors[ncomp] = 0;
1471 				hio->hio_errors[ncomp] = errno;
1476 				hio->hio_errors[ncomp] = 0;
1480 		if (ISMEMSYNCWRITE(hio)) {
1481 			if (refcnt_release(&hio->hio_writecount) == 0) {
1482 				write_complete(res, hio);
1485 		if (refcnt_release(&hio->hio_countdown) > 0)
1487 		if (ISSYNCREQ(hio)) {
1489 			SYNCREQDONE(hio);
1495 			    hio);
1496 			QUEUE_INSERT2(hio, done);
1550 	struct hio *hio;
1564 		QUEUE_TAKE1(hio, send, ncomp, HAST_KEEPALIVE);
1565 		if (hio == NULL) {
1573 		pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
1574 		ggio = &hio->hio_ggio;
1608 		if (ISMEMSYNCWRITE(hio))
1611 			hio->hio_errors[ncomp] = nv_error(nv);
1614 			    hio);
1627 			hio->hio_errors[ncomp] = ENOTCONN;
1637 		    hio);
1640 		TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1645 			hio->hio_errors[ncomp] = errno;
1648 			    "remote_send: (%p) Unable to send request.", hio);
1651 			    strerror(hio->hio_errors[ncomp]));
1662 		if (ISSYNCREQ(hio)) {
1663 			if (refcnt_release(&hio->hio_countdown) > 0)
1666 			SYNCREQDONE(hio);
1679 			if (ISMEMSYNCWRITE(hio)) {
1680 				if (refcnt_release(&hio->hio_writecount) == 0) {
1681 					if (hio->hio_errors[0] == 0)
1682 						write_complete(res, hio);
1686 		if (refcnt_release(&hio->hio_countdown) > 0)
1690 		    hio);
1691 		QUEUE_INSERT2(hio, done);
1706 	struct hio *hio;
1736 			hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
1737 			PJDLOG_ASSERT(hio != NULL);
1738 			TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1742 			hio->hio_errors[ncomp] = ENOTCONN;
1761 		TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
1762 			if (hio->hio_ggio.gctl_seq == seq) {
1763 				TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1770 		if (hio == NULL) {
1776 		ggio = &hio->hio_ggio;
1780 			hio->hio_errors[ncomp] = error;
1796 				hio->hio_errors[ncomp] = errno;
1813 		hio->hio_errors[ncomp] = 0;
1816 		if (ISMEMSYNCWRITE(hio)) {
1817 			if (!hio->hio_memsyncacked) {
1819 				    hio->hio_errors[ncomp] != 0);
1821 				if (refcnt_release(&hio->hio_writecount) == 0) {
1822 					if (hio->hio_errors[0] == 0)
1823 						write_complete(res, hio);
1825 				hio->hio_memsyncacked = true;
1826 				if (hio->hio_errors[ncomp] == 0) {
1829 					    "back to the recv queue.", hio);
1832 					    hio, hio_next[ncomp]);
1842 		if (refcnt_release(&hio->hio_countdown) > 0)
1844 		if (ISSYNCREQ(hio)) {
1846 			SYNCREQDONE(hio);
1852 			    hio);
1853 			QUEUE_INSERT2(hio, done);
1868 	struct hio *hio;
1875 		QUEUE_TAKE2(hio, done);
1876 		pjdlog_debug(2, "ggate_send: (%p) Got request.", hio);
1877 		ggio = &hio->hio_ggio;
1879 			if (hio->hio_errors[ii] == 0) {
1896 				ggio->gctl_error = hio->hio_errors[1];
1898 				ggio->gctl_error = hio->hio_errors[0];
1920 			if (!hio->hio_done)
1921 				write_complete(res, hio);
1928 		if (hio->hio_errors[0]) {
1945 		    "ggate_send: (%p) Moving request to the free queue.", hio);
1946 		QUEUE_INSERT2(hio, free);
1959 	struct hio *hio;
2068 		QUEUE_TAKE2(hio, free);
2069 		pjdlog_debug(2, "sync: (%p) Got free request.", hio);
2100 		SYNCREQ(hio);
2101 		ggio = &hio->hio_ggio;
2106 		hio->hio_done = false;
2107 		hio->hio_replication = res->hr_replication;
2109 			hio->hio_errors[ii] = EINVAL;
2111 		    hio);
2113 		    hio);
2132 		refcnt_init(&hio->hio_countdown, 1);
2133 		QUEUE_INSERT1(hio, send, ncomp);
2139 		while (!ISSYNCREQDONE(hio))
2143 		if (hio->hio_errors[ncomp] != 0) {
2145 			    strerror(hio->hio_errors[ncomp]));
2153 		SYNCREQ(hio);
2156 			hio->hio_errors[ii] = EINVAL;
2158 		    hio);
2160 		    hio);
2181 		    hio);
2182 		refcnt_init(&hio->hio_countdown, 1);
2183 		QUEUE_INSERT1(hio, send, ncomp);
2189 		while (!ISSYNCREQDONE(hio))
2193 		if (hio->hio_errors[ncomp] != 0) {
2195 			    strerror(hio->hio_errors[ncomp]));
2207 		    hio);
2208 		QUEUE_INSERT2(hio, free);