FreeBSD/sbin/hastd/primary.c
/*
 * Thread receives answer from secondary node and passes it to ggate_send
 * thread.
 */
Thread receives answer from secondary node and passes it to ggate_send
thread.
Thread receives answer from secondary node and passes it to ggate_send thread .
static void *
remote_recv_thread(void *arg)
{
	struct hast_resource *res = arg;
	struct g_gate_ctl_io *ggio;
	struct hio *hio;
	struct nv *nv;
	unsigned int ncomp;
	uint64_t seq;
	bool memsyncack;
	int error;
	ncomp = 1;
	for (;;) {
		mtx_lock(&hio_recv_list_lock[ncomp]);
		while (TAILQ_EMPTY(&hio_recv_list[ncomp])) {
			pjdlog_debug(2, "remote_recv: No requests, waiting.");
			cv_wait(&hio_recv_list_cond[ncomp],
			    &hio_recv_list_lock[ncomp]);
		}
		mtx_unlock(&hio_recv_list_lock[ncomp]);
		memsyncack = false;
		rw_rlock(&hio_remote_lock[ncomp]);
		if (!ISCONNECTED(res, ncomp)) {
			rw_unlock(&hio_remote_lock[ncomp]);
			mtx_lock(&hio_recv_list_lock[ncomp]);
			hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
			PJDLOG_ASSERT(hio != NULL);
			TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
			    hio_next[ncomp]);
			hio_recv_list_size[ncomp]--;
			mtx_unlock(&hio_recv_list_lock[ncomp]);
			hio->hio_errors[ncomp] = ENOTCONN;
			goto done_queue;
		}
		if (hast_proto_recv_hdr(res->hr_remotein, &nv) == -1) {
			pjdlog_errno(LOG_ERR,
			    "Unable to receive reply header");
			rw_unlock(&hio_remote_lock[ncomp]);
			remote_close(res, ncomp);
			continue;
		}
		rw_unlock(&hio_remote_lock[ncomp]);
		seq = nv_get_uint64(nv, "seq");
		if (seq == 0) {
			pjdlog_error("Header contains no 'seq' field.");
			nv_free(nv);
			continue;
		}
		memsyncack = nv_exists(nv, "received");
		mtx_lock(&hio_recv_list_lock[ncomp]);
		TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
			if (hio->hio_ggio.gctl_seq == seq) {
				TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
				    hio_next[ncomp]);
				hio_recv_list_size[ncomp]--;
				break;
			}
		}
		mtx_unlock(&hio_recv_list_lock[ncomp]);
		if (hio == NULL) {
			pjdlog_error("Found no request matching received 'seq' field (%ju).",
			    (uintmax_t)seq);
			nv_free(nv);
			continue;
		}
		ggio = &hio->hio_ggio;
		error = nv_get_int16(nv, "error");
		if (error != 0) {
			hio->hio_errors[ncomp] = error;
			reqlog(LOG_WARNING, 0, ggio,
			    "Remote request failed (%s): ", strerror(error));
			nv_free(nv);
			goto done_queue;
		}
		switch (ggio->gctl_cmd) {
		case BIO_READ:
			rw_rlock(&hio_remote_lock[ncomp]);
			if (!ISCONNECTED(res, ncomp)) {
				rw_unlock(&hio_remote_lock[ncomp]);
				nv_free(nv);
				goto done_queue;
			}
			if (hast_proto_recv_data(res, res->hr_remotein, nv,
			    ggio->gctl_data, ggio->gctl_length) == -1) {
				hio->hio_errors[ncomp] = errno;
				pjdlog_errno(LOG_ERR,
				    "Unable to receive reply data");
				rw_unlock(&hio_remote_lock[ncomp]);
				nv_free(nv);
				remote_close(res, ncomp);
				goto done_queue;
			}
			rw_unlock(&hio_remote_lock[ncomp]);
			break;
		case BIO_WRITE:
		case BIO_DELETE:
		case BIO_FLUSH:
			break;
		default:
			PJDLOG_ABORT("invalid condition");
		}
		hio->hio_errors[ncomp] = 0;
		nv_free(nv);
