FreeBSD/sbin/hastd/primary.c
/*
 * Thread sends request to secondary node.
 */
Thread sends request to secondary node.
Thread sends request to secondary node .
static void *
remote_send_thread(void *arg)
{
	struct hast_resource *res = arg;
	struct g_gate_ctl_io *ggio;
	time_t lastcheck, now;
	struct hio *hio;
	struct nv *nv;
	unsigned int ncomp;
	bool wakeup;
	uint64_t offset, length;
	uint8_t cmd;
	void *data;
	ncomp = 1;
	lastcheck = time(NULL);
	for (;;) {
		pjdlog_debug(2, "remote_send: Taking request.");
		QUEUE_TAKE1(hio, send, ncomp, HAST_KEEPALIVE);
		if (hio == NULL) {
			now = time(NULL);
			if (lastcheck + HAST_KEEPALIVE <= now) {
				keepalive_send(res, ncomp);
				lastcheck = now;
			}
			continue;
		}
		pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
		ggio = &hio->hio_ggio;
		switch (ggio->gctl_cmd) {
		case BIO_READ:
			cmd = HIO_READ;
			data = NULL;
			offset = ggio->gctl_offset;
			length = ggio->gctl_length;
			break;
		case BIO_WRITE:
			cmd = HIO_WRITE;
			data = ggio->gctl_data;
			offset = ggio->gctl_offset;
			length = ggio->gctl_length;
			break;
		case BIO_DELETE:
			cmd = HIO_DELETE;
			data = NULL;
			offset = ggio->gctl_offset;
			length = ggio->gctl_length;
			break;
		case BIO_FLUSH:
			cmd = HIO_FLUSH;
			data = NULL;
			offset = 0;
			length = 0;
			break;
		default:
			PJDLOG_ABORT("invalid condition");
		}
		nv = nv_alloc();
		nv_add_uint8(nv, cmd, "cmd");
		nv_add_uint64(nv, (uint64_t)ggio->gctl_seq, "seq");
		nv_add_uint64(nv, offset, "offset");
		nv_add_uint64(nv, length, "length");
		if (ISMEMSYNCWRITE(hio))
			nv_add_uint8(nv, 1, "memsync");
		if (nv_error(nv) != 0) {
			hio->hio_errors[ncomp] = nv_error(nv);
			pjdlog_debug(2,
			    "remote_send: (%p) Unable to prepare header to send.",
			    hio);
			reqlog(LOG_ERR, 0, ggio,
			    "Unable to prepare header to send (%s): ",
			    strerror(nv_error(nv)));
			goto done_queue;
		}
		rw_rlock(&hio_remote_lock[ncomp]);
		if (!ISCONNECTED(res, ncomp)) {
			rw_unlock(&hio_remote_lock[ncomp]);
			hio->hio_errors[ncomp] = ENOTCONN;
			goto done_queue;
		}
		pjdlog_debug(2,
		    "remote_send: (%p) Moving request to the recv queue.",
		    hio);
		mtx_lock(&hio_recv_list_lock[ncomp]);
		wakeup = TAILQ_EMPTY(&hio_recv_list[ncomp]);
		TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
		hio_recv_list_size[ncomp]++;
		mtx_unlock(&hio_recv_list_lock[ncomp]);
		if (hast_proto_send(res, res->hr_remoteout, nv, data,
		    data != NULL ? length : 0) == -1) {
			hio->hio_errors[ncomp] = errno;
			rw_unlock(&hio_remote_lock[ncomp]);
			pjdlog_debug(2,
			    "remote_send: (%p) Unable to send request.", hio);
			reqlog(LOG_ERR, 0, ggio,
			    "Unable to send request (%s): ",
			    strerror(hio->hio_errors[ncomp]));
			remote_close(res, ncomp);
		} else {
			rw_unlock(&hio_remote_lock[ncomp]);
		}
		nv_free(nv);
		if (wakeup)
			cv_signal(&hio_recv_list_cond[ncomp]);
		continue;
done_queue:
		nv_free(nv);
		if (ISSYNCREQ(hio)) {
			if (refcnt_release(&hio->hio_countdown) > 0)
				continue;
			mtx_lock(&sync_lock);
			SYNCREQDONE(hio);
			mtx_unlock(&sync_lock);
			cv_signal(&sync_cond);
			continue;
		}
		if (ggio->gctl_cmd == BIO_WRITE) {
			mtx_lock(&res->hr_amp_lock);
			if (activemap_need_sync(res->hr_amp, ggio->gctl_offset,
			    ggio->gctl_length)) {
				(void)hast_activemap_flush(res);
			} else {
				mtx_unlock(&res->hr_amp_lock);
			}
			if (ISMEMSYNCWRITE(hio)) {
				if (refcnt_release(&hio->hio_writecount) == 0) {
					if (hio->hio_errors[0] == 0)
						write_complete(res, hio);
				}
			}
		}
		if (refcnt_release(&hio->hio_countdown) > 0)
			continue;
		pjdlog_debug(2,
		    "remote_send: (%p) Moving request to the done queue.",
		    hio);
		QUEUE_INSERT2(hio, done);
	}
	return (NULL);
}
static void * remote_send_thread ( void * arg ) { struct hast_resource * res = arg ; struct g_gate_ctl_io * ggio ; time_t lastcheck , now ; struct hio * hio ; struct nv * nv ; unsigned int ncomp ; bool wakeup ; uint64_t offset , length ; uint8_t cmd ; void * data ; ncomp = 1 ; lastcheck = time ( NULL ) ; for ( ; ; ) { pjdlog_debug ( 2 , " " ) ; QUEUE_TAKE1 ( hio , send , ncomp , HAST_KEEPALIVE ) ; if ( hio == NULL ) { now = time ( NULL ) ; if ( lastcheck + HAST_KEEPALIVE <= now ) { keepalive_send ( res , ncomp ) ; lastcheck = now ; } continue ; } pjdlog_debug ( 2 , " " , hio ) ; ggio = & hio -> hio_ggio ; switch ( ggio -> gctl_cmd ) { case BIO_READ : cmd = HIO_READ ; data = NULL ; offset = ggio -> gctl_offset ; length = ggio -> gctl_length ; break ; case BIO_WRITE : cmd = HIO_WRITE ; data = ggio -> gctl_data ; offset = ggio -> gctl_offset ; length = ggio -> gctl_length ; break ; case BIO_DELETE : cmd = HIO_DELETE ; data = NULL ; offset = ggio -> gctl_offset ; length = ggio -> gctl_length ; break ; case BIO_FLUSH : cmd = HIO_FLUSH ; data = NULL ; offset = 0 ; length = 0 ; break ; default : PJDLOG_ABORT ( " " ) ; } nv = nv_alloc ( ) ; nv_add_uint8 ( nv , cmd , " " ) ; nv_add_uint64 ( nv , ( uint64_t ) ggio -> gctl_seq , " " ) ; nv_add_uint64 ( nv , offset , " " ) ; nv_add_uint64 ( nv , length , " " ) ; if ( ISMEMSYNCWRITE ( hio ) ) nv_add_uint8 ( nv , 1 , " " ) ; if ( nv_error ( nv ) != 0 ) { hio -> hio_errors [ ncomp ] = nv_error ( nv ) ; pjdlog_debug ( 2 , " " , hio ) ; reqlog ( LOG_ERR , 0 , ggio , " " , strerror ( nv_error ( nv ) ) ) ; goto done_queue ; } rw_rlock ( & hio_remote_lock [ ncomp ] ) ; if ( ! ISCONNECTED ( res , ncomp ) ) { rw_unlock ( & hio_remote_lock [ ncomp ] ) ; hio -> hio_errors [ ncomp ] = ENOTCONN ; goto done_queue ; } pjdlog_debug ( 2 , " " , hio ) ; mtx_lock ( & hio_recv_list_lock [ ncomp ] ) ; wakeup = TAILQ_EMPTY ( & hio_recv_list [ ncomp ] ) ; TAILQ_INSERT_TAIL ( & hio_recv_list [ ncomp ] , hio , hio_next [ ncomp ] ) ; hio_recv_list_size [ ncomp ] ++ ; mtx_unlock ( & hio_recv_list_lock [ ncomp ] ) ; if ( hast_proto_send ( res , res -> hr_remoteout , nv , data , data != NULL ? length : 0 ) == -1 ) { hio -> hio_errors [ ncomp ] = errno ; rw_unlock ( & hio_remote_lock [ ncomp ] ) ; pjdlog_debug ( 2 , " " , hio ) ; reqlog ( LOG_ERR , 0 , ggio , " " , strerror ( hio -> hio_errors [ ncomp ] ) ) ; remote_close ( res , ncomp ) ; } else { rw_unlock ( & hio_remote_lock [ ncomp ] ) ; } nv_free ( nv ) ; if ( wakeup ) cv_signal ( & hio_recv_list_cond [ ncomp ] ) ; continue ; done_queue : nv_free ( nv ) ; if ( ISSYNCREQ ( hio ) ) { if ( refcnt_release ( & hio -> hio_countdown ) > 0 ) continue ; mtx_lock ( & sync_lock ) ; SYNCREQDONE ( hio ) ; mtx_unlock ( & sync_lock ) ; cv_signal ( & sync_cond ) ; continue ; } if ( ggio -> gctl_cmd == BIO_WRITE ) { mtx_lock ( & res -> hr_amp_lock ) ; if ( activemap_need_sync ( res -> hr_amp , ggio -> gctl_offset , ggio -> gctl_length ) ) { ( void ) hast_activemap_flush ( res ) ; } else { mtx_unlock ( & res -> hr_amp_lock ) ; } if ( ISMEMSYNCWRITE ( hio ) ) { if ( refcnt_release ( & hio -> hio_writecount ) == 0 ) { if ( hio -> hio_errors [ 0 ] == 0 ) write_complete ( res , hio ) ; } } } if ( refcnt_release ( & hio -> hio_countdown ) > 0 ) continue ; pjdlog_debug ( 2 , " " , hio ) ; QUEUE_INSERT2 ( hio , done ) ; } return ( NULL ) ; }
Thread sends request to secondary node.
Thread sends request to secondary node .
/* Remote component is 1 for now. */ /* Move failed request immediately to the done queue. */ /*
		 * Protect connection from disappearing.
		 */ /*
		 * Move the request to recv queue before sending it, because
		 * in different order we can get reply before we move request
		 * to recv queue.
		 */ /* NOTREACHED */