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