| 720 | |
| 721 | |
| 722 | Try<bool> GroupProcess::cache() |
| 723 | { |
| 724 | // Invalidate first (if it's not already). |
| 725 | memberships = None(); |
| 726 | |
| 727 | // Get all children to determine current memberships. |
| 728 | vector<string> results; |
| 729 | |
| 730 | int code = zk->getChildren(znode, true, &results); // Sets the watch! |
| 731 | |
| 732 | if (code == ZINVALIDSTATE || (code != ZOK && zk->retryable(code))) { |
| 733 | CHECK_NE(zk->getState(), ZOO_AUTH_FAILED_STATE); |
| 734 | return false; |
| 735 | } else if (code != ZOK) { |
| 736 | return Error("Non-retryable error attempting to get children of '" + znode |
| 737 | + "' in ZooKeeper: " + zk->message(code)); |
| 738 | } |
| 739 | |
| 740 | // Convert results to sequence numbers and (optionally) labels. |
| 741 | hashmap<int32_t, Option<string>> sequences; |
| 742 | |
| 743 | foreach (const string& result, results) { |
| 744 | vector<string> tokens = strings::tokenize(result, "_"); |
| 745 | Option<string> label = None(); |
| 746 | if (tokens.size() > 1) { |
| 747 | label = tokens[0]; |
| 748 | } |
| 749 | |
| 750 | Try<int32_t> sequence = numify<int32_t>(tokens.back()); |
| 751 | |
| 752 | // Skip it if it couldn't be converted to a number. |
| 753 | // NOTE: This is currently possible when using a replicated log |
| 754 | // based registry because the log replicas register under |
| 755 | // "/log_replicas" at the same path as the masters' ephemeral |
| 756 | // znodes. |
| 757 | if (sequence.isError()) { |
| 758 | VLOG(1) << "Found non-sequence node '" << result |
| 759 | << "' at '" << znode << "' in ZooKeeper"; |
| 760 | continue; |
| 761 | } |
| 762 | |
| 763 | sequences[sequence.get()] = label; |
| 764 | } |
| 765 | |
| 766 | // Cache current memberships, cancelling those that are now missing. |
| 767 | set<Group::Membership> current; |
| 768 | |
| 769 | foreachpair (int32_t sequence, Promise<bool>* cancelled, utils::copy(owned)) { |
| 770 | if (!sequences.contains(sequence)) { |
| 771 | cancelled->set(false); |
| 772 | owned.erase(sequence); // Okay since iterating over a copy. |
| 773 | delete cancelled; |
| 774 | } else { |
| 775 | current.insert(Group::Membership( |
| 776 | sequence, sequences[sequence], cancelled->future())); |
| 777 | |
| 778 | sequences.erase(sequence); |
| 779 | } |