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