Merge pull request hyperledger#67 from hyperledger/master

Master

plenum/common/stack_manager.py

@@ -62,7 +62,7 @@ def ledger(self):
         return ledger
 
     @staticmethod
-    def parseLedgerForHaAndKeys(ledger, returnActive=True):
+    def parseLedgerForHaAndKeys(ledger, returnActive=True, ledger_size=None):
         """
         Returns validator ip, ports and keys
         :param ledger:
@@ -76,7 +76,8 @@ def parseLedgerForHaAndKeys(ledger, returnActive=True):
         activeValidators = set()
         try:
             TxnStackManager._parse_pool_transaction_file(
-                ledger, nodeReg, cliNodeReg, nodeKeys, activeValidators)
+                ledger, nodeReg, cliNodeReg, nodeKeys, activeValidators,
+                ledger_size=ledger_size)
         except ValueError:
             errMsg = 'Pool transaction file corrupted. Rebuild pool transactions.'
             logger.exception(errMsg)
@@ -96,11 +97,12 @@ def parseLedgerForHaAndKeys(ledger, returnActive=True):
 
     @staticmethod
     def _parse_pool_transaction_file(
-            ledger, nodeReg, cliNodeReg, nodeKeys, activeValidators):
+            ledger, nodeReg, cliNodeReg, nodeKeys, activeValidators,
+            ledger_size=None):
         """
         helper function for parseLedgerForHaAndKeys
         """
-        for _, txn in ledger.getAllTxn():
+        for _, txn in ledger.getAllTxn(to=ledger_size):
             if txn[TXN_TYPE] == NODE:
                 nodeName = txn[DATA][ALIAS]
                 clientStackName = nodeName + CLIENT_STACK_SUFFIX
@@ -234,6 +236,11 @@ def addRemoteKeysFromLedger(self, keys):
                 logger.error("Exception while initializing keep for remote {}".
                              format(ex))
 
+    def getNodeRegistry(self, ledger_size=None):
+        nodeReg, _, _ = self.parseLedgerForHaAndKeys(
+            self.ledger, ledger_size=ledger_size)
+        return nodeReg
+
     def nodeExistsInLedger(self, nym):
         # Since PoolLedger is going to be small so using
         # `getAllTxn` is fine

plenum/common/startable.py

@@ -58,8 +58,12 @@ class Mode(IntEnum):
 
     @classmethod
     def is_done_discovering(cls, mode):
+        if mode is None:
+            return False
         return mode >= cls.discovered
 
     @classmethod
     def is_done_syncing(cls, mode):
+        if mode is None:
+            return False
         return mode >= cls.synced

plenum/common/test_network_setup.py

@@ -192,13 +192,15 @@ def bootstrapTestNodes(cls, config, startingPort, nodeParamsFileName, domainTxnF
             bad_idxs = [x for x in args.nodeNum if not (1 <= x <= args.nodes)]
             assert not bad_idxs, "nodeNum should be less or equal to nodeCount"
 
+        node_num = [args.nodeNum, None] if args.nodeNum else [None]
+
         steward_defs, node_defs = cls.gen_defs(args.ips, args.nodes, startingPort)
         client_defs = cls.gen_client_defs(args.clients)
         trustee_def = cls.gen_trustee_def(1)
-        cls.bootstrapTestNodesCore(config, args.network, args.appendToLedgers,
-                                   domainTxnFieldOrder, trustee_def,
-                                   steward_defs, node_defs, client_defs,
-                                   args.nodeNum, nodeParamsFileName)
+
+        for n_num in node_num:
+            cls.bootstrapTestNodesCore(config, args.network, args.appendToLedgers, domainTxnFieldOrder, trustee_def,
+                                       steward_defs, node_defs, client_defs, n_num, nodeParamsFileName)
 
         # edit NETWORK_NAME in config
         for line in fileinput.input(['/etc/indy/indy_config.py'], inplace=True):
@@ -207,10 +209,10 @@ def bootstrapTestNodes(cls, config, startingPort, nodeParamsFileName, domainTxnF
         with open('/etc/indy/indy_config.py', 'a') as cfgfile:
             cfgfile.write("NETWORK_NAME = '{}'".format(args.network))
 
-        # in case of client only delete unnecessary key dir
-        if args.nodeNum is None:
-            key_dir = cls.setup_clibase_dir(config, args.network)
-            key_dir = os.path.join(key_dir, "keys")
+        # delete unnecessary key dir in client folder
+        key_dir = cls.setup_clibase_dir(config, args.network)
+        key_dir = os.path.join(key_dir, "keys")
+        if os.path.isdir(key_dir):
             shutil.rmtree(key_dir, ignore_errors=True)
 
     @staticmethod

plenum/common/util.py

@@ -84,15 +84,37 @@ def randomSeed(size=32):
     return randomString(size)
 
 
-def mostCommonElement(elements: Iterable[T]):
+def mostCommonElement(elements: Iterable[T], to_hashable_f: Callable=None):
     """
     Find the most frequent element of a collection.
 
     :param elements: An iterable of elements
