sbin/hastd/secondary.c

   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         /* Error in setting timeout is not critical, but why should it fail? */
 341         if (proto_timeout(res->hr_remotein, 0) < 0)
 342                 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
 343         if (proto_timeout(res->hr_remoteout, res->hr_timeout) < 0)
 344                 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
 345
 346         init_local(res);
 347         init_remote(res, nvin);
 348         init_environment();
 349
 350         error = pthread_create(&td, NULL, recv_thread, res);
 351         assert(error == 0);
 352         error = pthread_create(&td, NULL, disk_thread, res);
 353         assert(error == 0);
 354         error = pthread_create(&td, NULL, send_thread, res);
 355         assert(error == 0);
 356         (void)ctrl_thread(res);
 357 }
 358
 359 static void
 360 reqlog(int loglevel, int debuglevel, int error, struct hio *hio, const char *fmt, ...)
 361 {
 362         char msg[1024];
 363         va_list ap;
 364         int len;
 365
 366         va_start(ap, fmt);
 367         len = vsnprintf(msg, sizeof(msg), fmt, ap);
 368         va_end(ap);
 369         if ((size_t)len < sizeof(msg)) {
 370                 switch (hio->hio_cmd) {
 371                 case HIO_READ:
 372                         (void)snprintf(msg + len, sizeof(msg) - len,
 373                             "READ(%ju, %ju).", (uintmax_t)hio->hio_offset,
 374                             (uintmax_t)hio->hio_length);
 375                         break;
 376                 case HIO_DELETE:
 377                         (void)snprintf(msg + len, sizeof(msg) - len,
 378                             "DELETE(%ju, %ju).", (uintmax_t)hio->hio_offset,
 379                             (uintmax_t)hio->hio_length);
 380                         break;
 381                 case HIO_FLUSH:
 382                         (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
 383                         break;
 384                 case HIO_WRITE:
 385                         (void)snprintf(msg + len, sizeof(msg) - len,
 386                             "WRITE(%ju, %ju).", (uintmax_t)hio->hio_offset,
 387                             (uintmax_t)hio->hio_length);
 388                         break;
 389                 default:
 390                         (void)snprintf(msg + len, sizeof(msg) - len,
 391                             "UNKNOWN(%u).", (unsigned int)hio->hio_cmd);
 392                         break;
 393                 }
 394         }
 395         pjdlog_common(loglevel, debuglevel, error, "%s", msg);
 396 }
 397
 398 static int
 399 requnpack(struct hast_resource *res, struct hio *hio)
 400 {
 401
 402         hio->hio_cmd = nv_get_uint8(hio->hio_nv, "cmd");
 403         if (hio->hio_cmd == 0) {
 404                 pjdlog_error("Header contains no 'cmd' field.");
 405                 hio->hio_error = EINVAL;
 406                 goto end;
 407         }
 408         switch (hio->hio_cmd) {
 409         case HIO_READ:
 410         case HIO_WRITE:
 411         case HIO_DELETE:
 412                 hio->hio_offset = nv_get_uint64(hio->hio_nv, "offset");
 413                 if (nv_error(hio->hio_nv) != 0) {
 414                         pjdlog_error("Header is missing 'offset' field.");
 415                         hio->hio_error = EINVAL;
 416                         goto end;
 417                 }
 418                 hio->hio_length = nv_get_uint64(hio->hio_nv, "length");
 419                 if (nv_error(hio->hio_nv) != 0) {
 420                         pjdlog_error("Header is missing 'length' field.");
 421                         hio->hio_error = EINVAL;
 422                         goto end;
 423                 }
 424                 if (hio->hio_length == 0) {
 425                         pjdlog_error("Data length is zero.");
 426                         hio->hio_error = EINVAL;
 427                         goto end;
 428                 }
 429                 if (hio->hio_length > MAXPHYS) {
 430                         pjdlog_error("Data length is too large (%ju > %ju).",
 431                             (uintmax_t)hio->hio_length, (uintmax_t)MAXPHYS);
 432                         hio->hio_error = EINVAL;
 433                         goto end;
 434                 }
 435                 if ((hio->hio_offset % res->hr_local_sectorsize) != 0) {
 436                         pjdlog_error("Offset %ju is not multiple of sector size.",
 437                             (uintmax_t)hio->hio_offset);
 438                         hio->hio_error = EINVAL;
 439                         goto end;
 440                 }
 441                 if ((hio->hio_length % res->hr_local_sectorsize) != 0) {
 442                         pjdlog_error("Length %ju is not multiple of sector size.",
 443                             (uintmax_t)hio->hio_length);
 444                         hio->hio_error = EINVAL;
 445                         goto end;
 446                 }
 447                 if (hio->hio_offset + hio->hio_length >
 448                     (uint64_t)res->hr_datasize) {
 449                         pjdlog_error("Data offset is too large (%ju > %ju).",
 450                             (uintmax_t)(hio->hio_offset + hio->hio_length),
 451                             (uintmax_t)res->hr_datasize);
 452                         hio->hio_error = EINVAL;
 453                         goto end;
 454                 }
 455                 break;
 456         default:
 457                 pjdlog_error("Header contains invalid 'cmd' (%hhu).",
 458                     hio->hio_cmd);
 459                 hio->hio_error = EINVAL;
 460                 goto end;
 461         }
 462         hio->hio_error = 0;
 463 end:
 464         return (hio->hio_error);
 465 }
 466
 467 /*
 468  * Thread receives requests from the primary node.
 469  */
 470 static void *
 471 recv_thread(void *arg)
 472 {
 473         struct hast_resource *res = arg;
 474         struct hio *hio;
 475         bool wakeup;
 476
 477         for (;;) {
 478                 pjdlog_debug(2, "recv: Taking free request.");
 479                 mtx_lock(&hio_free_list_lock);
 480                 while ((hio = TAILQ_FIRST(&hio_free_list)) == NULL) {
 481                         pjdlog_debug(2, "recv: No free requests, waiting.");
 482                         cv_wait(&hio_free_list_cond, &hio_free_list_lock);
 483                 }
 484                 TAILQ_REMOVE(&hio_free_list, hio, hio_next);
 485                 mtx_unlock(&hio_free_list_lock);
 486                 pjdlog_debug(2, "recv: (%p) Got request.", hio);
 487                 if (hast_proto_recv_hdr(res->hr_remotein, &hio->hio_nv) < 0) {
 488                         pjdlog_exit(EX_TEMPFAIL,
 489                             "Unable to receive request header");
 490                 }
 491                 if (requnpack(res, hio) != 0)
 492                         goto send_queue;
 493                 reqlog(LOG_DEBUG, 2, -1, hio,
 494                     "recv: (%p) Got request header: ", hio);
 495                 if (hio->hio_cmd == HIO_WRITE) {
 496                         if (hast_proto_recv_data(res, res->hr_remotein,
 497                             hio->hio_nv, hio->hio_data, MAXPHYS) < 0) {
 498                                 pjdlog_exit(EX_TEMPFAIL,
 499                                     "Unable to receive reply data");
 500                         }
 501                 }
 502                 pjdlog_debug(2, "recv: (%p) Moving request to the disk queue.",
 503                     hio);
 504                 mtx_lock(&hio_disk_list_lock);
 505                 wakeup = TAILQ_EMPTY(&hio_disk_list);
 506                 TAILQ_INSERT_TAIL(&hio_disk_list, hio, hio_next);
 507                 mtx_unlock(&hio_disk_list_lock);
 508                 if (wakeup)
 509                         cv_signal(&hio_disk_list_cond);
 510                 continue;
 511 send_queue:
 512                 pjdlog_debug(2, "recv: (%p) Moving request to the send queue.",
 513                     hio);
 514                 mtx_lock(&hio_send_list_lock);
 515                 wakeup = TAILQ_EMPTY(&hio_send_list);
 516                 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
 517                 mtx_unlock(&hio_send_list_lock);
 518                 if (wakeup)
 519                         cv_signal(&hio_send_list_cond);
 520         }
 521         /* NOTREACHED */
 522         return (NULL);
 523 }
 524
 525 /*
 526  * Thread reads from or writes to local component and also handles DELETE and
 527  * FLUSH requests.
 528  */
 529 static void *
 530 disk_thread(void *arg)
 531 {
 532         struct hast_resource *res = arg;
 533         struct hio *hio;
 534         ssize_t ret;
 535         bool clear_activemap, wakeup;
 536
 537         clear_activemap = true;
 538
 539         for (;;) {
 540                 pjdlog_debug(2, "disk: Taking request.");
 541                 mtx_lock(&hio_disk_list_lock);
 542                 while ((hio = TAILQ_FIRST(&hio_disk_list)) == NULL) {
 543                         pjdlog_debug(2, "disk: No requests, waiting.");
 544                         cv_wait(&hio_disk_list_cond, &hio_disk_list_lock);
 545                 }
 546                 TAILQ_REMOVE(&hio_disk_list, hio, hio_next);
 547                 mtx_unlock(&hio_disk_list_lock);
 548                 while (clear_activemap) {
 549                         unsigned char *map;
 550                         size_t mapsize;
 551
 552                         /*
 553                          * When first request is received, it means that primary
 554                          * already received our activemap, merged it and stored
 555                          * locally. We can now safely clear our activemap.
 556                          */
 557                         mapsize =
 558                             activemap_calc_ondisk_size(res->hr_local_mediasize -
 559                             METADATA_SIZE, res->hr_extentsize,
 560                             res->hr_local_sectorsize);
 561                         map = calloc(1, mapsize);
 562                         if (map == NULL) {
 563                                 pjdlog_warning("Unable to allocate memory to clear local activemap.");
 564                                 break;
 565                         }
 566                         if (pwrite(res->hr_localfd, map, mapsize,
 567                             METADATA_SIZE) != (ssize_t)mapsize) {
 568                                 pjdlog_errno(LOG_WARNING,
 569                                     "Unable to store cleared activemap");
 570                                 free(map);
 571                                 break;
 572                         }
 573                         free(map);
 574                         clear_activemap = false;
 575                         pjdlog_debug(1, "Local activemap cleared.");
 576                 }
 577                 reqlog(LOG_DEBUG, 2, -1, hio, "disk: (%p) Got request: ", hio);
 578                 /* Handle the actual request. */
 579                 switch (hio->hio_cmd) {
 580                 case HIO_READ:
 581                         ret = pread(res->hr_localfd, hio->hio_data,
 582                             hio->hio_length,
 583                             hio->hio_offset + res->hr_localoff);
 584                         if (ret < 0)
 585                                 hio->hio_error = errno;
 586                         else if (ret != (int64_t)hio->hio_length)
 587                                 hio->hio_error = EIO;
 588                         else
 589                                 hio->hio_error = 0;
 590                         break;
 591                 case HIO_WRITE:
 592                         ret = pwrite(res->hr_localfd, hio->hio_data,
 593                             hio->hio_length,
 594                             hio->hio_offset + res->hr_localoff);
 595                         if (ret < 0)
 596                                 hio->hio_error = errno;
 597                         else if (ret != (int64_t)hio->hio_length)
 598                                 hio->hio_error = EIO;
 599                         else
 600                                 hio->hio_error = 0;
 601                         break;
 602                 case HIO_DELETE:
 603                         ret = g_delete(res->hr_localfd,
 604                             hio->hio_offset + res->hr_localoff,
 605                             hio->hio_length);
 606                         if (ret < 0)
 607                                 hio->hio_error = errno;
 608                         else
 609                                 hio->hio_error = 0;
 610                         break;
 611                 case HIO_FLUSH:
 612                         ret = g_flush(res->hr_localfd);
 613                         if (ret < 0)
 614                                 hio->hio_error = errno;
 615                         else
 616                                 hio->hio_error = 0;
 617                         break;
 618                 }
 619                 if (hio->hio_error != 0) {
 620                         reqlog(LOG_ERR, 0, hio->hio_error, hio,
 621                             "Request failed: ");
 622                 }
 623                 pjdlog_debug(2, "disk: (%p) Moving request to the send queue.",
 624                     hio);
 625                 mtx_lock(&hio_send_list_lock);
 626                 wakeup = TAILQ_EMPTY(&hio_send_list);
 627                 TAILQ_INSERT_TAIL(&hio_send_list, hio, hio_next);
 628                 mtx_unlock(&hio_send_list_lock);
 629                 if (wakeup)
 630                         cv_signal(&hio_send_list_cond);
 631         }
 632         /* NOTREACHED */
 633         return (NULL);
 634 }
 635
 636 /*
 637  * Thread sends requests back to primary node.
 638  */
 639 static void *
 640 send_thread(void *arg)
 641 {
 642         struct hast_resource *res = arg;
 643         struct nv *nvout;
 644         struct hio *hio;
 645         void *data;
 646         size_t length;
 647         bool wakeup;
 648
 649         for (;;) {
 650                 pjdlog_debug(2, "send: Taking request.");
 651                 mtx_lock(&hio_send_list_lock);
 652                 while ((hio = TAILQ_FIRST(&hio_send_list)) == NULL) {
 653                         pjdlog_debug(2, "send: No requests, waiting.");
 654                         cv_wait(&hio_send_list_cond, &hio_send_list_lock);
 655                 }
 656                 TAILQ_REMOVE(&hio_send_list, hio, hio_next);
 657                 mtx_unlock(&hio_send_list_lock);
 658                 reqlog(LOG_DEBUG, 2, -1, hio, "send: (%p) Got request: ", hio);
 659                 nvout = nv_alloc();
 660                 /* Copy sequence number. */
 661                 nv_add_uint64(nvout, nv_get_uint64(hio->hio_nv, "seq"), "seq");
 662                 switch (hio->hio_cmd) {
 663                 case HIO_READ:
 664                         if (hio->hio_error == 0) {
 665                                 data = hio->hio_data;
 666                                 length = hio->hio_length;
 667                                 break;
 668                         }
 669                         /*
 670                          * We send no data in case of an error.
 671                          */
 672                         /* FALLTHROUGH */
 673                 case HIO_DELETE:
 674                 case HIO_FLUSH:
 675                 case HIO_WRITE:
 676                         data = NULL;
 677                         length = 0;
 678                         break;
 679                 default:
 680                         abort();
 681                         break;
 682                 }
 683                 if (hio->hio_error != 0)
 684                         nv_add_int16(nvout, hio->hio_error, "error");
 685                 if (hast_proto_send(res, res->hr_remoteout, nvout, data,
 686                     length) < 0) {
 687                         pjdlog_exit(EX_TEMPFAIL, "Unable to send reply.");
 688                 }
 689                 nv_free(nvout);
 690                 pjdlog_debug(2, "disk: (%p) Moving request to the free queue.",
 691                     hio);
 692                 nv_free(hio->hio_nv);
 693                 hio->hio_error = 0;
 694                 mtx_lock(&hio_free_list_lock);
 695                 wakeup = TAILQ_EMPTY(&hio_free_list);
 696                 TAILQ_INSERT_TAIL(&hio_free_list, hio, hio_next);
 697                 mtx_unlock(&hio_free_list_lock);
 698                 if (wakeup)
 699                         cv_signal(&hio_free_list_cond);
 700         }
 701         /* NOTREACHED */
 702         return (NULL);
 703 }