sbin/hastd/primary.c

76 struct hio {
112 	TAILQ_ENTRY(hio)	*hio_next;
121 static TAILQ_HEAD(, hio) hio_free_list;
130 static TAILQ_HEAD(, hio) *hio_send_list;
140 static TAILQ_HEAD(, hio) *hio_recv_list;
149 static TAILQ_HEAD(, hio) hio_done_list;
184 #define	QUEUE_INSERT1(hio, name, ncomp)	do {				\
188 	TAILQ_INSERT_TAIL(&hio_##name##_list[(ncomp)], (hio),		\
193 #define	QUEUE_INSERT2(hio, name)	do {				\
197 	TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_##name##_next);\
201 #define	QUEUE_TAKE1(hio, name, ncomp, timeout)	do {			\
206 	while (((hio) = TAILQ_FIRST(&hio_##name##_list[(ncomp)])) == NULL && !_last) { \
212 	if (hio != NULL) {						\
215 		TAILQ_REMOVE(&hio_##name##_list[(ncomp)], (hio),	\
220 #define	QUEUE_TAKE2(hio, name)	do {					\
222 	while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) {	\
228 	TAILQ_REMOVE(&hio_##name##_list, (hio), hio_##name##_next);	\
232 #define ISFULLSYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_FULLSYNC)
233 #define ISMEMSYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_MEMSYNC)
234 #define ISASYNC(hio)	((hio)->hio_replication == HAST_REPLICATION_ASYNC)
236 #define	SYNCREQ(hio)		do {					\
237 	(hio)->hio_ggio.gctl_unit = -1;					\
238 	(hio)->hio_ggio.gctl_seq = 1;					\
240 #define	ISSYNCREQ(hio)		((hio)->hio_ggio.gctl_unit == -1)
241 #define	SYNCREQDONE(hio)	do { (hio)->hio_ggio.gctl_unit = -2; } while (0)
242 #define	ISSYNCREQDONE(hio)	((hio)->hio_ggio.gctl_unit == -2)
244 #define ISMEMSYNCWRITE(hio)	(ISMEMSYNC(hio) &&			\
245 	    (hio)->hio_ggio.gctl_cmd == BIO_WRITE && !ISSYNCREQ(hio))
381 	struct hio *hio;
473 		hio = malloc(sizeof(*hio));
474 		if (hio == NULL) {
476 			    "Unable to allocate %zu bytes of memory for hio request.",
477 			    sizeof(*hio));
479 		refcnt_init(&hio->hio_countdown, 0);
480 		hio->hio_errors = malloc(sizeof(hio->hio_errors[0]) * ncomps);
481 		if (hio->hio_errors == NULL) {
483 			    "Unable allocate %zu bytes of memory for hio errors.",
484 			    sizeof(hio->hio_errors[0]) * ncomps);
486 		hio->hio_next = malloc(sizeof(hio->hio_next[0]) * ncomps);
487 		if (hio->hio_next == NULL) {
490 			    sizeof(hio->hio_next[0]) * ncomps);
492 		hio->hio_ggio.gctl_version = G_GATE_VERSION;
493 		hio->hio_ggio.gctl_data = malloc(MAXPHYS);
494 		if (hio->hio_ggio.gctl_data == NULL) {
499 		hio->hio_ggio.gctl_length = MAXPHYS;
500 		hio->hio_ggio.gctl_error = 0;
501 		TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_free_next);
1181 write_complete(struct hast_resource *res, struct hio *hio)
1186 	PJDLOG_ASSERT(!hio->hio_done);
1188 	ggio = &hio->hio_ggio;
1210 	hio->hio_done = true;
1229 	struct hio *hio;
1235 		QUEUE_TAKE2(hio, free);
1236 		pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio);
1237 		ggio = &hio->hio_ggio;
1241 		hio->hio_done = false;
1242 		hio->hio_replication = res->hr_replication;
1245 		    hio);
1260 				    hio);
1280 			hio->hio_errors[ii] = EINVAL;
1283 		    hio);
1358 			if (ISMEMSYNC(hio)) {
1359 				hio->hio_memsyncacked = false;
1360 				refcnt_init(&hio->hio_writecount, ncomps);
1371 		    "ggate_recv: (%p) Moving request to the send queues.", hio);
1372 		refcnt_init(&hio->hio_countdown, ncomps);
1374 			QUEUE_INSERT1(hio, send, ii);
1389 	struct hio *hio;
1400 		QUEUE_TAKE1(hio, send, ncomp, 0);
1401 		pjdlog_debug(2, "local_send: (%p) Got request.", hio);
1402 		ggio = &hio->hio_ggio;
1409 				hio->hio_errors[ncomp] = 0;
1410 			else if (!ISSYNCREQ(hio)) {
1423 				QUEUE_INSERT1(hio, send, rncomp);
1432 				hio->hio_errors[ncomp] = errno;
1437 				hio->hio_errors[ncomp] = EIO;
1442 				hio->hio_errors[ncomp] = 0;
1443 				if (ISASYNC(hio)) {
1445 					write_complete(res, hio);
1454 				hio->hio_errors[ncomp] = errno;
1459 				hio->hio_errors[ncomp] = 0;
1472 				hio->hio_errors[ncomp] = errno;
1477 				hio->hio_errors[ncomp] = 0;
1481 		if (ISMEMSYNCWRITE(hio)) {
1482 			if (refcnt_release(&hio->hio_writecount) == 0) {
1483 				write_complete(res, hio);
1486 		if (refcnt_release(&hio->hio_countdown) > 0)
1488 		if (ISSYNCREQ(hio)) {
1490 			SYNCREQDONE(hio);
1496 			    hio);
1497 			QUEUE_INSERT2(hio, done);
1551 	struct hio *hio;
1565 		QUEUE_TAKE1(hio, send, ncomp, HAST_KEEPALIVE);
1566 		if (hio == NULL) {
1574 		pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
1575 		ggio = &hio->hio_ggio;
1609 		if (ISMEMSYNCWRITE(hio))
1612 			hio->hio_errors[ncomp] = nv_error(nv);
1615 			    hio);
1628 			hio->hio_errors[ncomp] = ENOTCONN;
1638 		    hio);
1641 		TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1646 			hio->hio_errors[ncomp] = errno;
1649 			    "remote_send: (%p) Unable to send request.", hio);
1652 			    strerror(hio->hio_errors[ncomp]));
1663 		if (ISSYNCREQ(hio)) {
1664 			if (refcnt_release(&hio->hio_countdown) > 0)
1667 			SYNCREQDONE(hio);
1680 			if (ISMEMSYNCWRITE(hio)) {
1681 				if (refcnt_release(&hio->hio_writecount) == 0) {
1682 					if (hio->hio_errors[0] == 0)
1683 						write_complete(res, hio);
1687 		if (refcnt_release(&hio->hio_countdown) > 0)
1691 		    hio);
1692 		QUEUE_INSERT2(hio, done);
1707 	struct hio *hio;
1737 			hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
1738 			PJDLOG_ASSERT(hio != NULL);
1739 			TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1743 			hio->hio_errors[ncomp] = ENOTCONN;
1762 		TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
1763 			if (hio->hio_ggio.gctl_seq == seq) {
1764 				TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1771 		if (hio == NULL) {
1777 		ggio = &hio->hio_ggio;
1781 			hio->hio_errors[ncomp] = error;
1797 				hio->hio_errors[ncomp] = errno;
1814 		hio->hio_errors[ncomp] = 0;
1817 		if (ISMEMSYNCWRITE(hio)) {
1818 			if (!hio->hio_memsyncacked) {
1820 				    hio->hio_errors[ncomp] != 0);
1822 				if (refcnt_release(&hio->hio_writecount) == 0) {
1823 					if (hio->hio_errors[0] == 0)
1824 						write_complete(res, hio);
1826 				hio->hio_memsyncacked = true;
1827 				if (hio->hio_errors[ncomp] == 0) {
1830 					    "back to the recv queue.", hio);
1833 					    hio, hio_next[ncomp]);
1843 		if (refcnt_release(&hio->hio_countdown) > 0)
1845 		if (ISSYNCREQ(hio)) {
1847 			SYNCREQDONE(hio);
1853 			    hio);
1854 			QUEUE_INSERT2(hio, done);
1869 	struct hio *hio;
1876 		QUEUE_TAKE2(hio, done);
1877 		pjdlog_debug(2, "ggate_send: (%p) Got request.", hio);
1878 		ggio = &hio->hio_ggio;
1880 			if (hio->hio_errors[ii] == 0) {
1897 				ggio->gctl_error = hio->hio_errors[1];
1899 				ggio->gctl_error = hio->hio_errors[0];
1921 			if (!hio->hio_done)
1922 				write_complete(res, hio);
1929 		if (hio->hio_errors[0]) {
1946 		    "ggate_send: (%p) Moving request to the free queue.", hio);
1947 		QUEUE_INSERT2(hio, free);
1960 	struct hio *hio;
2069 		QUEUE_TAKE2(hio, free);
2070 		pjdlog_debug(2, "sync: (%p) Got free request.", hio);
2101 		SYNCREQ(hio);
2102 		ggio = &hio->hio_ggio;
2107 		hio->hio_done = false;
2108 		hio->hio_replication = res->hr_replication;
2110 			hio->hio_errors[ii] = EINVAL;
2112 		    hio);
2114 		    hio);
2133 		refcnt_init(&hio->hio_countdown, 1);
2134 		QUEUE_INSERT1(hio, send, ncomp);
2140 		while (!ISSYNCREQDONE(hio))
2144 		if (hio->hio_errors[ncomp] != 0) {
2146 			    strerror(hio->hio_errors[ncomp]));
2154 		SYNCREQ(hio);
2157 			hio->hio_errors[ii] = EINVAL;
2159 		    hio);
2161 		    hio);
2182 		    hio);
2183 		refcnt_init(&hio->hio_countdown, 1);
2184 		QUEUE_INSERT1(hio, send, ncomp);
2190 		while (!ISSYNCREQDONE(hio))
2194 		if (hio->hio_errors[ncomp] != 0) {
2196 			    strerror(hio->hio_errors[ncomp]));
2208 		    hio);
2209 		QUEUE_INSERT2(hio, free);