+    :param to_hashable_f: (optional) if defined will be used to get
+        hashable presentation for non-hashable elements. Otherwise json.dumps
+        is used with sort_keys=True
     :return: element which is the most frequent in the collection and
         the number of its occurrences
     """
-    return Counter(elements).most_common(n=1)[0]
+    class _Hashable(collections.abc.Hashable):
+        def __init__(self, orig):
+            self.orig = orig
+
+            if isinstance(orig, collections.Hashable):
+                self.hashable = orig
+            elif to_hashable_f is not None:
+                self.hashable = to_hashable_f(orig)
+            else:
+                self.hashable = json.dumps(orig, sort_keys=True)
+
+        def __eq__(self, other):
+            return self.hashable == other.hashable
+
+        def __hash__(self):
+            return hash(self.hashable)
+
+    _elements = (_Hashable(el) for el in elements)
+    most_common, counter = Counter(_elements).most_common(n=1)[0]
+    return (most_common.orig, counter)
 
 
 def updateNamedTuple(tupleToUpdate: NamedTuple, **kwargs):

plenum/server/domain_req_handler.py

@@ -163,7 +163,9 @@ def make_proof(self, path):
         :param path: the path generate a state proof for
         :return: a state proof or None
         '''
-        proof = self.state.generate_state_proof(path, serialize=True)
+        proof = self.state.generate_state_proof(key=path,
+                                                root=self.state.committedHead,
+                                                serialize=True)
         root_hash = self.state.committedHeadHash
         encoded_proof = proof_nodes_serializer.serialize(proof)
         encoded_root_hash = state_roots_serializer.serialize(bytes(root_hash))

plenum/server/has_action_queue.py

@@ -18,6 +18,7 @@ def __init__(self):
         self.aqNextCheck = float('inf')  # next time to check
         self.aid = 0  # action id
         self.repeatingActions = set()
+        self.scheduled = dict()
 
     def _schedule(self, action: Callable, seconds: int=0) -> int:
         """
@@ -39,8 +40,37 @@ def _schedule(self, action: Callable, seconds: int=0) -> int:
             logger.trace("{} scheduling action {} with id {} to run now".
                          format(self, get_func_name(action), self.aid))
             self.actionQueue.append((action, self.aid))
+
+        if action not in self.scheduled:
+            self.scheduled[action] = []
+        self.scheduled[action].append(self.aid)
+
         return self.aid
 
+    def _cancel(self, action: Callable = None, aid: int = None):
+        """
+        Cancel scheduled events
+
+        :param action:  (optional) scheduled action. If specified, all
+                scheduled events for the action are cancelled.
+        :param aid:     (options) scheduled event id. If specified,
+                scheduled event with the aid is cancelled.
+        """
+        if action is not None:
+            if action in self.scheduled:
+                logger.trace("{} cancelling all events for action {}, ids: {}"
+                             "".format(self, action, self.scheduled[action]))
+                self.scheduled[action].clear()
+        elif aid is not None:
+            for action, aids in self.scheduled.items():
+                try:
+                    aids.remove(aid)
+                except ValueError:
+                    pass
+                else:
+                    logger.trace("{} cancelled action {} with id {}".format(self, action, aid))
+                    break
+
     def _serviceActions(self) -> int:
         """
         Run all pending actions in the action queue.
@@ -62,9 +92,15 @@ def _serviceActions(self) -> int:
         count = len(self.actionQueue)
         while self.actionQueue:
             action, aid = self.actionQueue.popleft()
-            logger.trace("{} running action {} with id {}".
-                         format(self, get_func_name(action), aid))
-            action()
+            assert action in self.scheduled
+            if aid in self.scheduled[action]:
+                self.scheduled[action].remove(aid)
+                logger.trace("{} running action {} with id {}".
+                             format(self, get_func_name(action), aid))
+                action()
+            else:
+                logger.trace("{} not running cancelled action {} with id {}".
+                             format(self, get_func_name(action), aid))
         return count
 
     def startRepeating(self, action: Callable, seconds: int):