sbin/hastd/primary.c

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