FreeBSD/sbin/hastd/primary.c
/*
 * Thread reads from or writes to local component.
 * If local read fails, it redirects it to remote_send thread.
 */
Thread reads from or writes to local component.
If local read fails, it redirects it to remote_send thread.
Thread reads from or writes to local component . If local read fails it redirects it to remote_send thread .
static void *
local_send_thread(void *arg)
{
	struct hast_resource *res = arg;
	struct g_gate_ctl_io *ggio;
	struct hio *hio;
	unsigned int ncomp, rncomp;
	ssize_t ret;
	ncomp = 0;
	rncomp = 1;
	for (;;) {
		pjdlog_debug(2, "local_send: Taking request.");
		QUEUE_TAKE1(hio, send, ncomp, 0);
		pjdlog_debug(2, "local_send: (%p) Got request.", hio);
		ggio = &hio->hio_ggio;
		switch (ggio->gctl_cmd) {
		case BIO_READ:
			ret = pread(res->hr_localfd, ggio->gctl_data,
			    ggio->gctl_length,
			    ggio->gctl_offset + res->hr_localoff);
			if (ret == ggio->gctl_length)
				hio->hio_errors[ncomp] = 0;
			else if (!ISSYNCREQ(hio)) {
				if (ret == -1) {
					reqlog(LOG_WARNING, 0, ggio,
					    "Local request failed (%s), trying remote node. ",
					    strerror(errno));
				} else if (ret != ggio->gctl_length) {
					reqlog(LOG_WARNING, 0, ggio,
					    "Local request failed (%zd != %jd), trying remote node. ",
					    ret, (intmax_t)ggio->gctl_length);
				}
				QUEUE_INSERT1(hio, send, rncomp);
				continue;
			}
			break;
		case BIO_WRITE:
			ret = pwrite(res->hr_localfd, ggio->gctl_data,
			    ggio->gctl_length,
			    ggio->gctl_offset + res->hr_localoff);
			if (ret == -1) {
				hio->hio_errors[ncomp] = errno;
				reqlog(LOG_WARNING, 0, ggio,
				    "Local request failed (%s): ",
				    strerror(errno));
			} else if (ret != ggio->gctl_length) {
				hio->hio_errors[ncomp] = EIO;
				reqlog(LOG_WARNING, 0, ggio,
				    "Local request failed (%zd != %jd): ",
				    ret, (intmax_t)ggio->gctl_length);
			} else {
				hio->hio_errors[ncomp] = 0;
				if (ISASYNC(hio)) {
					ggio->gctl_error = 0;
					write_complete(res, hio);
				}
			}
			break;
		case BIO_DELETE:
			ret = g_delete(res->hr_localfd,
			    ggio->gctl_offset + res->hr_localoff,
			    ggio->gctl_length);
			if (ret == -1) {
				hio->hio_errors[ncomp] = errno;
				reqlog(LOG_WARNING, 0, ggio,
				    "Local request failed (%s): ",
				    strerror(errno));
			} else {
				hio->hio_errors[ncomp] = 0;
			}
			break;
		case BIO_FLUSH:
			if (!res->hr_localflush) {
				ret = -1;
				errno = EOPNOTSUPP;
				break;
			}
			ret = g_flush(res->hr_localfd);
			if (ret == -1) {
				if (errno == EOPNOTSUPP)
					res->hr_localflush = false;
				hio->hio_errors[ncomp] = errno;
				reqlog(LOG_WARNING, 0, ggio,
				    "Local request failed (%s): ",
				    strerror(errno));
			} else {
				hio->hio_errors[ncomp] = 0;
			}
			break;
		}
		if (ISMEMSYNCWRITE(hio)) {
			if (refcnt_release(&hio->hio_writecount) == 0) {
				write_complete(res, hio);
			}
		}
		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,
			    "local_send: (%p) Moving request to the done queue.",
			    hio);
			QUEUE_INSERT2(hio, done);
		}
	}
	return (NULL);
}
static void * local_send_thread ( void * arg ) { struct hast_resource * res = arg ; struct g_gate_ctl_io * ggio ; struct hio * hio ; unsigned int ncomp , rncomp ; ssize_t ret ; ncomp = 0 ; rncomp = 1 ; for ( ; ; ) { pjdlog_debug ( 2 , " " ) ; QUEUE_TAKE1 ( hio , send , ncomp , 0 ) ; pjdlog_debug ( 2 , " " , hio ) ; ggio = & hio -> hio_ggio ; switch ( ggio -> gctl_cmd ) { case BIO_READ : ret = pread ( res -> hr_localfd , ggio -> gctl_data , ggio -> gctl_length , ggio -> gctl_offset + res -> hr_localoff ) ; if ( ret == ggio -> gctl_length ) hio -> hio_errors [ ncomp ] = 0 ; else if ( ! ISSYNCREQ ( hio ) ) { if ( ret == -1 ) { reqlog ( LOG_WARNING , 0 , ggio , " " , strerror ( errno ) ) ; } else if ( ret != ggio -> gctl_length ) { reqlog ( LOG_WARNING , 0 , ggio , " " , ret , ( intmax_t ) ggio -> gctl_length ) ; } QUEUE_INSERT1 ( hio , send , rncomp ) ; continue ; } break ; case BIO_WRITE : ret = pwrite ( res -> hr_localfd , ggio -> gctl_data , ggio -> gctl_length , ggio -> gctl_offset + res -> hr_localoff ) ; if ( ret == -1 ) { hio -> hio_errors [ ncomp ] = errno ; reqlog ( LOG_WARNING , 0 , ggio , " " , strerror ( errno ) ) ; } else if ( ret != ggio -> gctl_length ) { hio -> hio_errors [ ncomp ] = EIO ; reqlog ( LOG_WARNING , 0 , ggio , " " , ret , ( intmax_t ) ggio -> gctl_length ) ; } else { hio -> hio_errors [ ncomp ] = 0 ; if ( ISASYNC ( hio ) ) { ggio -> gctl_error = 0 ; write_complete ( res , hio ) ; } } break ; case BIO_DELETE : ret = g_delete ( res -> hr_localfd , ggio -> gctl_offset + res -> hr_localoff , ggio -> gctl_length ) ; if ( ret == -1 ) { hio -> hio_errors [ ncomp ] = errno ; reqlog ( LOG_WARNING , 0 , ggio , " " , strerror ( errno ) ) ; } else { hio -> hio_errors [ ncomp ] = 0 ; } break ; case BIO_FLUSH : if ( ! res -> hr_localflush ) { ret = -1 ; errno = EOPNOTSUPP ; break ; } ret = g_flush ( res -> hr_localfd ) ; if ( ret == -1 ) { if ( errno == EOPNOTSUPP ) res -> hr_localflush = false ; hio -> hio_errors [ ncomp ] = errno ; reqlog ( LOG_WARNING , 0 , ggio , " " , strerror ( errno ) ) ; } else { hio -> hio_errors [ ncomp ] = 0 ; } break ; } if ( ISMEMSYNCWRITE ( hio ) ) { if ( refcnt_release ( & hio -> hio_writecount ) == 0 ) { write_complete ( res , hio ) ; } } 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 reads from or writes to local component.
Thread reads from or writes to local component .
/* Local component is 0 for now. */ /* Remote component is 1 for now. */ /*
				 * If READ failed, try to read from remote node.
				 */ /* NOTREACHED */