+ def _wait_distributed_subtrees(self, count, status=None, rank=None, path=None):
+ try:
+ with contextutil.safe_while(sleep=5, tries=20) as proceed:
+ while proceed():
+ subtrees = self._get_subtrees(status=status, rank=rank, path=path)
+ subtrees = list(filter(lambda s: s['distributed_ephemeral_pin'] == True, subtrees))
+ log.info(f"len={len(subtrees)} {subtrees}")
+ if len(subtrees) >= count:
+ return subtrees
+ except contextutil.MaxWhileTries as e:
+ raise RuntimeError("rank {0} failed to reach desired subtree state".format(rank)) from e
+
+ def _wait_random_subtrees(self, count, status=None, rank=None, path=None):
+ try:
+ with contextutil.safe_while(sleep=5, tries=20) as proceed:
+ while proceed():
+ subtrees = self._get_subtrees(status=status, rank=rank, path=path)
+ subtrees = list(filter(lambda s: s['random_ephemeral_pin'] == True, subtrees))
+ log.info(f"len={len(subtrees)} {subtrees}")
+ if len(subtrees) >= count:
+ return subtrees
+ except contextutil.MaxWhileTries as e:
+ raise RuntimeError("rank {0} failed to reach desired subtree state".format(rank)) from e