sbin/hastd/primary.c

   1 /*-
   2  * Copyright (c) 2009 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$");
  33
  34 #include <sys/types.h>
  35 #include <sys/time.h>
  36 #include <sys/bio.h>
  37 #include <sys/disk.h>
  38 #include <sys/refcount.h>
  39 #include <sys/stat.h>
  40
  41 #include <geom/gate/g_gate.h>
  42
  43 #include <assert.h>
  44 #include <err.h>
  45 #include <errno.h>
  46 #include <fcntl.h>
  47 #include <libgeom.h>
  48 #include <pthread.h>
  49 #include <signal.h>
  50 #include <stdint.h>
  51 #include <stdio.h>
  52 #include <string.h>
  53 #include <sysexits.h>
  54 #include <unistd.h>
  55
  56 #include <activemap.h>
  57 #include <nv.h>
  58 #include <rangelock.h>
  59
  60 #include "control.h"
  61 #include "event.h"
  62 #include "hast.h"
  63 #include "hast_proto.h"
  64 #include "hastd.h"
  65 #include "hooks.h"
  66 #include "metadata.h"
  67 #include "proto.h"
  68 #include "pjdlog.h"
  69 #include "subr.h"
  70 #include "synch.h"
  71
  72 /* The is only one remote component for now. */
  73 #define ISREMOTE(no)    ((no) == 1)
  74
  75 struct hio {
  76         /*
  77          * Number of components we are still waiting for.
  78          * When this field goes to 0, we can send the request back to the
  79          * kernel. Each component has to decrease this counter by one
  80          * even on failure.
  81          */
  82         unsigned int             hio_countdown;
  83         /*
  84          * Each component has a place to store its own error.
  85          * Once the request is handled by all components we can decide if the
  86          * request overall is successful or not.
  87          */
  88         int                     *hio_errors;
  89         /*
  90          * Structure used to comunicate with GEOM Gate class.
  91          */
  92         struct g_gate_ctl_io     hio_ggio;
  93         TAILQ_ENTRY(hio)        *hio_next;
  94 };
  95 #define hio_free_next   hio_next[0]
  96 #define hio_done_next   hio_next[0]
  97
  98 /*
  99  * Free list holds unused structures. When free list is empty, we have to wait
 100  * until some in-progress requests are freed.
 101  */
 102 static TAILQ_HEAD(, hio) hio_free_list;
 103 static pthread_mutex_t hio_free_list_lock;
 104 static pthread_cond_t hio_free_list_cond;
 105 /*
 106  * There is one send list for every component. One requests is placed on all
 107  * send lists - each component gets the same request, but each component is
 108  * responsible for managing his own send list.
 109  */
 110 static TAILQ_HEAD(, hio) *hio_send_list;
 111 static pthread_mutex_t *hio_send_list_lock;
 112 static pthread_cond_t *hio_send_list_cond;
 113 /*
 114  * There is one recv list for every component, although local components don't
 115  * use recv lists as local requests are done synchronously.
 116  */
 117 static TAILQ_HEAD(, hio) *hio_recv_list;
 118 static pthread_mutex_t *hio_recv_list_lock;
 119 static pthread_cond_t *hio_recv_list_cond;
 120 /*
 121  * Request is placed on done list by the slowest component (the one that
 122  * decreased hio_countdown from 1 to 0).
 123  */
 124 static TAILQ_HEAD(, hio) hio_done_list;
 125 static pthread_mutex_t hio_done_list_lock;
 126 static pthread_cond_t hio_done_list_cond;
 127 /*
 128  * Structure below are for interaction with sync thread.
 129  */
 130 static bool sync_inprogress;
 131 static pthread_mutex_t sync_lock;
 132 static pthread_cond_t sync_cond;
 133 /*
 134  * The lock below allows to synchornize access to remote connections.
 135  */
 136 static pthread_rwlock_t *hio_remote_lock;
 137
 138 /*
 139  * Lock to synchronize metadata updates. Also synchronize access to
 140  * hr_primary_localcnt and hr_primary_remotecnt fields.
 141  */
 142 static pthread_mutex_t metadata_lock;
 143
 144 /*
 145  * Maximum number of outstanding I/O requests.
 146  */
 147 #define HAST_HIO_MAX    256
 148 /*
 149  * Number of components. At this point there are only two components: local
 150  * and remote, but in the future it might be possible to use multiple local
 151  * and remote components.
 152  */
 153 #define HAST_NCOMPONENTS        2
 154 /*
 155  * Number of seconds to sleep between reconnect retries or keepalive packets.
 156  */
 157 #define RETRY_SLEEP             10
 158
 159 #define ISCONNECTED(res, no)    \
 160         ((res)->hr_remotein != NULL && (res)->hr_remoteout != NULL)
 161
 162 #define QUEUE_INSERT1(hio, name, ncomp) do {                            \
 163         bool _wakeup;                                                   \
 164                                                                         \
 165         mtx_lock(&hio_##name##_list_lock[(ncomp)]);                     \
 166         _wakeup = TAILQ_EMPTY(&hio_##name##_list[(ncomp)]);             \
 167         TAILQ_INSERT_TAIL(&hio_##name##_list[(ncomp)], (hio),           \
 168             hio_next[(ncomp)]);                                         \
 169         mtx_unlock(&hio_##name##_list_lock[ncomp]);                     \
 170         if (_wakeup)                                                    \
 171                 cv_signal(&hio_##name##_list_cond[(ncomp)]);            \
 172 } while (0)
 173 #define QUEUE_INSERT2(hio, name)        do {                            \
 174         bool _wakeup;                                                   \
 175                                                                         \
 176         mtx_lock(&hio_##name##_list_lock);                              \
 177         _wakeup = TAILQ_EMPTY(&hio_##name##_list);                      \
 178         TAILQ_INSERT_TAIL(&hio_##name##_list, (hio), hio_##name##_next);\
 179         mtx_unlock(&hio_##name##_list_lock);                            \
 180         if (_wakeup)                                                    \
 181                 cv_signal(&hio_##name##_list_cond);                     \
 182 } while (0)
 183 #define QUEUE_TAKE1(hio, name, ncomp, timeout)  do {                    \
 184         bool _last;                                                     \
 185                                                                         \
 186         mtx_lock(&hio_##name##_list_lock[(ncomp)]);                     \
 187         _last = false;                                                  \
 188         while (((hio) = TAILQ_FIRST(&hio_##name##_list[(ncomp)])) == NULL && !_last) { \
 189                 cv_timedwait(&hio_##name##_list_cond[(ncomp)],          \
 190                     &hio_##name##_list_lock[(ncomp)], (timeout));       \
 191                 if ((timeout) != 0)                                     \
 192                         _last = true;                                   \
 193         }                                                               \
 194         if (hio != NULL) {                                              \
 195                 TAILQ_REMOVE(&hio_##name##_list[(ncomp)], (hio),        \
 196                     hio_next[(ncomp)]);                                 \
 197         }                                                               \
 198         mtx_unlock(&hio_##name##_list_lock[(ncomp)]);                   \
 199 } while (0)
 200 #define QUEUE_TAKE2(hio, name)  do {                                    \
 201         mtx_lock(&hio_##name##_list_lock);                              \
 202         while (((hio) = TAILQ_FIRST(&hio_##name##_list)) == NULL) {     \
 203                 cv_wait(&hio_##name##_list_cond,                        \
 204                     &hio_##name##_list_lock);                           \
 205         }                                                               \
 206         TAILQ_REMOVE(&hio_##name##_list, (hio), hio_##name##_next);     \
 207         mtx_unlock(&hio_##name##_list_lock);                            \
 208 } while (0)
 209
 210 #define SYNCREQ(hio)            do {                                    \
 211         (hio)->hio_ggio.gctl_unit = -1;                                 \
 212         (hio)->hio_ggio.gctl_seq = 1;                                   \
 213 } while (0)
 214 #define ISSYNCREQ(hio)          ((hio)->hio_ggio.gctl_unit == -1)
 215 #define SYNCREQDONE(hio)        do { (hio)->hio_ggio.gctl_unit = -2; } while (0)
 216 #define ISSYNCREQDONE(hio)      ((hio)->hio_ggio.gctl_unit == -2)
 217
 218 static struct hast_resource *gres;
 219
 220 static pthread_mutex_t range_lock;
 221 static struct rangelocks *range_regular;
 222 static bool range_regular_wait;
 223 static pthread_cond_t range_regular_cond;
 224 static struct rangelocks *range_sync;
 225 static bool range_sync_wait;
 226 static pthread_cond_t range_sync_cond;
 227
 228 static void *ggate_recv_thread(void *arg);
 229 static void *local_send_thread(void *arg);
 230 static void *remote_send_thread(void *arg);
 231 static void *remote_recv_thread(void *arg);
 232 static void *ggate_send_thread(void *arg);
 233 static void *sync_thread(void *arg);
 234 static void *guard_thread(void *arg);
 235
 236 static void
 237 cleanup(struct hast_resource *res)
 238 {
 239         int rerrno;
 240
 241         /* Remember errno. */
 242         rerrno = errno;
 243
 244         /* Destroy ggate provider if we created one. */
 245         if (res->hr_ggateunit >= 0) {
 246                 struct g_gate_ctl_destroy ggiod;
 247
 248                 bzero(&ggiod, sizeof(ggiod));
 249                 ggiod.gctl_version = G_GATE_VERSION;
 250                 ggiod.gctl_unit = res->hr_ggateunit;
 251                 ggiod.gctl_force = 1;
 252                 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DESTROY, &ggiod) < 0) {
 253                         pjdlog_errno(LOG_WARNING,
 254                             "Unable to destroy hast/%s device",
 255                             res->hr_provname);
 256                 }
 257                 res->hr_ggateunit = -1;
 258         }
 259
 260         /* Restore errno. */
 261         errno = rerrno;
 262 }
 263
 264 static __dead2 void
 265 primary_exit(int exitcode, const char *fmt, ...)
 266 {
 267         va_list ap;
 268
 269         assert(exitcode != EX_OK);
 270         va_start(ap, fmt);
 271         pjdlogv_errno(LOG_ERR, fmt, ap);
 272         va_end(ap);
 273         cleanup(gres);
 274         exit(exitcode);
 275 }
 276
 277 static __dead2 void
 278 primary_exitx(int exitcode, const char *fmt, ...)
 279 {
 280         va_list ap;
 281
 282         va_start(ap, fmt);
 283         pjdlogv(exitcode == EX_OK ? LOG_INFO : LOG_ERR, fmt, ap);
 284         va_end(ap);
 285         cleanup(gres);
 286         exit(exitcode);
 287 }
 288
 289 static int
 290 hast_activemap_flush(struct hast_resource *res)
 291 {
 292         const unsigned char *buf;
 293         size_t size;
 294
 295         buf = activemap_bitmap(res->hr_amp, &size);
 296         assert(buf != NULL);
 297         assert((size % res->hr_local_sectorsize) == 0);
 298         if (pwrite(res->hr_localfd, buf, size, METADATA_SIZE) !=
 299             (ssize_t)size) {
 300                 KEEP_ERRNO(pjdlog_errno(LOG_ERR,
 301                     "Unable to flush activemap to disk"));
 302                 return (-1);
 303         }
 304         return (0);
 305 }
 306
 307 static bool
 308 real_remote(const struct hast_resource *res)
 309 {
 310
 311         return (strcmp(res->hr_remoteaddr, "none") != 0);
 312 }
 313
 314 static void
 315 init_environment(struct hast_resource *res __unused)
 316 {
 317         struct hio *hio;
 318         unsigned int ii, ncomps;
 319
 320         /*
 321          * In the future it might be per-resource value.
 322          */
 323         ncomps = HAST_NCOMPONENTS;
 324
 325         /*
 326          * Allocate memory needed by lists.
 327          */
 328         hio_send_list = malloc(sizeof(hio_send_list[0]) * ncomps);
 329         if (hio_send_list == NULL) {
 330                 primary_exitx(EX_TEMPFAIL,
 331                     "Unable to allocate %zu bytes of memory for send lists.",
 332                     sizeof(hio_send_list[0]) * ncomps);
 333         }
 334         hio_send_list_lock = malloc(sizeof(hio_send_list_lock[0]) * ncomps);
 335         if (hio_send_list_lock == NULL) {
 336                 primary_exitx(EX_TEMPFAIL,
 337                     "Unable to allocate %zu bytes of memory for send list locks.",
 338                     sizeof(hio_send_list_lock[0]) * ncomps);
 339         }
 340         hio_send_list_cond = malloc(sizeof(hio_send_list_cond[0]) * ncomps);
 341         if (hio_send_list_cond == NULL) {
 342                 primary_exitx(EX_TEMPFAIL,
 343                     "Unable to allocate %zu bytes of memory for send list condition variables.",
 344                     sizeof(hio_send_list_cond[0]) * ncomps);
 345         }
 346         hio_recv_list = malloc(sizeof(hio_recv_list[0]) * ncomps);
 347         if (hio_recv_list == NULL) {
 348                 primary_exitx(EX_TEMPFAIL,
 349                     "Unable to allocate %zu bytes of memory for recv lists.",
 350                     sizeof(hio_recv_list[0]) * ncomps);
 351         }
 352         hio_recv_list_lock = malloc(sizeof(hio_recv_list_lock[0]) * ncomps);
 353         if (hio_recv_list_lock == NULL) {
 354                 primary_exitx(EX_TEMPFAIL,
 355                     "Unable to allocate %zu bytes of memory for recv list locks.",
 356                     sizeof(hio_recv_list_lock[0]) * ncomps);
 357         }
 358         hio_recv_list_cond = malloc(sizeof(hio_recv_list_cond[0]) * ncomps);
 359         if (hio_recv_list_cond == NULL) {
 360                 primary_exitx(EX_TEMPFAIL,
 361                     "Unable to allocate %zu bytes of memory for recv list condition variables.",
 362                     sizeof(hio_recv_list_cond[0]) * ncomps);
 363         }
 364         hio_remote_lock = malloc(sizeof(hio_remote_lock[0]) * ncomps);
 365         if (hio_remote_lock == NULL) {
 366                 primary_exitx(EX_TEMPFAIL,
 367                     "Unable to allocate %zu bytes of memory for remote connections locks.",
 368                     sizeof(hio_remote_lock[0]) * ncomps);
 369         }
 370
 371         /*
 372          * Initialize lists, their locks and theirs condition variables.
 373          */
 374         TAILQ_INIT(&hio_free_list);
 375         mtx_init(&hio_free_list_lock);
 376         cv_init(&hio_free_list_cond);
 377         for (ii = 0; ii < HAST_NCOMPONENTS; ii++) {
 378                 TAILQ_INIT(&hio_send_list[ii]);
 379                 mtx_init(&hio_send_list_lock[ii]);
 380                 cv_init(&hio_send_list_cond[ii]);
 381                 TAILQ_INIT(&hio_recv_list[ii]);
 382                 mtx_init(&hio_recv_list_lock[ii]);
 383                 cv_init(&hio_recv_list_cond[ii]);
 384                 rw_init(&hio_remote_lock[ii]);
 385         }
 386         TAILQ_INIT(&hio_done_list);
 387         mtx_init(&hio_done_list_lock);
 388         cv_init(&hio_done_list_cond);
 389         mtx_init(&metadata_lock);
 390
 391         /*
 392          * Allocate requests pool and initialize requests.
 393          */
 394         for (ii = 0; ii < HAST_HIO_MAX; ii++) {
 395                 hio = malloc(sizeof(*hio));
 396                 if (hio == NULL) {
 397                         primary_exitx(EX_TEMPFAIL,
 398                             "Unable to allocate %zu bytes of memory for hio request.",
 399                             sizeof(*hio));
 400                 }
 401                 hio->hio_countdown = 0;
 402                 hio->hio_errors = malloc(sizeof(hio->hio_errors[0]) * ncomps);
 403                 if (hio->hio_errors == NULL) {
 404                         primary_exitx(EX_TEMPFAIL,
 405                             "Unable allocate %zu bytes of memory for hio errors.",
 406                             sizeof(hio->hio_errors[0]) * ncomps);
 407                 }
 408                 hio->hio_next = malloc(sizeof(hio->hio_next[0]) * ncomps);
 409                 if (hio->hio_next == NULL) {
 410                         primary_exitx(EX_TEMPFAIL,
 411                             "Unable allocate %zu bytes of memory for hio_next field.",
 412                             sizeof(hio->hio_next[0]) * ncomps);
 413                 }
 414                 hio->hio_ggio.gctl_version = G_GATE_VERSION;
 415                 hio->hio_ggio.gctl_data = malloc(MAXPHYS);
 416                 if (hio->hio_ggio.gctl_data == NULL) {
 417                         primary_exitx(EX_TEMPFAIL,
 418                             "Unable to allocate %zu bytes of memory for gctl_data.",
 419                             MAXPHYS);
 420                 }
 421                 hio->hio_ggio.gctl_length = MAXPHYS;
 422                 hio->hio_ggio.gctl_error = 0;
 423                 TAILQ_INSERT_HEAD(&hio_free_list, hio, hio_free_next);
 424         }
 425 }
 426
 427 static bool
 428 init_resuid(struct hast_resource *res)
 429 {
 430
 431         mtx_lock(&metadata_lock);
 432         if (res->hr_resuid != 0) {
 433                 mtx_unlock(&metadata_lock);
 434                 return (false);
 435         } else {
 436                 /* Initialize unique resource identifier. */
 437                 arc4random_buf(&res->hr_resuid, sizeof(res->hr_resuid));
 438                 mtx_unlock(&metadata_lock);
 439                 if (metadata_write(res) < 0)
 440                         exit(EX_NOINPUT);
 441                 return (true);
 442         }
 443 }
 444
 445 static void
 446 init_local(struct hast_resource *res)
 447 {
 448         unsigned char *buf;
 449         size_t mapsize;
 450
 451         if (metadata_read(res, true) < 0)
 452                 exit(EX_NOINPUT);
 453         mtx_init(&res->hr_amp_lock);
 454         if (activemap_init(&res->hr_amp, res->hr_datasize, res->hr_extentsize,
 455             res->hr_local_sectorsize, res->hr_keepdirty) < 0) {
 456                 primary_exit(EX_TEMPFAIL, "Unable to create activemap");
 457         }
 458         mtx_init(&range_lock);
 459         cv_init(&range_regular_cond);
 460         if (rangelock_init(&range_regular) < 0)
 461                 primary_exit(EX_TEMPFAIL, "Unable to create regular range lock");
 462         cv_init(&range_sync_cond);
 463         if (rangelock_init(&range_sync) < 0)
 464                 primary_exit(EX_TEMPFAIL, "Unable to create sync range lock");
 465         mapsize = activemap_ondisk_size(res->hr_amp);
 466         buf = calloc(1, mapsize);
 467         if (buf == NULL) {
 468                 primary_exitx(EX_TEMPFAIL,
 469                     "Unable to allocate buffer for activemap.");
 470         }
 471         if (pread(res->hr_localfd, buf, mapsize, METADATA_SIZE) !=
 472             (ssize_t)mapsize) {
 473                 primary_exit(EX_NOINPUT, "Unable to read activemap");
 474         }
 475         activemap_copyin(res->hr_amp, buf, mapsize);
 476         free(buf);
 477         if (res->hr_resuid != 0)
 478                 return;
 479         /*
 480          * We're using provider for the first time. Initialize local and remote
 481          * counters. We don't initialize resuid here, as we want to do it just
 482          * in time. The reason for this is that we want to inform secondary
 483          * that there were no writes yet, so there is no need to synchronize
 484          * anything.
 485          */
 486         res->hr_primary_localcnt = 1;
 487         res->hr_primary_remotecnt = 0;
 488         if (metadata_write(res) < 0)
 489                 exit(EX_NOINPUT);
 490 }
 491
 492 static bool
 493 init_remote(struct hast_resource *res, struct proto_conn **inp,
 494     struct proto_conn **outp)
 495 {
 496         struct proto_conn *in, *out;
 497         struct nv *nvout, *nvin;
 498         const unsigned char *token;
 499         unsigned char *map;
 500         const char *errmsg;
 501         int32_t extentsize;
 502         int64_t datasize;
 503         uint32_t mapsize;
 504         size_t size;
 505
 506         assert((inp == NULL && outp == NULL) || (inp != NULL && outp != NULL));
 507         assert(real_remote(res));
 508
 509         in = out = NULL;
 510         errmsg = NULL;
 511
 512         /* Prepare outgoing connection with remote node. */
 513         if (proto_client(res->hr_remoteaddr, &out) < 0) {
 514                 primary_exit(EX_TEMPFAIL,
 515                     "Unable to create outgoing connection to %s",
 516                     res->hr_remoteaddr);
 517         }
 518         /* Try to connect, but accept failure. */
 519         if (proto_connect(out) < 0) {
 520                 pjdlog_errno(LOG_WARNING, "Unable to connect to %s",
 521                     res->hr_remoteaddr);
 522                 goto close;
 523         }
 524         /* Error in setting timeout is not critical, but why should it fail? */
 525         if (proto_timeout(out, res->hr_timeout) < 0)
 526                 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
 527         /*
 528          * First handshake step.
 529          * Setup outgoing connection with remote node.
 530          */
 531         nvout = nv_alloc();
 532         nv_add_string(nvout, res->hr_name, "resource");
 533         if (nv_error(nvout) != 0) {
 534                 pjdlog_common(LOG_WARNING, 0, nv_error(nvout),
 535                     "Unable to allocate header for connection with %s",
 536                     res->hr_remoteaddr);
 537                 nv_free(nvout);
 538                 goto close;
 539         }
 540         if (hast_proto_send(res, out, nvout, NULL, 0) < 0) {
 541                 pjdlog_errno(LOG_WARNING,
 542                     "Unable to send handshake header to %s",
 543                     res->hr_remoteaddr);
 544                 nv_free(nvout);
 545                 goto close;
 546         }
 547         nv_free(nvout);
 548         if (hast_proto_recv_hdr(out, &nvin) < 0) {
 549                 pjdlog_errno(LOG_WARNING,
 550                     "Unable to receive handshake header from %s",
 551                     res->hr_remoteaddr);
 552                 goto close;
 553         }
 554         errmsg = nv_get_string(nvin, "errmsg");
 555         if (errmsg != NULL) {
 556                 pjdlog_warning("%s", errmsg);
 557                 nv_free(nvin);
 558                 goto close;
 559         }
 560         token = nv_get_uint8_array(nvin, &size, "token");
 561         if (token == NULL) {
 562                 pjdlog_warning("Handshake header from %s has no 'token' field.",
 563                     res->hr_remoteaddr);
 564                 nv_free(nvin);
 565                 goto close;
 566         }
 567         if (size != sizeof(res->hr_token)) {
 568                 pjdlog_warning("Handshake header from %s contains 'token' of wrong size (got %zu, expected %zu).",
 569                     res->hr_remoteaddr, size, sizeof(res->hr_token));
 570                 nv_free(nvin);
 571                 goto close;
 572         }
 573         bcopy(token, res->hr_token, sizeof(res->hr_token));
 574         nv_free(nvin);
 575
 576         /*
 577          * Second handshake step.
 578          * Setup incoming connection with remote node.
 579          */
 580         if (proto_client(res->hr_remoteaddr, &in) < 0) {
 581                 primary_exit(EX_TEMPFAIL,
 582                     "Unable to create incoming connection to %s",
 583                     res->hr_remoteaddr);
 584         }
 585         /* Try to connect, but accept failure. */
 586         if (proto_connect(in) < 0) {
 587                 pjdlog_errno(LOG_WARNING, "Unable to connect to %s",
 588                     res->hr_remoteaddr);
 589                 goto close;
 590         }
 591         /* Error in setting timeout is not critical, but why should it fail? */
 592         if (proto_timeout(in, res->hr_timeout) < 0)
 593                 pjdlog_errno(LOG_WARNING, "Unable to set connection timeout");
 594         nvout = nv_alloc();
 595         nv_add_string(nvout, res->hr_name, "resource");
 596         nv_add_uint8_array(nvout, res->hr_token, sizeof(res->hr_token),
 597             "token");
 598         if (res->hr_resuid == 0) {
 599                 /*
 600                  * The resuid field was not yet initialized.
 601                  * Because we do synchronization inside init_resuid(), it is
 602                  * possible that someone already initialized it, the function
 603                  * will return false then, but if we successfully initialized
 604                  * it, we will get true. True means that there were no writes
 605                  * to this resource yet and we want to inform secondary that
 606                  * synchronization is not needed by sending "virgin" argument.
 607                  */
 608                 if (init_resuid(res))
 609                         nv_add_int8(nvout, 1, "virgin");
 610         }
 611         nv_add_uint64(nvout, res->hr_resuid, "resuid");
 612         nv_add_uint64(nvout, res->hr_primary_localcnt, "localcnt");
 613         nv_add_uint64(nvout, res->hr_primary_remotecnt, "remotecnt");
 614         if (nv_error(nvout) != 0) {
 615                 pjdlog_common(LOG_WARNING, 0, nv_error(nvout),
 616                     "Unable to allocate header for connection with %s",
 617                     res->hr_remoteaddr);
 618                 nv_free(nvout);
 619                 goto close;
 620         }
 621         if (hast_proto_send(res, in, nvout, NULL, 0) < 0) {
 622                 pjdlog_errno(LOG_WARNING,
 623                     "Unable to send handshake header to %s",
 624                     res->hr_remoteaddr);
 625                 nv_free(nvout);
 626                 goto close;
 627         }
 628         nv_free(nvout);
 629         if (hast_proto_recv_hdr(out, &nvin) < 0) {
 630                 pjdlog_errno(LOG_WARNING,
 631                     "Unable to receive handshake header from %s",
 632                     res->hr_remoteaddr);
 633                 goto close;
 634         }
 635         errmsg = nv_get_string(nvin, "errmsg");
 636         if (errmsg != NULL) {
 637                 pjdlog_warning("%s", errmsg);
 638                 nv_free(nvin);
 639                 goto close;
 640         }
 641         datasize = nv_get_int64(nvin, "datasize");
 642         if (datasize != res->hr_datasize) {
 643                 pjdlog_warning("Data size differs between nodes (local=%jd, remote=%jd).",
 644                     (intmax_t)res->hr_datasize, (intmax_t)datasize);
 645                 nv_free(nvin);
 646                 goto close;
 647         }
 648         extentsize = nv_get_int32(nvin, "extentsize");
 649         if (extentsize != res->hr_extentsize) {
 650                 pjdlog_warning("Extent size differs between nodes (local=%zd, remote=%zd).",
 651                     (ssize_t)res->hr_extentsize, (ssize_t)extentsize);
 652                 nv_free(nvin);
 653                 goto close;
 654         }
 655         res->hr_secondary_localcnt = nv_get_uint64(nvin, "localcnt");
 656         res->hr_secondary_remotecnt = nv_get_uint64(nvin, "remotecnt");
 657         res->hr_syncsrc = nv_get_uint8(nvin, "syncsrc");
 658         map = NULL;
 659         mapsize = nv_get_uint32(nvin, "mapsize");
 660         if (mapsize > 0) {
 661                 map = malloc(mapsize);
 662                 if (map == NULL) {
 663                         pjdlog_error("Unable to allocate memory for remote activemap (mapsize=%ju).",
 664                             (uintmax_t)mapsize);
 665                         nv_free(nvin);
 666                         goto close;
 667                 }
 668                 /*
 669                  * Remote node have some dirty extents on its own, lets
 670                  * download its activemap.
 671                  */
 672                 if (hast_proto_recv_data(res, out, nvin, map,
 673                     mapsize) < 0) {
 674                         pjdlog_errno(LOG_ERR,
 675                             "Unable to receive remote activemap");
 676                         nv_free(nvin);
 677                         free(map);
 678                         goto close;
 679                 }
 680                 /*
 681                  * Merge local and remote bitmaps.
 682                  */
 683                 activemap_merge(res->hr_amp, map, mapsize);
 684                 free(map);
 685                 /*
 686                  * Now that we merged bitmaps from both nodes, flush it to the
 687                  * disk before we start to synchronize.
 688                  */
 689                 (void)hast_activemap_flush(res);
 690         }
 691         nv_free(nvin);
 692         pjdlog_info("Connected to %s.", res->hr_remoteaddr);
 693         if (inp != NULL && outp != NULL) {
 694                 *inp = in;
 695                 *outp = out;
 696         } else {
 697                 res->hr_remotein = in;
 698                 res->hr_remoteout = out;
 699         }
 700         event_send(res, EVENT_CONNECT);
 701         return (true);
 702 close:
 703         if (errmsg != NULL && strcmp(errmsg, "Split-brain condition!") == 0)
 704                 event_send(res, EVENT_SPLITBRAIN);
 705         proto_close(out);
 706         if (in != NULL)
 707                 proto_close(in);
 708         return (false);
 709 }
 710
 711 static void
 712 sync_start(void)
 713 {
 714
 715         mtx_lock(&sync_lock);
 716         sync_inprogress = true;
 717         mtx_unlock(&sync_lock);
 718         cv_signal(&sync_cond);
 719 }
 720
 721 static void
 722 sync_stop(void)
 723 {
 724
 725         mtx_lock(&sync_lock);
 726         if (sync_inprogress)
 727                 sync_inprogress = false;
 728         mtx_unlock(&sync_lock);
 729 }
 730
 731 static void
 732 init_ggate(struct hast_resource *res)
 733 {
 734         struct g_gate_ctl_create ggiocreate;
 735         struct g_gate_ctl_cancel ggiocancel;
 736
 737         /*
 738          * We communicate with ggate via /dev/ggctl. Open it.
 739          */
 740         res->hr_ggatefd = open("/dev/" G_GATE_CTL_NAME, O_RDWR);
 741         if (res->hr_ggatefd < 0)
 742                 primary_exit(EX_OSFILE, "Unable to open /dev/" G_GATE_CTL_NAME);
 743         /*
 744          * Create provider before trying to connect, as connection failure
 745          * is not critical, but may take some time.
 746          */
 747         bzero(&ggiocreate, sizeof(ggiocreate));
 748         ggiocreate.gctl_version = G_GATE_VERSION;
 749         ggiocreate.gctl_mediasize = res->hr_datasize;
 750         ggiocreate.gctl_sectorsize = res->hr_local_sectorsize;
 751         ggiocreate.gctl_flags = 0;
 752         ggiocreate.gctl_maxcount = G_GATE_MAX_QUEUE_SIZE;
 753         ggiocreate.gctl_timeout = 0;
 754         ggiocreate.gctl_unit = G_GATE_NAME_GIVEN;
 755         snprintf(ggiocreate.gctl_name, sizeof(ggiocreate.gctl_name), "hast/%s",
 756             res->hr_provname);
 757         if (ioctl(res->hr_ggatefd, G_GATE_CMD_CREATE, &ggiocreate) == 0) {
 758                 pjdlog_info("Device hast/%s created.", res->hr_provname);
 759                 res->hr_ggateunit = ggiocreate.gctl_unit;
 760                 return;
 761         }
 762         if (errno != EEXIST) {
 763                 primary_exit(EX_OSERR, "Unable to create hast/%s device",
 764                     res->hr_provname);
 765         }
 766         pjdlog_debug(1,
 767             "Device hast/%s already exists, we will try to take it over.",
 768             res->hr_provname);
 769         /*
 770          * If we received EEXIST, we assume that the process who created the
 771          * provider died and didn't clean up. In that case we will start from
 772          * where he left of.
 773          */
 774         bzero(&ggiocancel, sizeof(ggiocancel));
 775         ggiocancel.gctl_version = G_GATE_VERSION;
 776         ggiocancel.gctl_unit = G_GATE_NAME_GIVEN;
 777         snprintf(ggiocancel.gctl_name, sizeof(ggiocancel.gctl_name), "hast/%s",
 778             res->hr_provname);
 779         if (ioctl(res->hr_ggatefd, G_GATE_CMD_CANCEL, &ggiocancel) == 0) {
 780                 pjdlog_info("Device hast/%s recovered.", res->hr_provname);
 781                 res->hr_ggateunit = ggiocancel.gctl_unit;
 782                 return;
 783         }
 784         primary_exit(EX_OSERR, "Unable to take over hast/%s device",
 785             res->hr_provname);
 786 }
 787
 788 void
 789 hastd_primary(struct hast_resource *res)
 790 {
 791         pthread_t td;
 792         pid_t pid;
 793         int error, mode;
 794
 795         /*
 796          * Create communication channel between parent and child.
 797          */
 798         if (proto_client("socketpair://", &res->hr_ctrl) < 0) {
 799                 /* TODO: There's no need for this to be fatal error. */
 800                 KEEP_ERRNO((void)pidfile_remove(pfh));
 801                 pjdlog_exit(EX_OSERR,
 802                     "Unable to create control sockets between parent and child");
 803         }
 804         /*
 805          * Create communication channel between child and parent.
 806          */
 807         if (proto_client("socketpair://", &res->hr_event) < 0) {
 808                 /* TODO: There's no need for this to be fatal error. */
 809                 KEEP_ERRNO((void)pidfile_remove(pfh));
 810                 pjdlog_exit(EX_OSERR,
 811                     "Unable to create event sockets between child and parent");
 812         }
 813
 814         pid = fork();
 815         if (pid < 0) {
 816                 /* TODO: There's no need for this to be fatal error. */
 817                 KEEP_ERRNO((void)pidfile_remove(pfh));
 818                 pjdlog_exit(EX_TEMPFAIL, "Unable to fork");
 819         }
 820
 821         if (pid > 0) {
 822                 /* This is parent. */
 823                 /* Declare that we are receiver. */
 824                 proto_recv(res->hr_event, NULL, 0);
 825                 /* Declare that we are sender. */
 826                 proto_send(res->hr_ctrl, NULL, 0);
 827                 res->hr_workerpid = pid;
 828                 return;
 829         }
 830
 831         gres = res;
 832         mode = pjdlog_mode_get();
 833
 834         /* Declare that we are sender. */
 835         proto_send(res->hr_event, NULL, 0);
 836         /* Declare that we are receiver. */
 837         proto_recv(res->hr_ctrl, NULL, 0);
 838         descriptors_cleanup(res);
 839
 840         pjdlog_init(mode);
 841         pjdlog_prefix_set("[%s] (%s) ", res->hr_name, role2str(res->hr_role));
 842         setproctitle("%s (primary)", res->hr_name);
 843
 844         init_local(res);
 845         init_ggate(res);
 846         init_environment(res);
 847
 848         /*
 849          * Create the guard thread first, so we can handle signals from the
 850          * very begining.
 851          */
 852         error = pthread_create(&td, NULL, guard_thread, res);
 853         assert(error == 0);
 854         /*
 855          * Create the control thread before sending any event to the parent,
 856          * as we can deadlock when parent sends control request to worker,
 857          * but worker has no control thread started yet, so parent waits.
 858          * In the meantime worker sends an event to the parent, but parent
 859          * is unable to handle the event, because it waits for control
 860          * request response.
 861          */
 862         error = pthread_create(&td, NULL, ctrl_thread, res);
 863         assert(error == 0);
 864         if (real_remote(res) && init_remote(res, NULL, NULL))
 865                 sync_start();
 866         error = pthread_create(&td, NULL, ggate_recv_thread, res);
 867         assert(error == 0);
 868         error = pthread_create(&td, NULL, local_send_thread, res);
 869         assert(error == 0);
 870         error = pthread_create(&td, NULL, remote_send_thread, res);
 871         assert(error == 0);
 872         error = pthread_create(&td, NULL, remote_recv_thread, res);
 873         assert(error == 0);
 874         error = pthread_create(&td, NULL, ggate_send_thread, res);
 875         assert(error == 0);
 876         (void)sync_thread(res);
 877 }
 878
 879 static void
 880 reqlog(int loglevel, int debuglevel, struct g_gate_ctl_io *ggio, const char *fmt, ...)
 881 {
 882         char msg[1024];
 883         va_list ap;
 884         int len;
 885
 886         va_start(ap, fmt);
 887         len = vsnprintf(msg, sizeof(msg), fmt, ap);
 888         va_end(ap);
 889         if ((size_t)len < sizeof(msg)) {
 890                 switch (ggio->gctl_cmd) {
 891                 case BIO_READ:
 892                         (void)snprintf(msg + len, sizeof(msg) - len,
 893                             "READ(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
 894                             (uintmax_t)ggio->gctl_length);
 895                         break;
 896                 case BIO_DELETE:
 897                         (void)snprintf(msg + len, sizeof(msg) - len,
 898                             "DELETE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
 899                             (uintmax_t)ggio->gctl_length);
 900                         break;
 901                 case BIO_FLUSH:
 902                         (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH.");
 903                         break;
 904                 case BIO_WRITE:
 905                         (void)snprintf(msg + len, sizeof(msg) - len,
 906                             "WRITE(%ju, %ju).", (uintmax_t)ggio->gctl_offset,
 907                             (uintmax_t)ggio->gctl_length);
 908                         break;
 909                 default:
 910                         (void)snprintf(msg + len, sizeof(msg) - len,
 911                             "UNKNOWN(%u).", (unsigned int)ggio->gctl_cmd);
 912                         break;
 913                 }
 914         }
 915         pjdlog_common(loglevel, debuglevel, -1, "%s", msg);
 916 }
 917
 918 static void
 919 remote_close(struct hast_resource *res, int ncomp)
 920 {
 921
 922         rw_wlock(&hio_remote_lock[ncomp]);
 923         /*
 924          * A race is possible between dropping rlock and acquiring wlock -
 925          * another thread can close connection in-between.
 926          */
 927         if (!ISCONNECTED(res, ncomp)) {
 928                 assert(res->hr_remotein == NULL);
 929                 assert(res->hr_remoteout == NULL);
 930                 rw_unlock(&hio_remote_lock[ncomp]);
 931                 return;
 932         }
 933
 934         assert(res->hr_remotein != NULL);
 935         assert(res->hr_remoteout != NULL);
 936
 937         pjdlog_debug(2, "Closing incoming connection to %s.",
 938             res->hr_remoteaddr);
 939         proto_close(res->hr_remotein);
 940         res->hr_remotein = NULL;
 941         pjdlog_debug(2, "Closing outgoing connection to %s.",
 942             res->hr_remoteaddr);
 943         proto_close(res->hr_remoteout);
 944         res->hr_remoteout = NULL;
 945
 946         rw_unlock(&hio_remote_lock[ncomp]);
 947
 948         pjdlog_warning("Disconnected from %s.", res->hr_remoteaddr);
 949
 950         /*
 951          * Stop synchronization if in-progress.
 952          */
 953         sync_stop();
 954
 955         event_send(res, EVENT_DISCONNECT);
 956 }
 957
 958 /*
 959  * Thread receives ggate I/O requests from the kernel and passes them to
 960  * appropriate threads:
 961  * WRITE - always goes to both local_send and remote_send threads
 962  * READ (when the block is up-to-date on local component) -
 963  *      only local_send thread
 964  * READ (when the block isn't up-to-date on local component) -
 965  *      only remote_send thread
 966  * DELETE - always goes to both local_send and remote_send threads
 967  * FLUSH - always goes to both local_send and remote_send threads
 968  */
 969 static void *
 970 ggate_recv_thread(void *arg)
 971 {
 972         struct hast_resource *res = arg;
 973         struct g_gate_ctl_io *ggio;
 974         struct hio *hio;
 975         unsigned int ii, ncomp, ncomps;
 976         int error;
 977
 978         ncomps = HAST_NCOMPONENTS;
 979
 980         for (;;) {
 981                 pjdlog_debug(2, "ggate_recv: Taking free request.");
 982                 QUEUE_TAKE2(hio, free);
 983                 pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio);
 984                 ggio = &hio->hio_ggio;
 985                 ggio->gctl_unit = res->hr_ggateunit;
 986                 ggio->gctl_length = MAXPHYS;
 987                 ggio->gctl_error = 0;
 988                 pjdlog_debug(2,
 989                     "ggate_recv: (%p) Waiting for request from the kernel.",
 990                     hio);
 991                 if (ioctl(res->hr_ggatefd, G_GATE_CMD_START, ggio) < 0) {
 992                         if (sigexit_received)
 993                                 pthread_exit(NULL);
 994                         primary_exit(EX_OSERR, "G_GATE_CMD_START failed");
 995                 }
 996                 error = ggio->gctl_error;
 997                 switch (error) {
 998                 case 0:
 999                         break;
1000                 case ECANCELED:
1001                         /* Exit gracefully. */
1002                         if (!sigexit_received) {
1003                                 pjdlog_debug(2,
1004                                     "ggate_recv: (%p) Received cancel from the kernel.",
1005                                     hio);
1006                                 pjdlog_info("Received cancel from the kernel, exiting.");
1007                         }
1008                         pthread_exit(NULL);
1009                 case ENOMEM:
1010                         /*
1011                          * Buffer too small? Impossible, we allocate MAXPHYS
1012                          * bytes - request can't be bigger than that.
1013                          */
1014                         /* FALLTHROUGH */
1015                 case ENXIO:
1016                 default:
1017                         primary_exitx(EX_OSERR, "G_GATE_CMD_START failed: %s.",
1018                             strerror(error));
1019                 }
1020                 for (ii = 0; ii < ncomps; ii++)
1021                         hio->hio_errors[ii] = EINVAL;
1022                 reqlog(LOG_DEBUG, 2, ggio,
1023                     "ggate_recv: (%p) Request received from the kernel: ",
1024                     hio);
1025                 /*
1026                  * Inform all components about new write request.
1027                  * For read request prefer local component unless the given
1028                  * range is out-of-date, then use remote component.
1029                  */
1030                 switch (ggio->gctl_cmd) {
1031                 case BIO_READ:
1032                         pjdlog_debug(2,
1033                             "ggate_recv: (%p) Moving request to the send queue.",
1034                             hio);
1035                         refcount_init(&hio->hio_countdown, 1);
1036                         mtx_lock(&metadata_lock);
1037                         if (res->hr_syncsrc == HAST_SYNCSRC_UNDEF ||
1038                             res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1039                                 /*
1040                                  * This range is up-to-date on local component,
1041                                  * so handle request locally.
1042                                  */
1043                                  /* Local component is 0 for now. */
1044                                 ncomp = 0;
1045                         } else /* if (res->hr_syncsrc ==
1046                             HAST_SYNCSRC_SECONDARY) */ {
1047                                 assert(res->hr_syncsrc ==
1048                                     HAST_SYNCSRC_SECONDARY);
1049                                 /*
1050                                  * This range is out-of-date on local component,
1051                                  * so send request to the remote node.
1052                                  */
1053                                  /* Remote component is 1 for now. */
1054                                 ncomp = 1;
1055                         }
1056                         mtx_unlock(&metadata_lock);
1057                         QUEUE_INSERT1(hio, send, ncomp);
1058                         break;
1059                 case BIO_WRITE:
1060                         if (res->hr_resuid == 0) {
1061                                 /* This is first write, initialize resuid. */
1062                                 (void)init_resuid(res);
1063                         }
1064                         for (;;) {
1065                                 mtx_lock(&range_lock);
1066                                 if (rangelock_islocked(range_sync,
1067                                     ggio->gctl_offset, ggio->gctl_length)) {
1068                                         pjdlog_debug(2,
1069                                             "regular: Range offset=%jd length=%zu locked.",
1070                                             (intmax_t)ggio->gctl_offset,
1071                                             (size_t)ggio->gctl_length);
1072                                         range_regular_wait = true;
1073                                         cv_wait(&range_regular_cond, &range_lock);
1074                                         range_regular_wait = false;
1075                                         mtx_unlock(&range_lock);
1076                                         continue;
1077                                 }
1078                                 if (rangelock_add(range_regular,
1079                                     ggio->gctl_offset, ggio->gctl_length) < 0) {
1080                                         mtx_unlock(&range_lock);
1081                                         pjdlog_debug(2,
1082                                             "regular: Range offset=%jd length=%zu is already locked, waiting.",
1083                                             (intmax_t)ggio->gctl_offset,
1084                                             (size_t)ggio->gctl_length);
1085                                         sleep(1);
1086                                         continue;
1087                                 }
1088                                 mtx_unlock(&range_lock);
1089                                 break;
1090                         }
1091                         mtx_lock(&res->hr_amp_lock);
1092                         if (activemap_write_start(res->hr_amp,
1093                             ggio->gctl_offset, ggio->gctl_length)) {
1094                                 (void)hast_activemap_flush(res);
1095                         }
1096                         mtx_unlock(&res->hr_amp_lock);
1097                         /* FALLTHROUGH */
1098                 case BIO_DELETE:
1099                 case BIO_FLUSH:
1100                         pjdlog_debug(2,
1101                             "ggate_recv: (%p) Moving request to the send queues.",
1102                             hio);
1103                         refcount_init(&hio->hio_countdown, ncomps);
1104                         for (ii = 0; ii < ncomps; ii++)
1105                                 QUEUE_INSERT1(hio, send, ii);
1106                         break;
1107                 }
1108         }
1109         /* NOTREACHED */
1110         return (NULL);
1111 }
1112
1113 /*
1114  * Thread reads from or writes to local component.
1115  * If local read fails, it redirects it to remote_send thread.
1116  */
1117 static void *
1118 local_send_thread(void *arg)
1119 {
1120         struct hast_resource *res = arg;
1121         struct g_gate_ctl_io *ggio;
1122         struct hio *hio;
1123         unsigned int ncomp, rncomp;
1124         ssize_t ret;
1125
1126         /* Local component is 0 for now. */
1127         ncomp = 0;
1128         /* Remote component is 1 for now. */
1129         rncomp = 1;
1130
1131         for (;;) {
1132                 pjdlog_debug(2, "local_send: Taking request.");
1133                 QUEUE_TAKE1(hio, send, ncomp, 0);
1134                 pjdlog_debug(2, "local_send: (%p) Got request.", hio);
1135                 ggio = &hio->hio_ggio;
1136                 switch (ggio->gctl_cmd) {
1137                 case BIO_READ:
1138                         ret = pread(res->hr_localfd, ggio->gctl_data,
1139                             ggio->gctl_length,
1140                             ggio->gctl_offset + res->hr_localoff);
1141                         if (ret == ggio->gctl_length)
1142                                 hio->hio_errors[ncomp] = 0;
1143                         else {
1144                                 /*
1145                                  * If READ failed, try to read from remote node.
1146                                  */
1147                                 if (ret < 0) {
1148                                         reqlog(LOG_WARNING, 0, ggio,
1149                                             "Local request failed (%s), trying remote node. ",
1150                                             strerror(errno));
1151                                 } else if (ret != ggio->gctl_length) {
1152                                         reqlog(LOG_WARNING, 0, ggio,
1153                                             "Local request failed (%zd != %jd), trying remote node. ",
1154                                             ret, (intmax_t)ggio->gctl_length);
1155                                 }
1156                                 QUEUE_INSERT1(hio, send, rncomp);
1157                                 continue;
1158                         }
1159                         break;
1160                 case BIO_WRITE:
1161                         ret = pwrite(res->hr_localfd, ggio->gctl_data,
1162                             ggio->gctl_length,
1163                             ggio->gctl_offset + res->hr_localoff);
1164                         if (ret < 0) {
1165                                 hio->hio_errors[ncomp] = errno;
1166                                 reqlog(LOG_WARNING, 0, ggio,
1167                                     "Local request failed (%s): ",
1168                                     strerror(errno));
1169                         } else if (ret != ggio->gctl_length) {
1170                                 hio->hio_errors[ncomp] = EIO;
1171                                 reqlog(LOG_WARNING, 0, ggio,
1172                                     "Local request failed (%zd != %jd): ",
1173                                     ret, (intmax_t)ggio->gctl_length);
1174                         } else {
1175                                 hio->hio_errors[ncomp] = 0;
1176                         }
1177                         break;
1178                 case BIO_DELETE:
1179                         ret = g_delete(res->hr_localfd,
1180                             ggio->gctl_offset + res->hr_localoff,
1181                             ggio->gctl_length);
1182                         if (ret < 0) {
1183                                 hio->hio_errors[ncomp] = errno;
1184                                 reqlog(LOG_WARNING, 0, ggio,
1185                                     "Local request failed (%s): ",
1186                                     strerror(errno));
1187                         } else {
1188                                 hio->hio_errors[ncomp] = 0;
1189                         }
1190                         break;
1191                 case BIO_FLUSH:
1192                         ret = g_flush(res->hr_localfd);
1193                         if (ret < 0) {
1194                                 hio->hio_errors[ncomp] = errno;
1195                                 reqlog(LOG_WARNING, 0, ggio,
1196                                     "Local request failed (%s): ",
1197                                     strerror(errno));
1198                         } else {
1199                                 hio->hio_errors[ncomp] = 0;
1200                         }
1201                         break;
1202                 }
1203                 if (refcount_release(&hio->hio_countdown)) {
1204                         if (ISSYNCREQ(hio)) {
1205                                 mtx_lock(&sync_lock);
1206                                 SYNCREQDONE(hio);
1207                                 mtx_unlock(&sync_lock);
1208                                 cv_signal(&sync_cond);
1209                         } else {
1210                                 pjdlog_debug(2,
1211                                     "local_send: (%p) Moving request to the done queue.",
1212                                     hio);
1213                                 QUEUE_INSERT2(hio, done);
1214                         }
1215                 }
1216         }
1217         /* NOTREACHED */
1218         return (NULL);
1219 }
1220
1221 static void
1222 keepalive_send(struct hast_resource *res, unsigned int ncomp)
1223 {
1224         struct nv *nv;
1225
1226         if (!ISCONNECTED(res, ncomp))
1227                 return;
1228
1229         assert(res->hr_remotein != NULL);
1230         assert(res->hr_remoteout != NULL);
1231
1232         nv = nv_alloc();
1233         nv_add_uint8(nv, HIO_KEEPALIVE, "cmd");
1234         if (nv_error(nv) != 0) {
1235                 nv_free(nv);
1236                 pjdlog_debug(1,
1237                     "keepalive_send: Unable to prepare header to send.");
1238                 return;
1239         }
1240         if (hast_proto_send(res, res->hr_remoteout, nv, NULL, 0) < 0) {
1241                 pjdlog_common(LOG_DEBUG, 1, errno,
1242                     "keepalive_send: Unable to send request");
1243                 nv_free(nv);
1244                 rw_unlock(&hio_remote_lock[ncomp]);
1245                 remote_close(res, ncomp);
1246                 rw_rlock(&hio_remote_lock[ncomp]);
1247                 return;
1248         }
1249         nv_free(nv);
1250         pjdlog_debug(2, "keepalive_send: Request sent.");
1251 }
1252
1253 /*
1254  * Thread sends request to secondary node.
1255  */
1256 static void *
1257 remote_send_thread(void *arg)
1258 {
1259         struct hast_resource *res = arg;
1260         struct g_gate_ctl_io *ggio;
1261         time_t lastcheck, now;
1262         struct hio *hio;
1263         struct nv *nv;
1264         unsigned int ncomp;
1265         bool wakeup;
1266         uint64_t offset, length;
1267         uint8_t cmd;
1268         void *data;
1269
1270         /* Remote component is 1 for now. */
1271         ncomp = 1;
1272         lastcheck = time(NULL);
1273
1274         for (;;) {
1275                 pjdlog_debug(2, "remote_send: Taking request.");
1276                 QUEUE_TAKE1(hio, send, ncomp, RETRY_SLEEP);
1277                 if (hio == NULL) {
1278                         now = time(NULL);
1279                         if (lastcheck + RETRY_SLEEP <= now) {
1280                                 keepalive_send(res, ncomp);
1281                                 lastcheck = now;
1282                         }
1283                         continue;
1284                 }
1285                 pjdlog_debug(2, "remote_send: (%p) Got request.", hio);
1286                 ggio = &hio->hio_ggio;
1287                 switch (ggio->gctl_cmd) {
1288                 case BIO_READ:
1289                         cmd = HIO_READ;
1290                         data = NULL;
1291                         offset = ggio->gctl_offset;
1292                         length = ggio->gctl_length;
1293                         break;
1294                 case BIO_WRITE:
1295                         cmd = HIO_WRITE;
1296                         data = ggio->gctl_data;
1297                         offset = ggio->gctl_offset;
1298                         length = ggio->gctl_length;
1299                         break;
1300                 case BIO_DELETE:
1301                         cmd = HIO_DELETE;
1302                         data = NULL;
1303                         offset = ggio->gctl_offset;
1304                         length = ggio->gctl_length;
1305                         break;
1306                 case BIO_FLUSH:
1307                         cmd = HIO_FLUSH;
1308                         data = NULL;
1309                         offset = 0;
1310                         length = 0;
1311                         break;
1312                 default:
1313                         assert(!"invalid condition");
1314                         abort();
1315                 }
1316                 nv = nv_alloc();
1317                 nv_add_uint8(nv, cmd, "cmd");
1318                 nv_add_uint64(nv, (uint64_t)ggio->gctl_seq, "seq");
1319                 nv_add_uint64(nv, offset, "offset");
1320                 nv_add_uint64(nv, length, "length");
1321                 if (nv_error(nv) != 0) {
1322                         hio->hio_errors[ncomp] = nv_error(nv);
1323                         pjdlog_debug(2,
1324                             "remote_send: (%p) Unable to prepare header to send.",
1325                             hio);
1326                         reqlog(LOG_ERR, 0, ggio,
1327                             "Unable to prepare header to send (%s): ",
1328                             strerror(nv_error(nv)));
1329                         /* Move failed request immediately to the done queue. */
1330                         goto done_queue;
1331                 }
1332                 pjdlog_debug(2,
1333                     "remote_send: (%p) Moving request to the recv queue.",
1334                     hio);
1335                 /*
1336                  * Protect connection from disappearing.
1337                  */
1338                 rw_rlock(&hio_remote_lock[ncomp]);
1339                 if (!ISCONNECTED(res, ncomp)) {
1340                         rw_unlock(&hio_remote_lock[ncomp]);
1341                         hio->hio_errors[ncomp] = ENOTCONN;
1342                         goto done_queue;
1343                 }
1344                 /*
1345                  * Move the request to recv queue before sending it, because
1346                  * in different order we can get reply before we move request
1347                  * to recv queue.
1348                  */
1349                 mtx_lock(&hio_recv_list_lock[ncomp]);
1350                 wakeup = TAILQ_EMPTY(&hio_recv_list[ncomp]);
1351                 TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1352                 mtx_unlock(&hio_recv_list_lock[ncomp]);
1353                 if (hast_proto_send(res, res->hr_remoteout, nv, data,
1354                     data != NULL ? length : 0) < 0) {
1355                         hio->hio_errors[ncomp] = errno;
1356                         rw_unlock(&hio_remote_lock[ncomp]);
1357                         pjdlog_debug(2,
1358                             "remote_send: (%p) Unable to send request.", hio);
1359                         reqlog(LOG_ERR, 0, ggio,
1360                             "Unable to send request (%s): ",
1361                             strerror(hio->hio_errors[ncomp]));
1362                         remote_close(res, ncomp);
1363                         /*
1364                          * Take request back from the receive queue and move
1365                          * it immediately to the done queue.
1366                          */
1367                         mtx_lock(&hio_recv_list_lock[ncomp]);
1368                         TAILQ_REMOVE(&hio_recv_list[ncomp], hio, hio_next[ncomp]);
1369                         mtx_unlock(&hio_recv_list_lock[ncomp]);
1370                         goto done_queue;
1371                 }
1372                 rw_unlock(&hio_remote_lock[ncomp]);
1373                 nv_free(nv);
1374                 if (wakeup)
1375                         cv_signal(&hio_recv_list_cond[ncomp]);
1376                 continue;
1377 done_queue:
1378                 nv_free(nv);
1379                 if (ISSYNCREQ(hio)) {
1380                         if (!refcount_release(&hio->hio_countdown))
1381                                 continue;
1382                         mtx_lock(&sync_lock);
1383                         SYNCREQDONE(hio);
1384                         mtx_unlock(&sync_lock);
1385                         cv_signal(&sync_cond);
1386                         continue;
1387                 }
1388                 if (ggio->gctl_cmd == BIO_WRITE) {
1389                         mtx_lock(&res->hr_amp_lock);
1390                         if (activemap_need_sync(res->hr_amp, ggio->gctl_offset,
1391                             ggio->gctl_length)) {
1392                                 (void)hast_activemap_flush(res);
1393                         }
1394                         mtx_unlock(&res->hr_amp_lock);
1395                 }
1396                 if (!refcount_release(&hio->hio_countdown))
1397                         continue;
1398                 pjdlog_debug(2,
1399                     "remote_send: (%p) Moving request to the done queue.",
1400                     hio);
1401                 QUEUE_INSERT2(hio, done);
1402         }
1403         /* NOTREACHED */
1404         return (NULL);
1405 }
1406
1407 /*
1408  * Thread receives answer from secondary node and passes it to ggate_send
1409  * thread.
1410  */
1411 static void *
1412 remote_recv_thread(void *arg)
1413 {
1414         struct hast_resource *res = arg;
1415         struct g_gate_ctl_io *ggio;
1416         struct hio *hio;
1417         struct nv *nv;
1418         unsigned int ncomp;
1419         uint64_t seq;
1420         int error;
1421
1422         /* Remote component is 1 for now. */
1423         ncomp = 1;
1424
1425         for (;;) {
1426                 /* Wait until there is anything to receive. */
1427                 mtx_lock(&hio_recv_list_lock[ncomp]);
1428                 while (TAILQ_EMPTY(&hio_recv_list[ncomp])) {
1429                         pjdlog_debug(2, "remote_recv: No requests, waiting.");
1430                         cv_wait(&hio_recv_list_cond[ncomp],
1431                             &hio_recv_list_lock[ncomp]);
1432                 }
1433                 mtx_unlock(&hio_recv_list_lock[ncomp]);
1434                 rw_rlock(&hio_remote_lock[ncomp]);
1435                 if (!ISCONNECTED(res, ncomp)) {
1436                         rw_unlock(&hio_remote_lock[ncomp]);
1437                         /*
1438                          * Connection is dead, so move all pending requests to
1439                          * the done queue (one-by-one).
1440                          */
1441                         mtx_lock(&hio_recv_list_lock[ncomp]);
1442                         hio = TAILQ_FIRST(&hio_recv_list[ncomp]);
1443                         assert(hio != NULL);
1444                         TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1445                             hio_next[ncomp]);
1446                         mtx_unlock(&hio_recv_list_lock[ncomp]);
1447                         goto done_queue;
1448                 }
1449                 if (hast_proto_recv_hdr(res->hr_remotein, &nv) < 0) {
1450                         pjdlog_errno(LOG_ERR,
1451                             "Unable to receive reply header");
1452                         rw_unlock(&hio_remote_lock[ncomp]);
1453                         remote_close(res, ncomp);
1454                         continue;
1455                 }
1456                 rw_unlock(&hio_remote_lock[ncomp]);
1457                 seq = nv_get_uint64(nv, "seq");
1458                 if (seq == 0) {
1459                         pjdlog_error("Header contains no 'seq' field.");
1460                         nv_free(nv);
1461                         continue;
1462                 }
1463                 mtx_lock(&hio_recv_list_lock[ncomp]);
1464                 TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) {
1465                         if (hio->hio_ggio.gctl_seq == seq) {
1466                                 TAILQ_REMOVE(&hio_recv_list[ncomp], hio,
1467                                     hio_next[ncomp]);
1468                                 break;
1469                         }
1470                 }
1471                 mtx_unlock(&hio_recv_list_lock[ncomp]);
1472                 if (hio == NULL) {
1473                         pjdlog_error("Found no request matching received 'seq' field (%ju).",
1474                             (uintmax_t)seq);
1475                         nv_free(nv);
1476                         continue;
1477                 }
1478                 error = nv_get_int16(nv, "error");
1479                 if (error != 0) {
1480                         /* Request failed on remote side. */
1481                         hio->hio_errors[ncomp] = error;
1482                         reqlog(LOG_WARNING, 0, &hio->hio_ggio,
1483                             "Remote request failed (%s): ", strerror(error));
1484                         nv_free(nv);
1485                         goto done_queue;
1486                 }
1487                 ggio = &hio->hio_ggio;
1488                 switch (ggio->gctl_cmd) {
1489                 case BIO_READ:
1490                         rw_rlock(&hio_remote_lock[ncomp]);
1491                         if (!ISCONNECTED(res, ncomp)) {
1492                                 rw_unlock(&hio_remote_lock[ncomp]);
1493                                 nv_free(nv);
1494                                 goto done_queue;
1495                         }
1496                         if (hast_proto_recv_data(res, res->hr_remotein, nv,
1497                             ggio->gctl_data, ggio->gctl_length) < 0) {
1498                                 hio->hio_errors[ncomp] = errno;
1499                                 pjdlog_errno(LOG_ERR,
1500                                     "Unable to receive reply data");
1501                                 rw_unlock(&hio_remote_lock[ncomp]);
1502                                 nv_free(nv);
1503                                 remote_close(res, ncomp);
1504                                 goto done_queue;
1505                         }
1506                         rw_unlock(&hio_remote_lock[ncomp]);
1507                         break;
1508                 case BIO_WRITE:
1509                 case BIO_DELETE:
1510                 case BIO_FLUSH:
1511                         break;
1512                 default:
1513                         assert(!"invalid condition");
1514                         abort();
1515                 }
1516                 hio->hio_errors[ncomp] = 0;
1517                 nv_free(nv);
1518 done_queue:
1519                 if (refcount_release(&hio->hio_countdown)) {
1520                         if (ISSYNCREQ(hio)) {
1521                                 mtx_lock(&sync_lock);
1522                                 SYNCREQDONE(hio);
1523                                 mtx_unlock(&sync_lock);
1524                                 cv_signal(&sync_cond);
1525                         } else {
1526                                 pjdlog_debug(2,
1527                                     "remote_recv: (%p) Moving request to the done queue.",
1528                                     hio);
1529                                 QUEUE_INSERT2(hio, done);
1530                         }
1531                 }
1532         }
1533         /* NOTREACHED */
1534         return (NULL);
1535 }
1536
1537 /*
1538  * Thread sends answer to the kernel.
1539  */
1540 static void *
1541 ggate_send_thread(void *arg)
1542 {
1543         struct hast_resource *res = arg;
1544         struct g_gate_ctl_io *ggio;
1545         struct hio *hio;
1546         unsigned int ii, ncomp, ncomps;
1547
1548         ncomps = HAST_NCOMPONENTS;
1549
1550         for (;;) {
1551                 pjdlog_debug(2, "ggate_send: Taking request.");
1552                 QUEUE_TAKE2(hio, done);
1553                 pjdlog_debug(2, "ggate_send: (%p) Got request.", hio);
1554                 ggio = &hio->hio_ggio;
1555                 for (ii = 0; ii < ncomps; ii++) {
1556                         if (hio->hio_errors[ii] == 0) {
1557                                 /*
1558                                  * One successful request is enough to declare
1559                                  * success.
1560                                  */
1561                                 ggio->gctl_error = 0;
1562                                 break;
1563                         }
1564                 }
1565                 if (ii == ncomps) {
1566                         /*
1567                          * None of the requests were successful.
1568                          * Use first error.
1569                          */
1570                         ggio->gctl_error = hio->hio_errors[0];
1571                 }
1572                 if (ggio->gctl_error == 0 && ggio->gctl_cmd == BIO_WRITE) {
1573                         mtx_lock(&res->hr_amp_lock);
1574                         activemap_write_complete(res->hr_amp,
1575                             ggio->gctl_offset, ggio->gctl_length);
1576                         mtx_unlock(&res->hr_amp_lock);
1577                 }
1578                 if (ggio->gctl_cmd == BIO_WRITE) {
1579                         /*
1580                          * Unlock range we locked.
1581                          */
1582                         mtx_lock(&range_lock);
1583                         rangelock_del(range_regular, ggio->gctl_offset,
1584                             ggio->gctl_length);
1585                         if (range_sync_wait)
1586                                 cv_signal(&range_sync_cond);
1587                         mtx_unlock(&range_lock);
1588                         /*
1589                          * Bump local count if this is first write after
1590                          * connection failure with remote node.
1591                          */
1592                         ncomp = 1;
1593                         rw_rlock(&hio_remote_lock[ncomp]);
1594                         if (!ISCONNECTED(res, ncomp)) {
1595                                 mtx_lock(&metadata_lock);
1596                                 if (res->hr_primary_localcnt ==
1597                                     res->hr_secondary_remotecnt) {
1598                                         res->hr_primary_localcnt++;
1599                                         pjdlog_debug(1,
1600                                             "Increasing localcnt to %ju.",
1601                                             (uintmax_t)res->hr_primary_localcnt);
1602                                         (void)metadata_write(res);
1603                                 }
1604                                 mtx_unlock(&metadata_lock);
1605                         }
1606                         rw_unlock(&hio_remote_lock[ncomp]);
1607                 }
1608                 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DONE, ggio) < 0)
1609                         primary_exit(EX_OSERR, "G_GATE_CMD_DONE failed");
1610                 pjdlog_debug(2,
1611                     "ggate_send: (%p) Moving request to the free queue.", hio);
1612                 QUEUE_INSERT2(hio, free);
1613         }
1614         /* NOTREACHED */
1615         return (NULL);
1616 }
1617
1618 /*
1619  * Thread synchronize local and remote components.
1620  */
1621 static void *
1622 sync_thread(void *arg __unused)
1623 {
1624         struct hast_resource *res = arg;
1625         struct hio *hio;
1626         struct g_gate_ctl_io *ggio;
1627         unsigned int ii, ncomp, ncomps;
1628         off_t offset, length, synced;
1629         bool dorewind;
1630         int syncext;
1631
1632         ncomps = HAST_NCOMPONENTS;
1633         dorewind = true;
1634         synced = 0;
1635         offset = -1;
1636
1637         for (;;) {
1638                 mtx_lock(&sync_lock);
1639                 if (offset >= 0 && !sync_inprogress) {
1640                         pjdlog_info("Synchronization interrupted. "
1641                             "%jd bytes synchronized so far.",
1642                             (intmax_t)synced);
1643                         event_send(res, EVENT_SYNCINTR);
1644                 }
1645                 while (!sync_inprogress) {
1646                         dorewind = true;
1647                         synced = 0;
1648                         cv_wait(&sync_cond, &sync_lock);
1649                 }
1650                 mtx_unlock(&sync_lock);
1651                 /*
1652                  * Obtain offset at which we should synchronize.
1653                  * Rewind synchronization if needed.
1654                  */
1655                 mtx_lock(&res->hr_amp_lock);
1656                 if (dorewind)
1657                         activemap_sync_rewind(res->hr_amp);
1658                 offset = activemap_sync_offset(res->hr_amp, &length, &syncext);
1659                 if (syncext != -1) {
1660                         /*
1661                          * We synchronized entire syncext extent, we can mark
1662                          * it as clean now.
1663                          */
1664                         if (activemap_extent_complete(res->hr_amp, syncext))
1665                                 (void)hast_activemap_flush(res);
1666                 }
1667                 mtx_unlock(&res->hr_amp_lock);
1668                 if (dorewind) {
1669                         dorewind = false;
1670                         if (offset < 0)
1671                                 pjdlog_info("Nodes are in sync.");
1672                         else {
1673                                 pjdlog_info("Synchronization started. %ju bytes to go.",
1674                                     (uintmax_t)(res->hr_extentsize *
1675                                     activemap_ndirty(res->hr_amp)));
1676                                 event_send(res, EVENT_SYNCSTART);
1677                         }
1678                 }
1679                 if (offset < 0) {
1680                         sync_stop();
1681                         pjdlog_debug(1, "Nothing to synchronize.");
1682                         /*
1683                          * Synchronization complete, make both localcnt and
1684                          * remotecnt equal.
1685                          */
1686                         ncomp = 1;
1687                         rw_rlock(&hio_remote_lock[ncomp]);
1688                         if (ISCONNECTED(res, ncomp)) {
1689                                 if (synced > 0) {
1690                                         pjdlog_info("Synchronization complete. "
1691                                             "%jd bytes synchronized.",
1692                                             (intmax_t)synced);
1693                                         event_send(res, EVENT_SYNCDONE);
1694                                 }
1695                                 mtx_lock(&metadata_lock);
1696                                 res->hr_syncsrc = HAST_SYNCSRC_UNDEF;
1697                                 res->hr_primary_localcnt =
1698                                     res->hr_secondary_localcnt;
1699                                 res->hr_primary_remotecnt =
1700                                     res->hr_secondary_remotecnt;
1701                                 pjdlog_debug(1,
1702                                     "Setting localcnt to %ju and remotecnt to %ju.",
1703                                     (uintmax_t)res->hr_primary_localcnt,
1704                                     (uintmax_t)res->hr_secondary_localcnt);
1705                                 (void)metadata_write(res);
1706                                 mtx_unlock(&metadata_lock);
1707                         }
1708                         rw_unlock(&hio_remote_lock[ncomp]);
1709                         continue;
1710                 }
1711                 pjdlog_debug(2, "sync: Taking free request.");
1712                 QUEUE_TAKE2(hio, free);
1713                 pjdlog_debug(2, "sync: (%p) Got free request.", hio);
1714                 /*
1715                  * Lock the range we are going to synchronize. We don't want
1716                  * race where someone writes between our read and write.
1717                  */
1718                 for (;;) {
1719                         mtx_lock(&range_lock);
1720                         if (rangelock_islocked(range_regular, offset, length)) {
1721                                 pjdlog_debug(2,
1722                                     "sync: Range offset=%jd length=%jd locked.",
1723                                     (intmax_t)offset, (intmax_t)length);
1724                                 range_sync_wait = true;
1725                                 cv_wait(&range_sync_cond, &range_lock);
1726                                 range_sync_wait = false;
1727                                 mtx_unlock(&range_lock);
1728                                 continue;
1729                         }
1730                         if (rangelock_add(range_sync, offset, length) < 0) {
1731                                 mtx_unlock(&range_lock);
1732                                 pjdlog_debug(2,
1733                                     "sync: Range offset=%jd length=%jd is already locked, waiting.",
1734                                     (intmax_t)offset, (intmax_t)length);
1735                                 sleep(1);
1736                                 continue;
1737                         }
1738                         mtx_unlock(&range_lock);
1739                         break;
1740                 }
1741                 /*
1742                  * First read the data from synchronization source.
1743                  */
1744                 SYNCREQ(hio);
1745                 ggio = &hio->hio_ggio;
1746                 ggio->gctl_cmd = BIO_READ;
1747                 ggio->gctl_offset = offset;
1748                 ggio->gctl_length = length;
1749                 ggio->gctl_error = 0;
1750                 for (ii = 0; ii < ncomps; ii++)
1751                         hio->hio_errors[ii] = EINVAL;
1752                 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1753                     hio);
1754                 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1755                     hio);
1756                 mtx_lock(&metadata_lock);
1757                 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1758                         /*
1759                          * This range is up-to-date on local component,
1760                          * so handle request locally.
1761                          */
1762                          /* Local component is 0 for now. */
1763                         ncomp = 0;
1764                 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1765                         assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1766                         /*
1767                          * This range is out-of-date on local component,
1768                          * so send request to the remote node.
1769                          */
1770                          /* Remote component is 1 for now. */
1771                         ncomp = 1;
1772                 }
1773                 mtx_unlock(&metadata_lock);
1774                 refcount_init(&hio->hio_countdown, 1);
1775                 QUEUE_INSERT1(hio, send, ncomp);
1776
1777                 /*
1778                  * Let's wait for READ to finish.
1779                  */
1780                 mtx_lock(&sync_lock);
1781                 while (!ISSYNCREQDONE(hio))
1782                         cv_wait(&sync_cond, &sync_lock);
1783                 mtx_unlock(&sync_lock);
1784
1785                 if (hio->hio_errors[ncomp] != 0) {
1786                         pjdlog_error("Unable to read synchronization data: %s.",
1787                             strerror(hio->hio_errors[ncomp]));
1788                         goto free_queue;
1789                 }
1790
1791                 /*
1792                  * We read the data from synchronization source, now write it
1793                  * to synchronization target.
1794                  */
1795                 SYNCREQ(hio);
1796                 ggio->gctl_cmd = BIO_WRITE;
1797                 for (ii = 0; ii < ncomps; ii++)
1798                         hio->hio_errors[ii] = EINVAL;
1799                 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ",
1800                     hio);
1801                 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.",
1802                     hio);
1803                 mtx_lock(&metadata_lock);
1804                 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) {
1805                         /*
1806                          * This range is up-to-date on local component,
1807                          * so we update remote component.
1808                          */
1809                          /* Remote component is 1 for now. */
1810                         ncomp = 1;
1811                 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ {
1812                         assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY);
1813                         /*
1814                          * This range is out-of-date on local component,
1815                          * so we update it.
1816                          */
1817                          /* Local component is 0 for now. */
1818                         ncomp = 0;
1819                 }
1820                 mtx_unlock(&metadata_lock);
1821
1822                 pjdlog_debug(2, "sync: (%p) Moving request to the send queues.",
1823                     hio);
1824                 refcount_init(&hio->hio_countdown, 1);
1825                 QUEUE_INSERT1(hio, send, ncomp);
1826
1827                 /*
1828                  * Let's wait for WRITE to finish.
1829                  */
1830                 mtx_lock(&sync_lock);
1831                 while (!ISSYNCREQDONE(hio))
1832                         cv_wait(&sync_cond, &sync_lock);
1833                 mtx_unlock(&sync_lock);
1834
1835                 if (hio->hio_errors[ncomp] != 0) {
1836                         pjdlog_error("Unable to write synchronization data: %s.",
1837                             strerror(hio->hio_errors[ncomp]));
1838                         goto free_queue;
1839                 }
1840
1841                 synced += length;
1842 free_queue:
1843                 mtx_lock(&range_lock);
1844                 rangelock_del(range_sync, offset, length);
1845                 if (range_regular_wait)
1846                         cv_signal(&range_regular_cond);
1847                 mtx_unlock(&range_lock);
1848                 pjdlog_debug(2, "sync: (%p) Moving request to the free queue.",
1849                     hio);
1850                 QUEUE_INSERT2(hio, free);
1851         }
1852         /* NOTREACHED */
1853         return (NULL);
1854 }
1855
1856 void
1857 primary_config_reload(struct hast_resource *res, struct nv *nv)
1858 {
1859         unsigned int ii, ncomps;
1860         int modified, vint;
1861         const char *vstr;
1862
1863         pjdlog_info("Reloading configuration...");
1864
1865         assert(res->hr_role == HAST_ROLE_PRIMARY);
1866         assert(gres == res);
1867         nv_assert(nv, "remoteaddr");
1868         nv_assert(nv, "replication");
1869         nv_assert(nv, "timeout");
1870         nv_assert(nv, "exec");
1871
1872         ncomps = HAST_NCOMPONENTS;
1873
1874 #define MODIFIED_REMOTEADDR     0x1
1875 #define MODIFIED_REPLICATION    0x2
1876 #define MODIFIED_TIMEOUT        0x4
1877 #define MODIFIED_EXEC           0x8
1878         modified = 0;
1879
1880         vstr = nv_get_string(nv, "remoteaddr");
1881         if (strcmp(gres->hr_remoteaddr, vstr) != 0) {
1882                 /*
1883                  * Don't copy res->hr_remoteaddr to gres just yet.
1884                  * We want remote_close() to log disconnect from the old
1885                  * addresses, not from the new ones.
1886                  */
1887                 modified |= MODIFIED_REMOTEADDR;
1888         }
1889         vint = nv_get_int32(nv, "replication");
1890         if (gres->hr_replication != vint) {
1891                 gres->hr_replication = vint;
1892                 modified |= MODIFIED_REPLICATION;
1893         }
1894         vint = nv_get_int32(nv, "timeout");
1895         if (gres->hr_timeout != vint) {
1896                 gres->hr_timeout = vint;
1897                 modified |= MODIFIED_TIMEOUT;
1898         }
1899         vstr = nv_get_string(nv, "exec");
1900         if (strcmp(gres->hr_exec, vstr) != 0) {
1901                 strlcpy(gres->hr_exec, vstr, sizeof(gres->hr_exec));
1902                 modified |= MODIFIED_EXEC;
1903         }
1904
1905         /*
1906          * If only timeout was modified we only need to change it without
1907          * reconnecting.
1908          */
1909         if (modified == MODIFIED_TIMEOUT) {
1910                 for (ii = 0; ii < ncomps; ii++) {
1911                         if (!ISREMOTE(ii))
1912                                 continue;
1913                         rw_rlock(&hio_remote_lock[ii]);
1914                         if (!ISCONNECTED(gres, ii)) {
1915                                 rw_unlock(&hio_remote_lock[ii]);
1916                                 continue;
1917                         }
1918                         rw_unlock(&hio_remote_lock[ii]);
1919                         if (proto_timeout(gres->hr_remotein,
1920                             gres->hr_timeout) < 0) {
1921                                 pjdlog_errno(LOG_WARNING,
1922                                     "Unable to set connection timeout");
1923                         }
1924                         if (proto_timeout(gres->hr_remoteout,
1925                             gres->hr_timeout) < 0) {
1926                                 pjdlog_errno(LOG_WARNING,
1927                                     "Unable to set connection timeout");
1928                         }
1929                 }
1930         } else if ((modified &
1931             (MODIFIED_REMOTEADDR | MODIFIED_REPLICATION)) != 0) {
1932                 for (ii = 0; ii < ncomps; ii++) {
1933                         if (!ISREMOTE(ii))
1934                                 continue;
1935                         remote_close(gres, ii);
1936                 }
1937                 if (modified & MODIFIED_REMOTEADDR) {
1938                         vstr = nv_get_string(nv, "remoteaddr");
1939                         strlcpy(gres->hr_remoteaddr, vstr,
1940                             sizeof(gres->hr_remoteaddr));
1941                 }
1942         }
1943 #undef  MODIFIED_REMOTEADDR
1944 #undef  MODIFIED_REPLICATION
1945 #undef  MODIFIED_TIMEOUT
1946 #undef  MODIFIED_EXEC
1947
1948         pjdlog_info("Configuration reloaded successfully.");
1949 }
1950
1951 static void
1952 guard_one(struct hast_resource *res, unsigned int ncomp)
1953 {
1954         struct proto_conn *in, *out;
1955
1956         if (!ISREMOTE(ncomp))
1957                 return;
1958
1959         rw_rlock(&hio_remote_lock[ncomp]);
1960
1961         if (!real_remote(res)) {
1962                 rw_unlock(&hio_remote_lock[ncomp]);
1963                 return;
1964         }
1965
1966         if (ISCONNECTED(res, ncomp)) {
1967                 assert(res->hr_remotein != NULL);
1968                 assert(res->hr_remoteout != NULL);
1969                 rw_unlock(&hio_remote_lock[ncomp]);
1970                 pjdlog_debug(2, "remote_guard: Connection to %s is ok.",
1971                     res->hr_remoteaddr);
1972                 return;
1973         }
1974
1975         assert(res->hr_remotein == NULL);
1976         assert(res->hr_remoteout == NULL);
1977         /*
1978          * Upgrade the lock. It doesn't have to be atomic as no other thread
1979          * can change connection status from disconnected to connected.
1980          */
1981         rw_unlock(&hio_remote_lock[ncomp]);
1982         pjdlog_debug(2, "remote_guard: Reconnecting to %s.",
1983             res->hr_remoteaddr);
1984         in = out = NULL;
1985         if (init_remote(res, &in, &out)) {
1986                 rw_wlock(&hio_remote_lock[ncomp]);
1987                 assert(res->hr_remotein == NULL);
1988                 assert(res->hr_remoteout == NULL);
1989                 assert(in != NULL && out != NULL);
1990                 res->hr_remotein = in;
1991                 res->hr_remoteout = out;
1992                 rw_unlock(&hio_remote_lock[ncomp]);
1993                 pjdlog_info("Successfully reconnected to %s.",
1994                     res->hr_remoteaddr);
1995                 sync_start();
1996         } else {
1997                 /* Both connections should be NULL. */
1998                 assert(res->hr_remotein == NULL);
1999                 assert(res->hr_remoteout == NULL);
2000                 assert(in == NULL && out == NULL);
2001                 pjdlog_debug(2, "remote_guard: Reconnect to %s failed.",
2002                     res->hr_remoteaddr);
2003         }
2004 }
2005
2006 /*
2007  * Thread guards remote connections and reconnects when needed, handles
2008  * signals, etc.
2009  */
2010 static void *
2011 guard_thread(void *arg)
2012 {
2013         struct hast_resource *res = arg;
2014         unsigned int ii, ncomps;
2015         struct timespec timeout;
2016         time_t lastcheck, now;
2017         sigset_t mask;
2018         int signo;
2019
2020         ncomps = HAST_NCOMPONENTS;
2021         lastcheck = time(NULL);
2022
2023         PJDLOG_VERIFY(sigemptyset(&mask) == 0);
2024         PJDLOG_VERIFY(sigaddset(&mask, SIGINT) == 0);
2025         PJDLOG_VERIFY(sigaddset(&mask, SIGTERM) == 0);
2026
2027         timeout.tv_sec = RETRY_SLEEP;
2028         timeout.tv_nsec = 0;
2029         signo = -1;
2030
2031         for (;;) {
2032                 switch (signo) {
2033                 case SIGINT:
2034                 case SIGTERM:
2035                         sigexit_received = true;
2036                         primary_exitx(EX_OK,
2037                             "Termination signal received, exiting.");
2038                         break;
2039                 default:
2040                         break;
2041                 }
2042
2043                 pjdlog_debug(2, "remote_guard: Checking connections.");
2044                 now = time(NULL);
2045                 if (lastcheck + RETRY_SLEEP <= now) {
2046                         for (ii = 0; ii < ncomps; ii++)
2047                                 guard_one(res, ii);
2048                         lastcheck = now;
2049                 }
2050                 signo = sigtimedwait(&mask, NULL, &timeout);
2051         }
2052         /* NOTREACHED */
2053         return (NULL);
2054 }