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");
|