primary.c revision 204076
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 204076 2010-02-18 23:16:19Z 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 void 464init_remote(struct hast_resource *res) 465{ 466 struct nv *nvout, *nvin; 467 const unsigned char *token; 468 unsigned char *map; 469 const char *errmsg; 470 int32_t extentsize; 471 int64_t datasize; 472 uint32_t mapsize; 473 size_t size; 474 475 /* Prepare outgoing connection with remote node. */ 476 if (proto_client(res->hr_remoteaddr, &res->hr_remoteout) < 0) { 477 primary_exit(EX_OSERR, "Unable to create connection to %s", 478 res->hr_remoteaddr); 479 } 480 /* Try to connect, but accept failure. */ 481 if (proto_connect(res->hr_remoteout) < 0) { 482 pjdlog_errno(LOG_WARNING, "Unable to connect to %s", 483 res->hr_remoteaddr); 484 goto close; 485 } 486 /* 487 * First handshake step. 488 * Setup outgoing connection with remote node. 489 */ 490 nvout = nv_alloc(); 491 nv_add_string(nvout, res->hr_name, "resource"); 492 if (nv_error(nvout) != 0) { 493 pjdlog_common(LOG_WARNING, 0, nv_error(nvout), 494 "Unable to allocate header for connection with %s", 495 res->hr_remoteaddr); 496 nv_free(nvout); 497 goto close; 498 } 499 if (hast_proto_send(res, res->hr_remoteout, nvout, NULL, 0) < 0) { 500 pjdlog_errno(LOG_WARNING, 501 "Unable to send handshake header to %s", 502 res->hr_remoteaddr); 503 nv_free(nvout); 504 goto close; 505 } 506 nv_free(nvout); 507 if (hast_proto_recv_hdr(res->hr_remoteout, &nvin) < 0) { 508 pjdlog_errno(LOG_WARNING, 509 "Unable to receive handshake header from %s", 510 res->hr_remoteaddr); 511 goto close; 512 } 513 errmsg = nv_get_string(nvin, "errmsg"); 514 if (errmsg != NULL) { 515 pjdlog_warning("%s", errmsg); 516 nv_free(nvin); 517 goto close; 518 } 519 token = nv_get_uint8_array(nvin, &size, "token"); 520 if (token == NULL) { 521 pjdlog_warning("Handshake header from %s has no 'token' field.", 522 res->hr_remoteaddr); 523 nv_free(nvin); 524 goto close; 525 } 526 if (size != sizeof(res->hr_token)) { 527 pjdlog_warning("Handshake header from %s contains 'token' of wrong size (got %zu, expected %zu).", 528 res->hr_remoteaddr, size, sizeof(res->hr_token)); 529 nv_free(nvin); 530 goto close; 531 } 532 bcopy(token, res->hr_token, sizeof(res->hr_token)); 533 nv_free(nvin); 534 535 /* 536 * Second handshake step. 537 * Setup incoming connection with remote node. 538 */ 539 if (proto_client(res->hr_remoteaddr, &res->hr_remotein) < 0) { 540 pjdlog_errno(LOG_WARNING, "Unable to create connection to %s", 541 res->hr_remoteaddr); 542 } 543 /* Try to connect, but accept failure. */ 544 if (proto_connect(res->hr_remotein) < 0) { 545 pjdlog_errno(LOG_WARNING, "Unable to connect to %s", 546 res->hr_remoteaddr); 547 goto close; 548 } 549 nvout = nv_alloc(); 550 nv_add_string(nvout, res->hr_name, "resource"); 551 nv_add_uint8_array(nvout, res->hr_token, sizeof(res->hr_token), 552 "token"); 553 nv_add_uint64(nvout, res->hr_resuid, "resuid"); 554 nv_add_uint64(nvout, res->hr_primary_localcnt, "localcnt"); 555 nv_add_uint64(nvout, res->hr_primary_remotecnt, "remotecnt"); 556 if (nv_error(nvout) != 0) { 557 pjdlog_common(LOG_WARNING, 0, nv_error(nvout), 558 "Unable to allocate header for connection with %s", 559 res->hr_remoteaddr); 560 nv_free(nvout); 561 goto close; 562 } 563 if (hast_proto_send(res, res->hr_remotein, nvout, NULL, 0) < 0) { 564 pjdlog_errno(LOG_WARNING, 565 "Unable to send handshake header to %s", 566 res->hr_remoteaddr); 567 nv_free(nvout); 568 goto close; 569 } 570 nv_free(nvout); 571 if (hast_proto_recv_hdr(res->hr_remoteout, &nvin) < 0) { 572 pjdlog_errno(LOG_WARNING, 573 "Unable to receive handshake header from %s", 574 res->hr_remoteaddr); 575 goto close; 576 } 577 errmsg = nv_get_string(nvin, "errmsg"); 578 if (errmsg != NULL) { 579 pjdlog_warning("%s", errmsg); 580 nv_free(nvin); 581 goto close; 582 } 583 datasize = nv_get_int64(nvin, "datasize"); 584 if (datasize != res->hr_datasize) { 585 pjdlog_warning("Data size differs between nodes (local=%jd, remote=%jd).", 586 (intmax_t)res->hr_datasize, (intmax_t)datasize); 587 nv_free(nvin); 588 goto close; 589 } 590 extentsize = nv_get_int32(nvin, "extentsize"); 591 if (extentsize != res->hr_extentsize) { 592 pjdlog_warning("Extent size differs between nodes (local=%zd, remote=%zd).", 593 (ssize_t)res->hr_extentsize, (ssize_t)extentsize); 594 nv_free(nvin); 595 goto close; 596 } 597 res->hr_secondary_localcnt = nv_get_uint64(nvin, "localcnt"); 598 res->hr_secondary_remotecnt = nv_get_uint64(nvin, "remotecnt"); 599 res->hr_syncsrc = nv_get_uint8(nvin, "syncsrc"); 600 map = NULL; 601 mapsize = nv_get_uint32(nvin, "mapsize"); 602 if (mapsize > 0) { 603 map = malloc(mapsize); 604 if (map == NULL) { 605 pjdlog_error("Unable to allocate memory for remote activemap (mapsize=%ju).", 606 (uintmax_t)mapsize); 607 nv_free(nvin); 608 goto close; 609 } 610 /* 611 * Remote node have some dirty extents on its own, lets 612 * download its activemap. 613 */ 614 if (hast_proto_recv_data(res, res->hr_remoteout, nvin, map, 615 mapsize) < 0) { 616 pjdlog_errno(LOG_ERR, 617 "Unable to receive remote activemap"); 618 nv_free(nvin); 619 free(map); 620 goto close; 621 } 622 /* 623 * Merge local and remote bitmaps. 624 */ 625 activemap_merge(res->hr_amp, map, mapsize); 626 free(map); 627 /* 628 * Now that we merged bitmaps from both nodes, flush it to the 629 * disk before we start to synchronize. 630 */ 631 (void)hast_activemap_flush(res); 632 } 633 pjdlog_info("Connected to %s.", res->hr_remoteaddr); 634 mtx_lock(&sync_lock); 635 sync_inprogress = true; 636 mtx_unlock(&sync_lock); 637 cv_signal(&sync_cond); 638 return; 639close: 640 proto_close(res->hr_remoteout); 641 res->hr_remoteout = NULL; 642 if (res->hr_remotein != NULL) { 643 proto_close(res->hr_remotein); 644 res->hr_remotein = NULL; 645 } 646} 647 648static void 649init_ggate(struct hast_resource *res) 650{ 651 struct g_gate_ctl_create ggiocreate; 652 struct g_gate_ctl_cancel ggiocancel; 653 654 /* 655 * We communicate with ggate via /dev/ggctl. Open it. 656 */ 657 res->hr_ggatefd = open("/dev/" G_GATE_CTL_NAME, O_RDWR); 658 if (res->hr_ggatefd < 0) 659 primary_exit(EX_OSFILE, "Unable to open /dev/" G_GATE_CTL_NAME); 660 /* 661 * Create provider before trying to connect, as connection failure 662 * is not critical, but may take some time. 663 */ 664 ggiocreate.gctl_version = G_GATE_VERSION; 665 ggiocreate.gctl_mediasize = res->hr_datasize; 666 ggiocreate.gctl_sectorsize = res->hr_local_sectorsize; 667 ggiocreate.gctl_flags = 0; 668 ggiocreate.gctl_maxcount = 128; 669 ggiocreate.gctl_timeout = 0; 670 ggiocreate.gctl_unit = G_GATE_NAME_GIVEN; 671 snprintf(ggiocreate.gctl_name, sizeof(ggiocreate.gctl_name), "hast/%s", 672 res->hr_provname); 673 bzero(ggiocreate.gctl_info, sizeof(ggiocreate.gctl_info)); 674 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CREATE, &ggiocreate) == 0) { 675 pjdlog_info("Device hast/%s created.", res->hr_provname); 676 res->hr_ggateunit = ggiocreate.gctl_unit; 677 return; 678 } 679 if (errno != EEXIST) { 680 primary_exit(EX_OSERR, "Unable to create hast/%s device", 681 res->hr_provname); 682 } 683 pjdlog_debug(1, 684 "Device hast/%s already exists, we will try to take it over.", 685 res->hr_provname); 686 /* 687 * If we received EEXIST, we assume that the process who created the 688 * provider died and didn't clean up. In that case we will start from 689 * where he left of. 690 */ 691 ggiocancel.gctl_version = G_GATE_VERSION; 692 ggiocancel.gctl_unit = G_GATE_NAME_GIVEN; 693 snprintf(ggiocancel.gctl_name, sizeof(ggiocancel.gctl_name), "hast/%s", 694 res->hr_provname); 695 if (ioctl(res->hr_ggatefd, G_GATE_CMD_CANCEL, &ggiocancel) == 0) { 696 pjdlog_info("Device hast/%s recovered.", res->hr_provname); 697 res->hr_ggateunit = ggiocancel.gctl_unit; 698 return; 699 } 700 primary_exit(EX_OSERR, "Unable to take over hast/%s device", 701 res->hr_provname); 702} 703 704void 705hastd_primary(struct hast_resource *res) 706{ 707 pthread_t td; 708 pid_t pid; 709 int error; 710 711 gres = res; 712 713 /* 714 * Create communication channel between parent and child. 715 */ 716 if (proto_client("socketpair://", &res->hr_ctrl) < 0) { 717 KEEP_ERRNO((void)pidfile_remove(pfh)); 718 primary_exit(EX_OSERR, 719 "Unable to create control sockets between parent and child"); 720 } 721 722 pid = fork(); 723 if (pid < 0) { 724 KEEP_ERRNO((void)pidfile_remove(pfh)); 725 primary_exit(EX_OSERR, "Unable to fork"); 726 } 727 728 if (pid > 0) { 729 /* This is parent. */ 730 res->hr_workerpid = pid; 731 return; 732 } 733 (void)pidfile_close(pfh); 734 735 setproctitle("%s (primary)", res->hr_name); 736 737 init_local(res); 738 init_remote(res); 739 init_ggate(res); 740 init_environment(res); 741 error = pthread_create(&td, NULL, ggate_recv_thread, res); 742 assert(error == 0); 743 error = pthread_create(&td, NULL, local_send_thread, res); 744 assert(error == 0); 745 error = pthread_create(&td, NULL, remote_send_thread, res); 746 assert(error == 0); 747 error = pthread_create(&td, NULL, remote_recv_thread, res); 748 assert(error == 0); 749 error = pthread_create(&td, NULL, ggate_send_thread, res); 750 assert(error == 0); 751 error = pthread_create(&td, NULL, sync_thread, res); 752 assert(error == 0); 753 error = pthread_create(&td, NULL, ctrl_thread, res); 754 assert(error == 0); 755 (void)guard_thread(res); 756} 757 758static void 759reqlog(int loglevel, int debuglevel, struct g_gate_ctl_io *ggio, const char *fmt, ...) 760{ 761 char msg[1024]; 762 va_list ap; 763 int len; 764 765 va_start(ap, fmt); 766 len = vsnprintf(msg, sizeof(msg), fmt, ap); 767 va_end(ap); 768 if ((size_t)len < sizeof(msg)) { 769 switch (ggio->gctl_cmd) { 770 case BIO_READ: 771 (void)snprintf(msg + len, sizeof(msg) - len, 772 "READ(%ju, %ju).", (uintmax_t)ggio->gctl_offset, 773 (uintmax_t)ggio->gctl_length); 774 break; 775 case BIO_DELETE: 776 (void)snprintf(msg + len, sizeof(msg) - len, 777 "DELETE(%ju, %ju).", (uintmax_t)ggio->gctl_offset, 778 (uintmax_t)ggio->gctl_length); 779 break; 780 case BIO_FLUSH: 781 (void)snprintf(msg + len, sizeof(msg) - len, "FLUSH."); 782 break; 783 case BIO_WRITE: 784 (void)snprintf(msg + len, sizeof(msg) - len, 785 "WRITE(%ju, %ju).", (uintmax_t)ggio->gctl_offset, 786 (uintmax_t)ggio->gctl_length); 787 break; 788 default: 789 (void)snprintf(msg + len, sizeof(msg) - len, 790 "UNKNOWN(%u).", (unsigned int)ggio->gctl_cmd); 791 break; 792 } 793 } 794 pjdlog_common(loglevel, debuglevel, -1, "%s", msg); 795} 796 797static void 798remote_close(struct hast_resource *res, int ncomp) 799{ 800 801 rw_wlock(&hio_remote_lock[ncomp]); 802 /* 803 * A race is possible between dropping rlock and acquiring wlock - 804 * another thread can close connection in-between. 805 */ 806 if (!ISCONNECTED(res, ncomp)) { 807 assert(res->hr_remotein == NULL); 808 assert(res->hr_remoteout == NULL); 809 rw_unlock(&hio_remote_lock[ncomp]); 810 return; 811 } 812 813 assert(res->hr_remotein != NULL); 814 assert(res->hr_remoteout != NULL); 815 816 pjdlog_debug(2, "Closing old incoming connection to %s.", 817 res->hr_remoteaddr); 818 proto_close(res->hr_remotein); 819 res->hr_remotein = NULL; 820 pjdlog_debug(2, "Closing old outgoing connection to %s.", 821 res->hr_remoteaddr); 822 proto_close(res->hr_remoteout); 823 res->hr_remoteout = NULL; 824 825 rw_unlock(&hio_remote_lock[ncomp]); 826 827 /* 828 * Stop synchronization if in-progress. 829 */ 830 mtx_lock(&sync_lock); 831 if (sync_inprogress) 832 sync_inprogress = false; 833 mtx_unlock(&sync_lock); 834 835 /* 836 * Wake up guard thread, so it can immediately start reconnect. 837 */ 838 mtx_lock(&hio_guard_lock); 839 cv_signal(&hio_guard_cond); 840 mtx_unlock(&hio_guard_lock); 841} 842 843/* 844 * Thread receives ggate I/O requests from the kernel and passes them to 845 * appropriate threads: 846 * WRITE - always goes to both local_send and remote_send threads 847 * READ (when the block is up-to-date on local component) - 848 * only local_send thread 849 * READ (when the block isn't up-to-date on local component) - 850 * only remote_send thread 851 * DELETE - always goes to both local_send and remote_send threads 852 * FLUSH - always goes to both local_send and remote_send threads 853 */ 854static void * 855ggate_recv_thread(void *arg) 856{ 857 struct hast_resource *res = arg; 858 struct g_gate_ctl_io *ggio; 859 struct hio *hio; 860 unsigned int ii, ncomp, ncomps; 861 int error; 862 863 ncomps = HAST_NCOMPONENTS; 864 865 for (;;) { 866 pjdlog_debug(2, "ggate_recv: Taking free request."); 867 QUEUE_TAKE2(hio, free); 868 pjdlog_debug(2, "ggate_recv: (%p) Got free request.", hio); 869 ggio = &hio->hio_ggio; 870 ggio->gctl_unit = res->hr_ggateunit; 871 ggio->gctl_length = MAXPHYS; 872 ggio->gctl_error = 0; 873 pjdlog_debug(2, 874 "ggate_recv: (%p) Waiting for request from the kernel.", 875 hio); 876 if (ioctl(res->hr_ggatefd, G_GATE_CMD_START, ggio) < 0) { 877 if (sigexit_received) 878 pthread_exit(NULL); 879 primary_exit(EX_OSERR, "G_GATE_CMD_START failed"); 880 } 881 error = ggio->gctl_error; 882 switch (error) { 883 case 0: 884 break; 885 case ECANCELED: 886 /* Exit gracefully. */ 887 if (!sigexit_received) { 888 pjdlog_debug(2, 889 "ggate_recv: (%p) Received cancel from the kernel.", 890 hio); 891 pjdlog_info("Received cancel from the kernel, exiting."); 892 } 893 pthread_exit(NULL); 894 case ENOMEM: 895 /* 896 * Buffer too small? Impossible, we allocate MAXPHYS 897 * bytes - request can't be bigger than that. 898 */ 899 /* FALLTHROUGH */ 900 case ENXIO: 901 default: 902 primary_exitx(EX_OSERR, "G_GATE_CMD_START failed: %s.", 903 strerror(error)); 904 } 905 for (ii = 0; ii < ncomps; ii++) 906 hio->hio_errors[ii] = EINVAL; 907 reqlog(LOG_DEBUG, 2, ggio, 908 "ggate_recv: (%p) Request received from the kernel: ", 909 hio); 910 /* 911 * Inform all components about new write request. 912 * For read request prefer local component unless the given 913 * range is out-of-date, then use remote component. 914 */ 915 switch (ggio->gctl_cmd) { 916 case BIO_READ: 917 pjdlog_debug(2, 918 "ggate_recv: (%p) Moving request to the send queue.", 919 hio); 920 refcount_init(&hio->hio_countdown, 1); 921 mtx_lock(&metadata_lock); 922 if (res->hr_syncsrc == HAST_SYNCSRC_UNDEF || 923 res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) { 924 /* 925 * This range is up-to-date on local component, 926 * so handle request locally. 927 */ 928 /* Local component is 0 for now. */ 929 ncomp = 0; 930 } else /* if (res->hr_syncsrc == 931 HAST_SYNCSRC_SECONDARY) */ { 932 assert(res->hr_syncsrc == 933 HAST_SYNCSRC_SECONDARY); 934 /* 935 * This range is out-of-date on local component, 936 * so send request to the remote node. 937 */ 938 /* Remote component is 1 for now. */ 939 ncomp = 1; 940 } 941 mtx_unlock(&metadata_lock); 942 QUEUE_INSERT1(hio, send, ncomp); 943 break; 944 case BIO_WRITE: 945 for (;;) { 946 mtx_lock(&range_lock); 947 if (rangelock_islocked(range_sync, 948 ggio->gctl_offset, ggio->gctl_length)) { 949 pjdlog_debug(2, 950 "regular: Range offset=%jd length=%zu locked.", 951 (intmax_t)ggio->gctl_offset, 952 (size_t)ggio->gctl_length); 953 range_regular_wait = true; 954 cv_wait(&range_regular_cond, &range_lock); 955 range_regular_wait = false; 956 mtx_unlock(&range_lock); 957 continue; 958 } 959 if (rangelock_add(range_regular, 960 ggio->gctl_offset, ggio->gctl_length) < 0) { 961 mtx_unlock(&range_lock); 962 pjdlog_debug(2, 963 "regular: Range offset=%jd length=%zu is already locked, waiting.", 964 (intmax_t)ggio->gctl_offset, 965 (size_t)ggio->gctl_length); 966 sleep(1); 967 continue; 968 } 969 mtx_unlock(&range_lock); 970 break; 971 } 972 mtx_lock(&res->hr_amp_lock); 973 if (activemap_write_start(res->hr_amp, 974 ggio->gctl_offset, ggio->gctl_length)) { 975 (void)hast_activemap_flush(res); 976 } 977 mtx_unlock(&res->hr_amp_lock); 978 /* FALLTHROUGH */ 979 case BIO_DELETE: 980 case BIO_FLUSH: 981 pjdlog_debug(2, 982 "ggate_recv: (%p) Moving request to the send queues.", 983 hio); 984 refcount_init(&hio->hio_countdown, ncomps); 985 for (ii = 0; ii < ncomps; ii++) 986 QUEUE_INSERT1(hio, send, ii); 987 break; 988 } 989 } 990 /* NOTREACHED */ 991 return (NULL); 992} 993 994/* 995 * Thread reads from or writes to local component. 996 * If local read fails, it redirects it to remote_send thread. 997 */ 998static void * 999local_send_thread(void *arg) 1000{ 1001 struct hast_resource *res = arg; 1002 struct g_gate_ctl_io *ggio; 1003 struct hio *hio; 1004 unsigned int ncomp, rncomp; 1005 ssize_t ret; 1006 1007 /* Local component is 0 for now. */ 1008 ncomp = 0; 1009 /* Remote component is 1 for now. */ 1010 rncomp = 1; 1011 1012 for (;;) { 1013 pjdlog_debug(2, "local_send: Taking request."); 1014 QUEUE_TAKE1(hio, send, ncomp); 1015 pjdlog_debug(2, "local_send: (%p) Got request.", hio); 1016 ggio = &hio->hio_ggio; 1017 switch (ggio->gctl_cmd) { 1018 case BIO_READ: 1019 ret = pread(res->hr_localfd, ggio->gctl_data, 1020 ggio->gctl_length, 1021 ggio->gctl_offset + res->hr_localoff); 1022 if (ret == ggio->gctl_length) 1023 hio->hio_errors[ncomp] = 0; 1024 else { 1025 /* 1026 * If READ failed, try to read from remote node. 1027 */ 1028 QUEUE_INSERT1(hio, send, rncomp); 1029 continue; 1030 } 1031 break; 1032 case BIO_WRITE: 1033 ret = pwrite(res->hr_localfd, ggio->gctl_data, 1034 ggio->gctl_length, 1035 ggio->gctl_offset + res->hr_localoff); 1036 if (ret < 0) 1037 hio->hio_errors[ncomp] = errno; 1038 else if (ret != ggio->gctl_length) 1039 hio->hio_errors[ncomp] = EIO; 1040 else 1041 hio->hio_errors[ncomp] = 0; 1042 break; 1043 case BIO_DELETE: 1044 ret = g_delete(res->hr_localfd, 1045 ggio->gctl_offset + res->hr_localoff, 1046 ggio->gctl_length); 1047 if (ret < 0) 1048 hio->hio_errors[ncomp] = errno; 1049 else 1050 hio->hio_errors[ncomp] = 0; 1051 break; 1052 case BIO_FLUSH: 1053 ret = g_flush(res->hr_localfd); 1054 if (ret < 0) 1055 hio->hio_errors[ncomp] = errno; 1056 else 1057 hio->hio_errors[ncomp] = 0; 1058 break; 1059 } 1060 if (refcount_release(&hio->hio_countdown)) { 1061 if (ISSYNCREQ(hio)) { 1062 mtx_lock(&sync_lock); 1063 SYNCREQDONE(hio); 1064 mtx_unlock(&sync_lock); 1065 cv_signal(&sync_cond); 1066 } else { 1067 pjdlog_debug(2, 1068 "local_send: (%p) Moving request to the done queue.", 1069 hio); 1070 QUEUE_INSERT2(hio, done); 1071 } 1072 } 1073 } 1074 /* NOTREACHED */ 1075 return (NULL); 1076} 1077 1078/* 1079 * Thread sends request to secondary node. 1080 */ 1081static void * 1082remote_send_thread(void *arg) 1083{ 1084 struct hast_resource *res = arg; 1085 struct g_gate_ctl_io *ggio; 1086 struct hio *hio; 1087 struct nv *nv; 1088 unsigned int ncomp; 1089 bool wakeup; 1090 uint64_t offset, length; 1091 uint8_t cmd; 1092 void *data; 1093 1094 /* Remote component is 1 for now. */ 1095 ncomp = 1; 1096 1097 for (;;) { 1098 pjdlog_debug(2, "remote_send: Taking request."); 1099 QUEUE_TAKE1(hio, send, ncomp); 1100 pjdlog_debug(2, "remote_send: (%p) Got request.", hio); 1101 ggio = &hio->hio_ggio; 1102 switch (ggio->gctl_cmd) { 1103 case BIO_READ: 1104 cmd = HIO_READ; 1105 data = NULL; 1106 offset = ggio->gctl_offset; 1107 length = ggio->gctl_length; 1108 break; 1109 case BIO_WRITE: 1110 cmd = HIO_WRITE; 1111 data = ggio->gctl_data; 1112 offset = ggio->gctl_offset; 1113 length = ggio->gctl_length; 1114 break; 1115 case BIO_DELETE: 1116 cmd = HIO_DELETE; 1117 data = NULL; 1118 offset = ggio->gctl_offset; 1119 length = ggio->gctl_length; 1120 break; 1121 case BIO_FLUSH: 1122 cmd = HIO_FLUSH; 1123 data = NULL; 1124 offset = 0; 1125 length = 0; 1126 break; 1127 default: 1128 assert(!"invalid condition"); 1129 abort(); 1130 } 1131 nv = nv_alloc(); 1132 nv_add_uint8(nv, cmd, "cmd"); 1133 nv_add_uint64(nv, (uint64_t)ggio->gctl_seq, "seq"); 1134 nv_add_uint64(nv, offset, "offset"); 1135 nv_add_uint64(nv, length, "length"); 1136 if (nv_error(nv) != 0) { 1137 hio->hio_errors[ncomp] = nv_error(nv); 1138 pjdlog_debug(2, 1139 "remote_send: (%p) Unable to prepare header to send.", 1140 hio); 1141 reqlog(LOG_ERR, 0, ggio, 1142 "Unable to prepare header to send (%s): ", 1143 strerror(nv_error(nv))); 1144 /* Move failed request immediately to the done queue. */ 1145 goto done_queue; 1146 } 1147 pjdlog_debug(2, 1148 "remote_send: (%p) Moving request to the recv queue.", 1149 hio); 1150 /* 1151 * Protect connection from disappearing. 1152 */ 1153 rw_rlock(&hio_remote_lock[ncomp]); 1154 if (!ISCONNECTED(res, ncomp)) { 1155 rw_unlock(&hio_remote_lock[ncomp]); 1156 hio->hio_errors[ncomp] = ENOTCONN; 1157 goto done_queue; 1158 } 1159 /* 1160 * Move the request to recv queue before sending it, because 1161 * in different order we can get reply before we move request 1162 * to recv queue. 1163 */ 1164 mtx_lock(&hio_recv_list_lock[ncomp]); 1165 wakeup = TAILQ_EMPTY(&hio_recv_list[ncomp]); 1166 TAILQ_INSERT_TAIL(&hio_recv_list[ncomp], hio, hio_next[ncomp]); 1167 mtx_unlock(&hio_recv_list_lock[ncomp]); 1168 if (hast_proto_send(res, res->hr_remoteout, nv, data, 1169 data != NULL ? length : 0) < 0) { 1170 hio->hio_errors[ncomp] = errno; 1171 rw_unlock(&hio_remote_lock[ncomp]); 1172 remote_close(res, ncomp); 1173 pjdlog_debug(2, 1174 "remote_send: (%p) Unable to send request.", hio); 1175 reqlog(LOG_ERR, 0, ggio, 1176 "Unable to send request (%s): ", 1177 strerror(hio->hio_errors[ncomp])); 1178 /* 1179 * Take request back from the receive queue and move 1180 * it immediately to the done queue. 1181 */ 1182 mtx_lock(&hio_recv_list_lock[ncomp]); 1183 TAILQ_REMOVE(&hio_recv_list[ncomp], hio, hio_next[ncomp]); 1184 mtx_unlock(&hio_recv_list_lock[ncomp]); 1185 goto done_queue; 1186 } 1187 rw_unlock(&hio_remote_lock[ncomp]); 1188 nv_free(nv); 1189 if (wakeup) 1190 cv_signal(&hio_recv_list_cond[ncomp]); 1191 continue; 1192done_queue: 1193 nv_free(nv); 1194 if (ISSYNCREQ(hio)) { 1195 if (!refcount_release(&hio->hio_countdown)) 1196 continue; 1197 mtx_lock(&sync_lock); 1198 SYNCREQDONE(hio); 1199 mtx_unlock(&sync_lock); 1200 cv_signal(&sync_cond); 1201 continue; 1202 } 1203 if (ggio->gctl_cmd == BIO_WRITE) { 1204 mtx_lock(&res->hr_amp_lock); 1205 if (activemap_need_sync(res->hr_amp, ggio->gctl_offset, 1206 ggio->gctl_length)) { 1207 (void)hast_activemap_flush(res); 1208 } 1209 mtx_unlock(&res->hr_amp_lock); 1210 } 1211 if (!refcount_release(&hio->hio_countdown)) 1212 continue; 1213 pjdlog_debug(2, 1214 "remote_send: (%p) Moving request to the done queue.", 1215 hio); 1216 QUEUE_INSERT2(hio, done); 1217 } 1218 /* NOTREACHED */ 1219 return (NULL); 1220} 1221 1222/* 1223 * Thread receives answer from secondary node and passes it to ggate_send 1224 * thread. 1225 */ 1226static void * 1227remote_recv_thread(void *arg) 1228{ 1229 struct hast_resource *res = arg; 1230 struct g_gate_ctl_io *ggio; 1231 struct hio *hio; 1232 struct nv *nv; 1233 unsigned int ncomp; 1234 uint64_t seq; 1235 int error; 1236 1237 /* Remote component is 1 for now. */ 1238 ncomp = 1; 1239 1240 for (;;) { 1241 /* Wait until there is anything to receive. */ 1242 mtx_lock(&hio_recv_list_lock[ncomp]); 1243 while (TAILQ_EMPTY(&hio_recv_list[ncomp])) { 1244 pjdlog_debug(2, "remote_recv: No requests, waiting."); 1245 cv_wait(&hio_recv_list_cond[ncomp], 1246 &hio_recv_list_lock[ncomp]); 1247 } 1248 mtx_unlock(&hio_recv_list_lock[ncomp]); 1249 rw_rlock(&hio_remote_lock[ncomp]); 1250 if (!ISCONNECTED(res, ncomp)) { 1251 rw_unlock(&hio_remote_lock[ncomp]); 1252 /* 1253 * Connection is dead, so move all pending requests to 1254 * the done queue (one-by-one). 1255 */ 1256 mtx_lock(&hio_recv_list_lock[ncomp]); 1257 hio = TAILQ_FIRST(&hio_recv_list[ncomp]); 1258 assert(hio != NULL); 1259 TAILQ_REMOVE(&hio_recv_list[ncomp], hio, 1260 hio_next[ncomp]); 1261 mtx_unlock(&hio_recv_list_lock[ncomp]); 1262 goto done_queue; 1263 } 1264 if (hast_proto_recv_hdr(res->hr_remotein, &nv) < 0) { 1265 pjdlog_errno(LOG_ERR, 1266 "Unable to receive reply header"); 1267 rw_unlock(&hio_remote_lock[ncomp]); 1268 remote_close(res, ncomp); 1269 continue; 1270 } 1271 rw_unlock(&hio_remote_lock[ncomp]); 1272 seq = nv_get_uint64(nv, "seq"); 1273 if (seq == 0) { 1274 pjdlog_error("Header contains no 'seq' field."); 1275 nv_free(nv); 1276 continue; 1277 } 1278 mtx_lock(&hio_recv_list_lock[ncomp]); 1279 TAILQ_FOREACH(hio, &hio_recv_list[ncomp], hio_next[ncomp]) { 1280 if (hio->hio_ggio.gctl_seq == seq) { 1281 TAILQ_REMOVE(&hio_recv_list[ncomp], hio, 1282 hio_next[ncomp]); 1283 break; 1284 } 1285 } 1286 mtx_unlock(&hio_recv_list_lock[ncomp]); 1287 if (hio == NULL) { 1288 pjdlog_error("Found no request matching received 'seq' field (%ju).", 1289 (uintmax_t)seq); 1290 nv_free(nv); 1291 continue; 1292 } 1293 error = nv_get_int16(nv, "error"); 1294 if (error != 0) { 1295 /* Request failed on remote side. */ 1296 hio->hio_errors[ncomp] = 0; 1297 nv_free(nv); 1298 goto done_queue; 1299 } 1300 ggio = &hio->hio_ggio; 1301 switch (ggio->gctl_cmd) { 1302 case BIO_READ: 1303 rw_rlock(&hio_remote_lock[ncomp]); 1304 if (!ISCONNECTED(res, ncomp)) { 1305 rw_unlock(&hio_remote_lock[ncomp]); 1306 nv_free(nv); 1307 goto done_queue; 1308 } 1309 if (hast_proto_recv_data(res, res->hr_remotein, nv, 1310 ggio->gctl_data, ggio->gctl_length) < 0) { 1311 hio->hio_errors[ncomp] = errno; 1312 pjdlog_errno(LOG_ERR, 1313 "Unable to receive reply data"); 1314 rw_unlock(&hio_remote_lock[ncomp]); 1315 nv_free(nv); 1316 remote_close(res, ncomp); 1317 goto done_queue; 1318 } 1319 rw_unlock(&hio_remote_lock[ncomp]); 1320 break; 1321 case BIO_WRITE: 1322 case BIO_DELETE: 1323 case BIO_FLUSH: 1324 break; 1325 default: 1326 assert(!"invalid condition"); 1327 abort(); 1328 } 1329 hio->hio_errors[ncomp] = 0; 1330 nv_free(nv); 1331done_queue: 1332 if (refcount_release(&hio->hio_countdown)) { 1333 if (ISSYNCREQ(hio)) { 1334 mtx_lock(&sync_lock); 1335 SYNCREQDONE(hio); 1336 mtx_unlock(&sync_lock); 1337 cv_signal(&sync_cond); 1338 } else { 1339 pjdlog_debug(2, 1340 "remote_recv: (%p) Moving request to the done queue.", 1341 hio); 1342 QUEUE_INSERT2(hio, done); 1343 } 1344 } 1345 } 1346 /* NOTREACHED */ 1347 return (NULL); 1348} 1349 1350/* 1351 * Thread sends answer to the kernel. 1352 */ 1353static void * 1354ggate_send_thread(void *arg) 1355{ 1356 struct hast_resource *res = arg; 1357 struct g_gate_ctl_io *ggio; 1358 struct hio *hio; 1359 unsigned int ii, ncomp, ncomps; 1360 1361 ncomps = HAST_NCOMPONENTS; 1362 1363 for (;;) { 1364 pjdlog_debug(2, "ggate_send: Taking request."); 1365 QUEUE_TAKE2(hio, done); 1366 pjdlog_debug(2, "ggate_send: (%p) Got request.", hio); 1367 ggio = &hio->hio_ggio; 1368 for (ii = 0; ii < ncomps; ii++) { 1369 if (hio->hio_errors[ii] == 0) { 1370 /* 1371 * One successful request is enough to declare 1372 * success. 1373 */ 1374 ggio->gctl_error = 0; 1375 break; 1376 } 1377 } 1378 if (ii == ncomps) { 1379 /* 1380 * None of the requests were successful. 1381 * Use first error. 1382 */ 1383 ggio->gctl_error = hio->hio_errors[0]; 1384 } 1385 if (ggio->gctl_error == 0 && ggio->gctl_cmd == BIO_WRITE) { 1386 mtx_lock(&res->hr_amp_lock); 1387 activemap_write_complete(res->hr_amp, 1388 ggio->gctl_offset, ggio->gctl_length); 1389 mtx_unlock(&res->hr_amp_lock); 1390 } 1391 if (ggio->gctl_cmd == BIO_WRITE) { 1392 /* 1393 * Unlock range we locked. 1394 */ 1395 mtx_lock(&range_lock); 1396 rangelock_del(range_regular, ggio->gctl_offset, 1397 ggio->gctl_length); 1398 if (range_sync_wait) 1399 cv_signal(&range_sync_cond); 1400 mtx_unlock(&range_lock); 1401 /* 1402 * Bump local count if this is first write after 1403 * connection failure with remote node. 1404 */ 1405 ncomp = 1; 1406 rw_rlock(&hio_remote_lock[ncomp]); 1407 if (!ISCONNECTED(res, ncomp)) { 1408 mtx_lock(&metadata_lock); 1409 if (res->hr_primary_localcnt == 1410 res->hr_secondary_remotecnt) { 1411 res->hr_primary_localcnt++; 1412 pjdlog_debug(1, 1413 "Increasing localcnt to %ju.", 1414 (uintmax_t)res->hr_primary_localcnt); 1415 (void)metadata_write(res); 1416 } 1417 mtx_unlock(&metadata_lock); 1418 } 1419 rw_unlock(&hio_remote_lock[ncomp]); 1420 } 1421 if (ioctl(res->hr_ggatefd, G_GATE_CMD_DONE, ggio) < 0) 1422 primary_exit(EX_OSERR, "G_GATE_CMD_DONE failed"); 1423 pjdlog_debug(2, 1424 "ggate_send: (%p) Moving request to the free queue.", hio); 1425 QUEUE_INSERT2(hio, free); 1426 } 1427 /* NOTREACHED */ 1428 return (NULL); 1429} 1430 1431/* 1432 * Thread synchronize local and remote components. 1433 */ 1434static void * 1435sync_thread(void *arg __unused) 1436{ 1437 struct hast_resource *res = arg; 1438 struct hio *hio; 1439 struct g_gate_ctl_io *ggio; 1440 unsigned int ii, ncomp, ncomps; 1441 off_t offset, length, synced; 1442 bool dorewind; 1443 int syncext; 1444 1445 ncomps = HAST_NCOMPONENTS; 1446 dorewind = true; 1447 synced = 0; 1448 1449 for (;;) { 1450 mtx_lock(&sync_lock); 1451 while (!sync_inprogress) { 1452 dorewind = true; 1453 synced = 0; 1454 cv_wait(&sync_cond, &sync_lock); 1455 } 1456 mtx_unlock(&sync_lock); 1457 /* 1458 * Obtain offset at which we should synchronize. 1459 * Rewind synchronization if needed. 1460 */ 1461 mtx_lock(&res->hr_amp_lock); 1462 if (dorewind) 1463 activemap_sync_rewind(res->hr_amp); 1464 offset = activemap_sync_offset(res->hr_amp, &length, &syncext); 1465 if (syncext != -1) { 1466 /* 1467 * We synchronized entire syncext extent, we can mark 1468 * it as clean now. 1469 */ 1470 if (activemap_extent_complete(res->hr_amp, syncext)) 1471 (void)hast_activemap_flush(res); 1472 } 1473 mtx_unlock(&res->hr_amp_lock); 1474 if (dorewind) { 1475 dorewind = false; 1476 if (offset < 0) 1477 pjdlog_info("Nodes are in sync."); 1478 else { 1479 pjdlog_info("Synchronization started. %ju bytes to go.", 1480 (uintmax_t)(res->hr_extentsize * 1481 activemap_ndirty(res->hr_amp))); 1482 } 1483 } 1484 if (offset < 0) { 1485 mtx_lock(&sync_lock); 1486 sync_inprogress = false; 1487 mtx_unlock(&sync_lock); 1488 pjdlog_debug(1, "Nothing to synchronize."); 1489 /* 1490 * Synchronization complete, make both localcnt and 1491 * remotecnt equal. 1492 */ 1493 ncomp = 1; 1494 rw_rlock(&hio_remote_lock[ncomp]); 1495 if (ISCONNECTED(res, ncomp)) { 1496 if (synced > 0) { 1497 pjdlog_info("Synchronization complete. " 1498 "%jd bytes synchronized.", 1499 (intmax_t)synced); 1500 } 1501 mtx_lock(&metadata_lock); 1502 res->hr_syncsrc = HAST_SYNCSRC_UNDEF; 1503 res->hr_primary_localcnt = 1504 res->hr_secondary_localcnt; 1505 res->hr_primary_remotecnt = 1506 res->hr_secondary_remotecnt; 1507 pjdlog_debug(1, 1508 "Setting localcnt to %ju and remotecnt to %ju.", 1509 (uintmax_t)res->hr_primary_localcnt, 1510 (uintmax_t)res->hr_secondary_localcnt); 1511 (void)metadata_write(res); 1512 mtx_unlock(&metadata_lock); 1513 } else if (synced > 0) { 1514 pjdlog_info("Synchronization interrupted. " 1515 "%jd bytes synchronized so far.", 1516 (intmax_t)synced); 1517 } 1518 rw_unlock(&hio_remote_lock[ncomp]); 1519 continue; 1520 } 1521 pjdlog_debug(2, "sync: Taking free request."); 1522 QUEUE_TAKE2(hio, free); 1523 pjdlog_debug(2, "sync: (%p) Got free request.", hio); 1524 /* 1525 * Lock the range we are going to synchronize. We don't want 1526 * race where someone writes between our read and write. 1527 */ 1528 for (;;) { 1529 mtx_lock(&range_lock); 1530 if (rangelock_islocked(range_regular, offset, length)) { 1531 pjdlog_debug(2, 1532 "sync: Range offset=%jd length=%jd locked.", 1533 (intmax_t)offset, (intmax_t)length); 1534 range_sync_wait = true; 1535 cv_wait(&range_sync_cond, &range_lock); 1536 range_sync_wait = false; 1537 mtx_unlock(&range_lock); 1538 continue; 1539 } 1540 if (rangelock_add(range_sync, offset, length) < 0) { 1541 mtx_unlock(&range_lock); 1542 pjdlog_debug(2, 1543 "sync: Range offset=%jd length=%jd is already locked, waiting.", 1544 (intmax_t)offset, (intmax_t)length); 1545 sleep(1); 1546 continue; 1547 } 1548 mtx_unlock(&range_lock); 1549 break; 1550 } 1551 /* 1552 * First read the data from synchronization source. 1553 */ 1554 SYNCREQ(hio); 1555 ggio = &hio->hio_ggio; 1556 ggio->gctl_cmd = BIO_READ; 1557 ggio->gctl_offset = offset; 1558 ggio->gctl_length = length; 1559 ggio->gctl_error = 0; 1560 for (ii = 0; ii < ncomps; ii++) 1561 hio->hio_errors[ii] = EINVAL; 1562 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ", 1563 hio); 1564 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.", 1565 hio); 1566 mtx_lock(&metadata_lock); 1567 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) { 1568 /* 1569 * This range is up-to-date on local component, 1570 * so handle request locally. 1571 */ 1572 /* Local component is 0 for now. */ 1573 ncomp = 0; 1574 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ { 1575 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY); 1576 /* 1577 * This range is out-of-date on local component, 1578 * so send request to the remote node. 1579 */ 1580 /* Remote component is 1 for now. */ 1581 ncomp = 1; 1582 } 1583 mtx_unlock(&metadata_lock); 1584 refcount_init(&hio->hio_countdown, 1); 1585 QUEUE_INSERT1(hio, send, ncomp); 1586 1587 /* 1588 * Let's wait for READ to finish. 1589 */ 1590 mtx_lock(&sync_lock); 1591 while (!ISSYNCREQDONE(hio)) 1592 cv_wait(&sync_cond, &sync_lock); 1593 mtx_unlock(&sync_lock); 1594 1595 if (hio->hio_errors[ncomp] != 0) { 1596 pjdlog_error("Unable to read synchronization data: %s.", 1597 strerror(hio->hio_errors[ncomp])); 1598 goto free_queue; 1599 } 1600 1601 /* 1602 * We read the data from synchronization source, now write it 1603 * to synchronization target. 1604 */ 1605 SYNCREQ(hio); 1606 ggio->gctl_cmd = BIO_WRITE; 1607 for (ii = 0; ii < ncomps; ii++) 1608 hio->hio_errors[ii] = EINVAL; 1609 reqlog(LOG_DEBUG, 2, ggio, "sync: (%p) Sending sync request: ", 1610 hio); 1611 pjdlog_debug(2, "sync: (%p) Moving request to the send queue.", 1612 hio); 1613 mtx_lock(&metadata_lock); 1614 if (res->hr_syncsrc == HAST_SYNCSRC_PRIMARY) { 1615 /* 1616 * This range is up-to-date on local component, 1617 * so we update remote component. 1618 */ 1619 /* Remote component is 1 for now. */ 1620 ncomp = 1; 1621 } else /* if (res->hr_syncsrc == HAST_SYNCSRC_SECONDARY) */ { 1622 assert(res->hr_syncsrc == HAST_SYNCSRC_SECONDARY); 1623 /* 1624 * This range is out-of-date on local component, 1625 * so we update it. 1626 */ 1627 /* Local component is 0 for now. */ 1628 ncomp = 0; 1629 } 1630 mtx_unlock(&metadata_lock); 1631 1632 pjdlog_debug(2, "sync: (%p) Moving request to the send queues.", 1633 hio); 1634 refcount_init(&hio->hio_countdown, 1); 1635 QUEUE_INSERT1(hio, send, ncomp); 1636 1637 /* 1638 * Let's wait for WRITE to finish. 1639 */ 1640 mtx_lock(&sync_lock); 1641 while (!ISSYNCREQDONE(hio)) 1642 cv_wait(&sync_cond, &sync_lock); 1643 mtx_unlock(&sync_lock); 1644 1645 if (hio->hio_errors[ncomp] != 0) { 1646 pjdlog_error("Unable to write synchronization data: %s.", 1647 strerror(hio->hio_errors[ncomp])); 1648 goto free_queue; 1649 } 1650free_queue: 1651 mtx_lock(&range_lock); 1652 rangelock_del(range_sync, offset, length); 1653 if (range_regular_wait) 1654 cv_signal(&range_regular_cond); 1655 mtx_unlock(&range_lock); 1656 1657 synced += length; 1658 1659 pjdlog_debug(2, "sync: (%p) Moving request to the free queue.", 1660 hio); 1661 QUEUE_INSERT2(hio, free); 1662 } 1663 /* NOTREACHED */ 1664 return (NULL); 1665} 1666 1667static void 1668sighandler(int sig) 1669{ 1670 bool unlock; 1671 1672 switch (sig) { 1673 case SIGINT: 1674 case SIGTERM: 1675 sigexit_received = true; 1676 break; 1677 default: 1678 assert(!"invalid condition"); 1679 } 1680 /* 1681 * XXX: Racy, but if we cannot obtain hio_guard_lock here, we don't 1682 * want to risk deadlock. 1683 */ 1684 unlock = mtx_trylock(&hio_guard_lock); 1685 cv_signal(&hio_guard_cond); 1686 if (unlock) 1687 mtx_unlock(&hio_guard_lock); 1688} 1689 1690/* 1691 * Thread guards remote connections and reconnects when needed, handles 1692 * signals, etc. 1693 */ 1694static void * 1695guard_thread(void *arg) 1696{ 1697 struct hast_resource *res = arg; 1698 unsigned int ii, ncomps; 1699 int timeout; 1700 1701 ncomps = HAST_NCOMPONENTS; 1702 /* The is only one remote component for now. */ 1703#define ISREMOTE(no) ((no) == 1) 1704 1705 for (;;) { 1706 if (sigexit_received) { 1707 primary_exitx(EX_OK, 1708 "Termination signal received, exiting."); 1709 } 1710 /* 1711 * If all the connection will be fine, we will sleep until 1712 * someone wakes us up. 1713 * If any of the connections will be broken and we won't be 1714 * able to connect, we will sleep only for RECONNECT_SLEEP 1715 * seconds so we can retry soon. 1716 */ 1717 timeout = 0; 1718 pjdlog_debug(2, "remote_guard: Checking connections."); 1719 mtx_lock(&hio_guard_lock); 1720 for (ii = 0; ii < ncomps; ii++) { 1721 if (!ISREMOTE(ii)) 1722 continue; 1723 rw_rlock(&hio_remote_lock[ii]); 1724 if (ISCONNECTED(res, ii)) { 1725 assert(res->hr_remotein != NULL); 1726 assert(res->hr_remoteout != NULL); 1727 rw_unlock(&hio_remote_lock[ii]); 1728 pjdlog_debug(2, 1729 "remote_guard: Connection to %s is ok.", 1730 res->hr_remoteaddr); 1731 } else { 1732 assert(res->hr_remotein == NULL); 1733 assert(res->hr_remoteout == NULL); 1734 /* 1735 * Upgrade the lock. It doesn't have to be 1736 * atomic as no other thread can change 1737 * connection status from disconnected to 1738 * connected. 1739 */ 1740 rw_unlock(&hio_remote_lock[ii]); 1741 rw_wlock(&hio_remote_lock[ii]); 1742 assert(res->hr_remotein == NULL); 1743 assert(res->hr_remoteout == NULL); 1744 pjdlog_debug(2, 1745 "remote_guard: Reconnecting to %s.", 1746 res->hr_remoteaddr); 1747 init_remote(res); 1748 if (ISCONNECTED(res, ii)) { 1749 pjdlog_info("Successfully reconnected to %s.", 1750 res->hr_remoteaddr); 1751 } else { 1752 /* Both connections should be NULL. */ 1753 assert(res->hr_remotein == NULL); 1754 assert(res->hr_remoteout == NULL); 1755 pjdlog_debug(2, 1756 "remote_guard: Reconnect to %s failed.", 1757 res->hr_remoteaddr); 1758 timeout = RECONNECT_SLEEP; 1759 } 1760 rw_unlock(&hio_remote_lock[ii]); 1761 } 1762 } 1763 (void)cv_timedwait(&hio_guard_cond, &hio_guard_lock, timeout); 1764 mtx_unlock(&hio_guard_lock); 1765 } 1766#undef ISREMOTE 1767 /* NOTREACHED */ 1768 return (NULL); 1769} 1770