secondary.c revision 211877
1/*-
2 * Copyright (c) 2009-2010 The FreeBSD Foundation
3 * Copyright (c) 2010 Pawel Jakub Dawidek <pjd@FreeBSD.org>
4 * All rights reserved.
5 *
6 * This software was developed by Pawel Jakub Dawidek under sponsorship from
7 * the FreeBSD Foundation.
8 *
9 * Redistribution and use in source and binary forms, with or without
10 * modification, are permitted provided that the following conditions
11 * are met:
12 * 1. Redistributions of source code must retain the above copyright
13 *    notice, this list of conditions and the following disclaimer.
14 * 2. Redistributions in binary form must reproduce the above copyright
15 *    notice, this list of conditions and the following disclaimer in the
16 *    documentation and/or other materials provided with the distribution.
17 *
18 * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
19 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
22 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
24 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
27 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
28 * SUCH DAMAGE.
29 */
30
31#include <sys/cdefs.h>
32__FBSDID("$FreeBSD: head/sbin/hastd/secondary.c 211877 2010-08-27 14:01:28Z pjd $");
33
34#include <sys/param.h>
35#include <sys/time.h>
36#include <sys/bio.h>
37#include <sys/disk.h>
38#include <sys/stat.h>
39
40#include <assert.h>
41#include <err.h>
42#include <errno.h>
43#include <fcntl.h>
44#include <libgeom.h>
45#include <pthread.h>
46#include <stdint.h>
47#include <stdio.h>
48#include <string.h>
49#include <sysexits.h>
50#include <unistd.h>
51
52#include <activemap.h>
53#include <nv.h>
54#include <pjdlog.h>
55
56#include "control.h"
57#include "hast.h"
58#include "hast_proto.h"
59#include "hastd.h"
60#include "metadata.h"
61#include "proto.h"
62#include "subr.h"
63#include "synch.h"
64
65struct hio {
66	uint64_t 	 hio_seq;
67	int	 	 hio_error;
68	struct nv	*hio_nv;
69	void		*hio_data;
70	uint8_t		 hio_cmd;
71	uint64_t	 hio_offset;
72	uint64_t	 hio_length;
73	TAILQ_ENTRY(hio) hio_next;
74};
75
76/*
77 * Free list holds unused structures. When free list is empty, we have to wait
78 * until some in-progress requests are freed.
79 */
80static TAILQ_HEAD(, hio) hio_free_list;
81static pthread_mutex_t hio_free_list_lock;
82static pthread_cond_t hio_free_list_cond;
83/*
84 * Disk thread (the one that do I/O requests) takes requests from this list.
85 */
86static TAILQ_HEAD(, hio) hio_disk_list;
87static pthread_mutex_t hio_disk_list_lock;
88static pthread_cond_t hio_disk_list_cond;
89/*
90 * There is one recv list for every component, although local components don't
91 * use recv lists as local requests are done synchronously.
92 */
93static TAILQ_HEAD(, hio) hio_send_list;
94static pthread_mutex_t hio_send_list_lock;
95static pthread_cond_t hio_send_list_cond;
96
97/*
98 * Maximum number of outstanding I/O requests.
99 */
100#define	HAST_HIO_MAX	256
101
102static void *recv_thread(void *arg);
103static void *disk_thread(void *arg);
104static void *send_thread(void *arg);
105
106#define	QUEUE_INSERT(name, hio)	do {					\
107	bool _wakeup;							\
108									\
109	mtx_lock(&hio_##name##_list_lock);				\
110	_wakeup = TAILQ_EMPTY(&hio_##name##_list);			\
111	TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_next);		\
112	mtx_unlock(&hio_##name##_list_lock);				\
113	if (_wakeup)							\
114		cv_signal(&hio_##name##_list_cond);			\
115} while (0)
116#define	QUEUE_TAKE(name, hio)	do {					\
117	mtx_lock(&hio_##name##_list_lock);				\
118	while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) {	\
119		cv_wait(&hio_##name##_list_cond,			\
120		    &hio_##name##_list_lock);				\
121	}								\
122	TAILQ_REMOVE(&hio_##name##_list, (hio), hio_next);		\
123	mtx_unlock(&hio_##name##_list_lock);				\
124} while (0)
125
126static void
127init_environment(void)
128{
129	struct hio *hio;
130	unsigned int ii;
131
132	/*
133	 * Initialize lists, their locks and theirs condition variables.
134	 */
135	TAILQ_INIT(&hio_free_list);
136	mtx_init(&hio_free_list_lock);
137	cv_init(&hio_free_list_cond);
138	TAILQ_INIT(&hio_disk_list);
139	mtx_init(&hio_disk_list_lock);
140	cv_init(&hio_disk_list_cond);
141	TAILQ_INIT(&hio_send_list);
142	mtx_init(&hio_send_list_lock);
143	cv_init(&hio_send_list_cond);
144
145	/*
146	 * Allocate requests pool and initialize requests.
147	 */
148	for (ii = 0; ii < HAST_HIO_MAX; ii++) {
149		hio = malloc(sizeof(*hio));
150		if (hio == NULL) {
151			pjdlog_exitx(EX_TEMPFAIL,
152			    "Unable to allocate memory (%zu bytes) for hio request.",
153			    sizeof(*hio));
154		}
155		hio->hio_error = 0;
156		hio->hio_data = malloc(MAXPHYS);
157		if (hio->hio_data == NULL) {
158			pjdlog_exitx(EX_TEMPFAIL,
159			    "Unable to allocate memory (%zu bytes) for gctl_data.",
160			    (size_t)MAXPHYS);
161		}
162		TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_next);
163	}
164}
165
166static void
167init_local(struct hast_resource *res)
168{
169
170	if (metadata_read(res, true) < 0)
171		exit(EX_NOINPUT);
172}
173
174static void
175init_remote(struct hast_resource *res, struct nv *nvin)
176{
177	uint64_t resuid;
178	struct nv *nvout;
179	unsigned char *map;
180	size_t mapsize;
181
182	map = NULL;
183	mapsize = 0;
184	nvout = nv_alloc();
185	nv_add_int64(nvout, (int64_t)res->hr_datasize, "datasize");
186	nv_add_int32(nvout, (int32_t)res->hr_extentsize, "extentsize");
187	resuid = nv_get_uint64(nvin, "resuid");
188	res->hr_primary_localcnt = nv_get_uint64(nvin, "localcnt");
189	res->hr_primary_remotecnt = nv_get_uint64(nvin, "remotecnt");
190	nv_add_uint64(nvout, res->hr_secondary_localcnt, "localcnt");
191	nv_add_uint64(nvout, res->hr_secondary_remotecnt, "remotecnt");
192	mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
193	    METADATA_SIZE, res->hr_extentsize, res->hr_local_sectorsize);
194	map = malloc(mapsize);
195	if (map == NULL) {
196		pjdlog_exitx(EX_TEMPFAIL,
197		    "Unable to allocate memory (%zu bytes) for activemap.",
198		    mapsize);
199	}
200	nv_add_uint32(nvout, (uint32_t)mapsize, "mapsize");
201	/*
202	 * When we work as primary and secondary is missing we will increase
203	 * localcnt in our metadata. When secondary is connected and synced
204	 * we make localcnt be equal to remotecnt, which means nodes are more
205	 * or less in sync.
206	 * Split-brain condition is when both nodes are not able to communicate
207	 * and are both configured as primary nodes. In turn, they can both
208	 * make incompatible changes to the data and we have to detect that.
209	 * Under split-brain condition we will increase our localcnt on first
210	 * write and remote node will increase its localcnt on first write.
211	 * When we connect we can see that primary's localcnt is greater than
212	 * our remotecnt (primary was modified while we weren't watching) and
213	 * our localcnt is greater than primary's remotecnt (we were modified
214	 * while primary wasn't watching).
215	 * There are many possible combinations which are all gathered below.
216	 * Don't pay too much attention to exact numbers, the more important
217	 * is to compare them. We compare secondary's local with primary's
218	 * remote and secondary's remote with primary's local.
219	 * Note that every case where primary's localcnt is smaller than
220	 * secondary's remotecnt and where secondary's localcnt is smaller than
221	 * primary's remotecnt should be impossible in practise. We will perform
222	 * full synchronization then. Those cases are marked with an asterisk.
223	 * Regular synchronization means that only extents marked as dirty are
224	 * synchronized (regular synchronization).
225	 *
226	 * SECONDARY METADATA PRIMARY METADATA
227	 * local=3 remote=3   local=2 remote=2*  ?! Full sync from secondary.
228	 * local=3 remote=3   local=2 remote=3*  ?! Full sync from primary.
229	 * local=3 remote=3   local=2 remote=4*  ?! Full sync from primary.
230	 * local=3 remote=3   local=3 remote=2   Primary is out-of-date,
231	 *                                       regular sync from secondary.
232	 * local=3 remote=3   local=3 remote=3   Regular sync just in case.
233	 * local=3 remote=3   local=3 remote=4*  ?! Full sync from primary.
234	 * local=3 remote=3   local=4 remote=2   Split-brain condition.
235	 * local=3 remote=3   local=4 remote=3   Secondary out-of-date,
236	 *                                       regular sync from primary.
237	 * local=3 remote=3   local=4 remote=4*  ?! Full sync from primary.
238	 */
239	if (res->hr_resuid == 0) {
240		/*
241		 * Provider is used for the first time. Initialize everything.
242		 */
243		assert(res->hr_secondary_localcnt == 0);
244		res->hr_resuid = resuid;
245		if (metadata_write(res) < 0)
246			exit(EX_NOINPUT);
247		memset(map, 0xff, mapsize);
248		nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
249	} else if (
250	    /* Is primary is out-of-date? */
251	    (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
252	     res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
253	    /* Node are more or less in sync? */
254	    (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
255	     res->hr_secondary_remotecnt == res->hr_primary_localcnt) ||
256	    /* Is secondary is out-of-date? */
257	    (res->hr_secondary_localcnt == res->hr_primary_remotecnt &&
258	     res->hr_secondary_remotecnt < res->hr_primary_localcnt)) {
259		/*
260		 * Nodes are more or less in sync or one of the nodes is
261		 * out-of-date.
262		 * It doesn't matter at this point which one, we just have to
263		 * send out local bitmap to the remote node.
264		 */
265		if (pread(res->hr_localfd, map, mapsize, METADATA_SIZE) !=
266		    (ssize_t)mapsize) {
267			pjdlog_exit(LOG_ERR, "Unable to read activemap");
268		}
269		if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
270		     res->hr_secondary_remotecnt == res->hr_primary_localcnt) {
271			/* Primary is out-of-date, sync from secondary. */
272			nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
273		} else {
274			/*
275			 * Secondary is out-of-date or counts match.
276			 * Sync from primary.
277			 */
278			nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
279		}
280	} else if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
281	     res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
282		/*
283		 * Not good, we have split-brain condition.
284		 */
285		pjdlog_error("Split-brain detected, exiting.");
286		nv_add_string(nvout, "Split-brain condition!", "errmsg");
287		free(map);
288		map = NULL;
289		mapsize = 0;
290	} else /* if (res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
291	    res->hr_primary_localcnt < res->hr_secondary_remotecnt) */ {
292		/*
293		 * This should never happen in practise, but we will perform
294		 * full synchronization.
295		 */
296		assert(res->hr_secondary_localcnt < res->hr_primary_remotecnt ||
297		    res->hr_primary_localcnt < res->hr_secondary_remotecnt);
298		mapsize = activemap_calc_ondisk_size(res->hr_local_mediasize -
299		    METADATA_SIZE, res->hr_extentsize,
300		    res->hr_local_sectorsize);
301		memset(map, 0xff, mapsize);
302		if (res->hr_secondary_localcnt > res->hr_primary_remotecnt) {
303			/* In this one of five cases sync from secondary. */
304			nv_add_uint8(nvout, HAST_SYNCSRC_SECONDARY, "syncsrc");
305		} else {
306			/* For the rest four cases sync from primary. */
307			nv_add_uint8(nvout, HAST_SYNCSRC_PRIMARY, "syncsrc");
308		}
309		pjdlog_warning("This should never happen, asking for full synchronization (primary(local=%ju, remote=%ju), secondary(local=%ju, remote=%ju)).",
310		    (uintmax_t)res->hr_primary_localcnt,
311		    (uintmax_t)res->hr_primary_remotecnt,
312		    (uintmax_t)res->hr_secondary_localcnt,
313		    (uintmax_t)res->hr_secondary_remotecnt);
314	}
315	if (hast_proto_send(res, res->hr_remotein, nvout, map, mapsize) < 0) {
316		pjdlog_errno(LOG_WARNING, "Unable to send activemap to %s",
317		    res->hr_remoteaddr);
318		nv_free(nvout);
319		exit(EX_TEMPFAIL);
320	}
321	nv_free(nvout);
322	if (res->hr_secondary_localcnt > res->hr_primary_remotecnt &&
323	     res->hr_primary_localcnt > res->hr_secondary_remotecnt) {
324		/* Exit on split-brain. */
325		exit(EX_CONFIG);
326	}
327}
328
329void
330hastd_secondary(struct hast_resource *res, struct nv *nvin)
331{
332	pthread_t td;
333	pid_t pid;
334	int error;
335
336	/*
337	 * Create communication channel between parent and child.
338	 */
339	if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
340		KEEP_ERRNO((void)pidfile_remove(pfh));
341		pjdlog_exit(EX_OSERR,
342		    "Unable to create control sockets between parent and child");
343	}
344
345	pid = fork();
346	if (pid < 0) {
347		KEEP_ERRNO((void)pidfile_remove(pfh));
348		pjdlog_exit(EX_OSERR, "Unable to fork");
349	}
350
351	if (pid > 0) {
352		/* This is parent. */
353		proto_close(res->hr_remotein);
354		res->hr_remotein = NULL;
355		proto_close(res->hr_remoteout);
356		res->hr_remoteout = NULL;
357		res->hr_workerpid = pid;
358		return;
359	}
360	(void)pidfile_close(pfh);
361
362	setproctitle("%s (secondary)", res->hr_name);
363
364	signal(SIGHUP, SIG_DFL);
365	signal(SIGCHLD, SIG_DFL);
366
367	/* Error in setting timeout is not critical, but why should it fail? */
368	if (proto_timeout(res->hr_remotein, 0) < 0)
369		pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
370	if (proto_timeout(res->hr_remoteout, res->hr_timeout) < 0)
371		pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
372
373	init_local(res);
374	init_remote(res, nvin);
375	init_environment();
376
377	error = pthread_create(&td, NULL, recv_thread, res);
378	assert(error == 0);
379	error = pthread_create(&td, NULL, disk_thread, res);
380	assert(error == 0);
381	error = pthread_create(&td, NULL, send_thread, res);
382	assert(error == 0);
383	(void)ctrl_thread(res);
384}
385
386static void
387reqlog(int loglevel, int debuglevel, int error, struct hio *hio, const char *fmt, ...)
388{
389	char msg[1024];
390	va_list ap;
391	int len;
392
393	va_start(ap, fmt);
394	len = vsnprintf(msg, sizeof(msg), fmt, ap);
395	va_end(ap);
396	if ((size_t)len < sizeof(msg)) {
397		switch (hio->hio_cmd) {
398		case HIO_READ:
399			(void)snprintf(msg + len, sizeof(msg) - len,
400			    "READ(%ju, %ju).", (uintmax_t)hio->hio_offset,
401			    (uintmax_t)hio->hio_length);
402			break;
403		case HIO_DELETE:
404			(void)snprintf(msg + len, sizeof(msg) - len,
405			    "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset,
406			    (uintmax_t)hio->hio_length);
407			break;
408		case HIO_FLUSH:
409			(void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
410			break;
411		case HIO_WRITE:
412			(void)snprintf(msg + len, sizeof(msg) - len,
413			    "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset,
414			    (uintmax_t)hio->hio_length);
415			break;
416		default:
417			(void)snprintf(msg + len, sizeof(msg) - len,
418			    "UNKNOWN(%u).", (unsigned int)hio->hio_cmd);
419			break;
420		}
421	}
422	pjdlog_common(loglevel, debuglevel, error, "%s", msg);
423}
424
425static int
426requnpack(struct hast_resource *res, struct hio *hio)
427{
428
429	hio->hio_cmd = nv_get_uint8(hio->hio_nv, "cmd");
430	if (hio->hio_cmd == 0) {
431		pjdlog_error("Header contains no 'cmd' field.");
432		hio->hio_error = EINVAL;
433		goto end;
434	}
435	switch (hio->hio_cmd) {
436	case HIO_READ:
437	case HIO_WRITE:
438	case HIO_DELETE:
439		hio->hio_offset = nv_get_uint64(hio->hio_nv, "offset");
440		if (nv_error(hio->hio_nv) != 0) {
441			pjdlog_error("Header is missing 'offset' field.");
442			hio->hio_error = EINVAL;
443			goto end;
444		}
445		hio->hio_length = nv_get_uint64(hio->hio_nv, "length");
446		if (nv_error(hio->hio_nv) != 0) {
447			pjdlog_error("Header is missing 'length' field.");
448			hio->hio_error = EINVAL;
449			goto end;
450		}
451		if (hio->hio_length == 0) {
452			pjdlog_error("Data length is zero.");
453			hio->hio_error = EINVAL;
454			goto end;
455		}
456		if (hio->hio_length > MAXPHYS) {
457			pjdlog_error("Data length is too large (%ju > %ju).",
458			    (uintmax_t)hio->hio_length, (uintmax_t)MAXPHYS);
459			hio->hio_error = EINVAL;
460			goto end;
461		}
462		if ((hio->hio_offset % res->hr_local_sectorsize) != 0) {
463			pjdlog_error("Offset %ju is not multiple of sector size.",
464			    (uintmax_t)hio->hio_offset);
465			hio->hio_error = EINVAL;
466			goto end;
467		}
468		if ((hio->hio_length % res->hr_local_sectorsize) != 0) {
469			pjdlog_error("Length %ju is not multiple of sector size.",
470			    (uintmax_t)hio->hio_length);
471			hio->hio_error = EINVAL;
472			goto end;
473		}
474		if (hio->hio_offset + hio->hio_length >
475		    (uint64_t)res->hr_datasize) {
476			pjdlog_error("Data offset is too large (%ju > %ju).",
477			    (uintmax_t)(hio->hio_offset + hio->hio_length),
478			    (uintmax_t)res->hr_datasize);
479			hio->hio_error = EINVAL;
480			goto end;
481		}
482		break;
483	default:
484		pjdlog_error("Header contains invalid 'cmd' (%hhu).",
485		    hio->hio_cmd);
486		hio->hio_error = EINVAL;
487		goto end;
488	}
489	hio->hio_error = 0;
490end:
491	return (hio->hio_error);
492}
493
494/*
495 * Thread receives requests from the primary node.
496 */
497static void *
498recv_thread(void *arg)
499{
500	struct hast_resource *res = arg;
501	struct hio *hio;
502
503	for (;;) {
504		pjdlog_debug(2, "recv: Taking free request.");
505		QUEUE_TAKE(free, hio);
506		pjdlog_debug(2, "recv: (%p) Got request.", hio);
507		if (hast_proto_recv_hdr(res->hr_remotein, &hio->hio_nv) < 0) {
508			pjdlog_exit(EX_TEMPFAIL,
509			    "Unable to receive request header");
510		}
511		if (requnpack(res, hio) != 0) {
512			pjdlog_debug(2,
513			    "recv: (%p) Moving request to the send queue.",
514			    hio);
515			QUEUE_INSERT(send, hio);
516			continue;
517		}
518		reqlog(LOG_DEBUG, 2, -1, hio,
519		    "recv: (%p) Got request header: ", hio);
520		if (hio->hio_cmd == HIO_WRITE) {
521			if (hast_proto_recv_data(res, res->hr_remotein,
522			    hio->hio_nv, hio->hio_data, MAXPHYS) < 0) {
523				pjdlog_exit(EX_TEMPFAIL,
524				    "Unable to receive reply data");
525			}
526		}
527		pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.",
528		    hio);
529		QUEUE_INSERT(disk, hio);
530	}
531	/* NOTREACHED */
532	return (NULL);
533}
534
535/*
536 * Thread reads from or writes to local component and also handles DELETE and
537 * FLUSH requests.
538 */
539static void *
540disk_thread(void *arg)
541{
542	struct hast_resource *res = arg;
543	struct hio *hio;
544	ssize_t ret;
545	bool clear_activemap;
546
547	clear_activemap = true;
548
549	for (;;) {
550		pjdlog_debug(2, "disk: Taking request.");
551		QUEUE_TAKE(disk, hio);
552		while (clear_activemap) {
553			unsigned char *map;
554			size_t mapsize;
555
556			/*
557			 * When first request is received, it means that primary
558			 * already received our activemap, merged it and stored
559			 * locally. We can now safely clear our activemap.
560			 */
561			mapsize =
562			    activemap_calc_ondisk_size(res->hr_local_mediasize -
563			    METADATA_SIZE, res->hr_extentsize,
564			    res->hr_local_sectorsize);
565			map = calloc(1, mapsize);
566			if (map == NULL) {
567				pjdlog_warning("Unable to allocate memory to clear local activemap.");
568				break;
569			}
570			if (pwrite(res->hr_localfd, map, mapsize,
571			    METADATA_SIZE) != (ssize_t)mapsize) {
572				pjdlog_errno(LOG_WARNING,
573				    "Unable to store cleared activemap");
574				free(map);
575				break;
576			}
577			free(map);
578			clear_activemap = false;
579			pjdlog_debug(1, "Local activemap cleared.");
580		}
581		reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio);
582		/* Handle the actual request. */
583		switch (hio->hio_cmd) {
584		case HIO_READ:
585			ret = pread(res->hr_localfd, hio->hio_data,
586			    hio->hio_length,
587			    hio->hio_offset + res->hr_localoff);
588			if (ret < 0)
589				hio->hio_error = errno;
590			else if (ret != (int64_t)hio->hio_length)
591				hio->hio_error = EIO;
592			else
593				hio->hio_error = 0;
594			break;
595		case HIO_WRITE:
596			ret = pwrite(res->hr_localfd, hio->hio_data,
597			    hio->hio_length,
598			    hio->hio_offset + res->hr_localoff);
599			if (ret < 0)
600				hio->hio_error = errno;
601			else if (ret != (int64_t)hio->hio_length)
602				hio->hio_error = EIO;
603			else
604				hio->hio_error = 0;
605			break;
606		case HIO_DELETE:
607			ret = g_delete(res->hr_localfd,
608			    hio->hio_offset + res->hr_localoff,
609			    hio->hio_length);
610			if (ret < 0)
611				hio->hio_error = errno;
612			else
613				hio->hio_error = 0;
614			break;
615		case HIO_FLUSH:
616			ret = g_flush(res->hr_localfd);
617			if (ret < 0)
618				hio->hio_error = errno;
619			else
620				hio->hio_error = 0;
621			break;
622		}
623		if (hio->hio_error != 0) {
624			reqlog(LOG_ERR, 0, hio->hio_error, hio,
625			    "Request failed: ");
626		}
627		pjdlog_debug(2, "disk: (%p) Moving request to the send queue.",
628		    hio);
629		QUEUE_INSERT(send, hio);
630	}
631	/* NOTREACHED */
632	return (NULL);
633}
634
635/*
636 * Thread sends requests back to primary node.
637 */
638static void *
639send_thread(void *arg)
640{
641	struct hast_resource *res = arg;
642	struct nv *nvout;
643	struct hio *hio;
644	void *data;
645	size_t length;
646
647	for (;;) {
648		pjdlog_debug(2, "send: Taking request.");
649		QUEUE_TAKE(send, hio);
650		reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio);
651		nvout = nv_alloc();
652		/* Copy sequence number. */
653		nv_add_uint64(nvout, nv_get_uint64(hio->hio_nv, "seq"), "seq");
654		switch (hio->hio_cmd) {
655		case HIO_READ:
656			if (hio->hio_error == 0) {
657				data = hio->hio_data;
658				length = hio->hio_length;
659				break;
660			}
661			/*
662			 * We send no data in case of an error.
663			 */
664			/* FALLTHROUGH */
665		case HIO_DELETE:
666		case HIO_FLUSH:
667		case HIO_WRITE:
668			data = NULL;
669			length = 0;
670			break;
671		default:
672			abort();
673			break;
674		}
675		if (hio->hio_error != 0)
676			nv_add_int16(nvout, hio->hio_error, "error");
677		if (hast_proto_send(res, res->hr_remoteout, nvout, data,
678		    length) < 0) {
679			pjdlog_exit(EX_TEMPFAIL, "Unable to send reply.");
680		}
681		nv_free(nvout);
682		pjdlog_debug(2, "send: (%p) Moving request to the free queue.",
683		    hio);
684		nv_free(hio->hio_nv);
685		hio->hio_error = 0;
686		QUEUE_INSERT(free, hio);
687	}
688	/* NOTREACHED */
689	return (NULL);
690}
691