primary.c revision 207371
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: head/sbin/hastd/primary.c 207371 2010-04-29 15:36:32Z pjd $"); 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 68struct 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 */ 95static TAILQ_HEAD(, hio) hio_free_list; 96static pthread_mutex_t hio_free_list_lock; 97static 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 */ 103static TAILQ_HEAD(, hio) *hio_send_list; 104static pthread_mutex_t *hio_send_list_lock; 105static 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 */ 110static TAILQ_HEAD(, hio) *hio_recv_list; 111static pthread_mutex_t *hio_recv_list_lock; 112static 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 */ 117static TAILQ_HEAD(, hio) hio_done_list; 118static pthread_mutex_t hio_done_list_lock; 119static pthread_cond_t hio_done_list_cond; 120/* 121 * Structure below are for interaction with sync thread. 122 */ 123static bool sync_inprogress; 124static pthread_mutex_t sync_lock; 125static pthread_cond_t sync_cond; 126/* 127 * The lock below allows to synchornize access to remote connections. 128 */ 129static pthread_rwlock_t *hio_remote_lock; 130static pthread_mutex_t hio_guard_lock; 131static 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 */ 137static 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 203static struct hast_resource *gres; 204 205static pthread_mutex_t range_lock; 206static struct rangelocks *range_regular; 207static bool range_regular_wait; 208static pthread_cond_t range_regular_cond; 209static struct rangelocks *range_sync; 210static bool range_sync_wait; 211static pthread_cond_t range_sync_cond; 212 213static void *ggate_recv_thread(void *arg); 214static void *local_send_thread(void *arg); 215static void *remote_send_thread(void *arg); 216static void *remote_recv_thread(void *arg); 217static void *ggate_send_thread(void *arg); 218static void *sync_thread(void *arg); 219static void *guard_thread(void *arg); 220 221static void sighandler(int sig); 222 223static void 224cleanup(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 255static void 256primary_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 268static void 269primary_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 280static int 281hast_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 298static void 299init_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 419static void 420init_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 463static bool 464init_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); 654close: 655 proto_close(out); 656 if (in != NULL) 657 proto_close(in); 658 return (false); 659} 660 661static void 662sync_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 671static void 672init_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 727void 728hastd_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 782static void 783reqlog(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 821static void 822remote_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 */ 878static void * 879ggate_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 */ 1022static void * 1023local_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 */ 1105static void * 1106remote_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; 1216done_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 */ 1250static void * 1251remote_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); 1355done_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 */ 1377static void * 1378ggate_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 */ 1458static void * 1459sync_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 } 1674free_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 1691static void 1692sighandler(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 */ 1718static void * 1719guard_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} 1800