done_queue:
		if (ISMEMSYNCWRITE(hio)) {
			if (!hio->hio_memsyncacked) {
				PJDLOG_ASSERT(memsyncack ||
				    hio->hio_errors[ncomp] != 0);
				if (refcnt_release(&hio->hio_writecount) == 0) {
					if (hio->hio_errors[0] == 0)
						write_complete(res, hio);
				}
				hio->hio_memsyncacked = true;
				if (hio->hio_errors[ncomp] == 0) {
					pjdlog_debug(2,
					    "remote_recv: (%p) Moving request "
					    "back to the recv queue.", hio);
					mtx_lock(&hio_recv_list_lock[ncomp]);
					TAILQ_INSERT_TAIL(&hio_recv_list[ncomp],
					    hio, hio_next[ncomp]);
					hio_recv_list_size[ncomp]++;
					mtx_unlock(&hio_recv_list_lock[ncomp]);
					continue;
				}
			} else {
				PJDLOG_ASSERT(!memsyncack);
			}
		}
		if (refcnt_release(&hio->hio_countdown) > 0)
			continue;
		if (ISSYNCREQ(hio)) {
			mtx_lock(&sync_lock);
			SYNCREQDONE(hio);
			mtx_unlock(&sync_lock);
			cv_signal(&sync_cond);
		} else {
			pjdlog_debug(2,
			    "remote_recv: (%p) Moving request to the done queue.",
			    hio);
			QUEUE_INSERT2(hio, done);
		}
	}
	return (NULL);
}
static void * remote_recv_thread ( void * arg ) { struct hast_resource * res = arg ; struct g_gate_ctl_io * ggio ; struct hio * hio ; struct nv * nv ; unsigned int ncomp ; uint64_t seq ; bool memsyncack ; int error ; ncomp = 1 ; for ( ; ; ) { mtx_lock ( & hio_recv_list_lock [ ncomp ] ) ; while ( TAILQ_EMPTY ( & hio_recv_list [ ncomp ] ) ) { pjdlog_debug ( 2 , " " ) ; cv_wait ( & hio_recv_list_cond [ ncomp ] , & hio_recv_list_lock [ ncomp ] ) ; } mtx_unlock ( & hio_recv_list_lock [ ncomp ] ) ; memsyncack = false ; rw_rlock ( & hio_remote_lock [ ncomp ] ) ; if ( ! ISCONNECTED ( res , ncomp ) ) { rw_unlock ( & hio_remote_lock [ ncomp ] ) ; mtx_lock ( & hio_recv_list_lock [ ncomp ] ) ; hio = TAILQ_FIRST ( & hio_recv_list [ ncomp ] ) ; PJDLOG_ASSERT ( hio != NULL ) ; TAILQ_REMOVE ( & hio_recv_list [ ncomp ] , hio , hio_next [ ncomp ] ) ; hio_recv_list_size [ ncomp ] -- ; mtx_unlock ( & hio_recv_list_lock [ ncomp ] ) ; hio -> hio_errors [ ncomp ] = ENOTCONN ; goto done_queue ; } if ( hast_proto_recv_hdr ( res -> hr_remotein , & nv ) == -1 ) { pjdlog_errno ( LOG_ERR , " " ) ; rw_unlock ( & hio_remote_lock [ ncomp ] ) ; remote_close ( res , ncomp ) ; continue ; } rw_unlock ( & hio_remote_lock [ ncomp ] ) ; seq = nv_get_uint64 ( nv , " " ) ; if ( seq == 0 ) { pjdlog_error ( " " ) ; nv_free ( nv ) ; continue ; } memsyncack = nv_exists ( nv , " " ) ; mtx_lock ( & hio_recv_list_lock [ ncomp ] ) ; TAILQ_FOREACH ( hio , & hio_recv_list [ ncomp ] , hio_next [ ncomp ] ) { if ( hio -> hio_ggio . gctl_seq == seq ) { TAILQ_REMOVE ( & hio_recv_list [ ncomp ] , hio , hio_next [ ncomp ] ) ; hio_recv_list_size [ ncomp ] -- ; break ; } } mtx_unlock ( & hio_recv_list_lock [ ncomp ] ) ; if ( hio == NULL ) { pjdlog_error ( " " , ( uintmax_t ) seq ) ; nv_free ( nv ) ; continue ; } ggio = & hio -> hio_ggio ; error = nv_get_int16 ( nv , " " ) ; if ( error != 0 ) { hio -> hio_errors [ ncomp ] = error ; reqlog ( LOG_WARNING , 0 , ggio , " " , strerror ( error ) ) ; nv_free ( nv ) ; goto done_queue ; } switch ( ggio -> gctl_cmd ) { case BIO_READ : rw_rlock ( & hio_remote_lock [ ncomp ] ) ; if ( ! ISCONNECTED ( res , ncomp ) ) { rw_unlock ( & hio_remote_lock [ ncomp ] ) ; nv_free ( nv ) ; goto done_queue ; } if ( hast_proto_recv_data ( res , res -> hr_remotein , nv , ggio -> gctl_data , ggio -> gctl_length ) == -1 ) { hio -> hio_errors [ ncomp ] = errno ; pjdlog_errno ( LOG_ERR , " " ) ; rw_unlock ( & hio_remote_lock [ ncomp ] ) ; nv_free ( nv ) ; remote_close ( res , ncomp ) ; goto done_queue ; } rw_unlock ( & hio_remote_lock [ ncomp ] ) ; break ; case BIO_WRITE : case BIO_DELETE : case BIO_FLUSH : break ; default : PJDLOG_ABORT ( " " ) ; } hio -> hio_errors [ ncomp ] = 0 ; nv_free ( nv ) ; done_queue : if ( ISMEMSYNCWRITE ( hio ) ) { if ( ! hio -> hio_memsyncacked ) { PJDLOG_ASSERT ( memsyncack || hio -> hio_errors [ ncomp ] != 0 ) ; if ( refcnt_release ( & hio -> hio_writecount ) == 0 ) { if ( hio -> hio_errors [ 0 ] == 0 ) write_complete ( res , hio ) ; } hio -> hio_memsyncacked = true ; if ( hio -> hio_errors [ ncomp ] == 0 ) { pjdlog_debug ( 2 , " " " " , hio ) ; mtx_lock ( & hio_recv_list_lock [ ncomp ] ) ; TAILQ_INSERT_TAIL ( & hio_recv_list [ ncomp ] , hio , hio_next [ ncomp ] ) ; hio_recv_list_size [ ncomp ] ++ ; mtx_unlock ( & hio_recv_list_lock [ ncomp ] ) ; continue ; } } else { PJDLOG_ASSERT ( ! memsyncack ) ; } } if ( refcnt_release ( & hio -> hio_countdown ) > 0 ) continue ; if ( ISSYNCREQ ( hio ) ) { mtx_lock ( & sync_lock ) ; SYNCREQDONE ( hio ) ; mtx_unlock ( & sync_lock ) ; cv_signal ( & sync_cond ) ; } else { pjdlog_debug ( 2 , " " , hio ) ; QUEUE_INSERT2 ( hio , done ) ; } } return ( NULL ) ; }
Thread receives answer from secondary node and passes it to ggate_send
 thread.
Thread receives answer from secondary node and passes it to ggate_send thread .
/* Remote component is 1 for now. */ /* Wait until there is anything to receive. */ /*
			 * Connection is dead, so move all pending requests to
			 * the done queue (one-by-one).
			 */ /* Request failed on remote side. */ /* Remote ack arrived. */ /* Remote final reply arrived. */ /* NOTREACHED */