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         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 }