Fixes + docker

colony/core/network/core.py

@@ -65,10 +65,11 @@ class Network:
 
 
     @staticmethod
-    def execute_to_unregistered_node(ip, port, command: Enum, *args, **kwargs):
+    def execute_to_unregistered_node(ip, port, command: Enum, *args, timeout=None, **kwargs):
         request = CommandRequest.form(command, *args, **kwargs)
 
-        return Whisperer(ip, port).send(request.collapse())
+        options = {} if timeout is None else {"timeout": timeout}
+        return Whisperer(ip, port).send(request.collapse(), **options)
 
 
     def load_network_from_snapshot(self, file_path: str = "storage/colony_snapshot.json"):

colony/core/network/membership.py

@@ -1,5 +1,6 @@
 import copy
 import ipaddress
+import math
 import threading
 import time
 from dataclasses import dataclass, field, asdict
@@ -19,7 +20,12 @@ def _is_int(value):
 
 
 def _is_number(value):
-    return isinstance(value, (int, float)) and not isinstance(value, bool)
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return False
+    try:
+        return math.isfinite(value)
+    except OverflowError:
+        return False
 
 
 def _text(value, name):
@@ -108,16 +114,15 @@ class Membership:
                  admission=None, clock=time.monotonic, wall=time.time):
         self.clock = clock
         self.wall = wall
-        self.member_timeout = setting("membership.member_timeout") if member_timeout is None else member_timeout
-        self.forget_after = setting("membership.forget_after") if forget_after is None else forget_after
-        self.max_members = setting("membership.max_members") if max_members is None else max_members
-        self._admission = resolve_admission(setting("membership.admission") if admission is None else admission)
+        self.member_timeout = setting("membership.member_timeout", member_timeout)
+        self.forget_after = setting("membership.forget_after", forget_after)
+        self.max_members = setting("membership.max_members", max_members)
+        self._admission = resolve_admission(setting("membership.admission", admission))
         self.denied = False
 
         self._lock = threading.RLock()
         self._self = MemberRecord(ip, port, hostname, incarnation=int(wall() * 1000), seq=1)
-        self._records = {}
-        self._advanced = {}
+        self._entries = {}
         self._declines = {}
 
     @property
@@ -172,35 +177,36 @@ class Membership:
             now = self.clock()
             for record in records:
                 if record.key == self.key:
+                    if record.version > self._self.version:
+                        self._self.incarnation = record.incarnation
+                        self._self.seq = record.seq + 1
                     continue
-                current = self._records.get(record.key)
-                if current is not None and record.version <= current.version:
+                current = self._entries.get(record.key)
+                if current is not None and record.version <= current[0].version:
                     continue
-                if current is None and len(self._records) >= self.max_members:
+                if current is None and len(self._entries) >= self.max_members:
                     continue
-                self._records[record.key] = record
-                self._advanced[record.key] = now
+                self._entries[record.key] = (record, now)
                 changed += 1
             self._purge(now)
         return changed
 
     def _purge(self, now):
-        for key in [key for key, advanced in self._advanced.items() if now - advanced > self.forget_after]:
-            del self._records[key]
-            del self._advanced[key]
+        for key in [key for key, (_, advanced) in self._entries.items() if now - advanced > self.forget_after]:
+            del self._entries[key]
 
     def alive_records(self):
         with self._lock:
             now = self.clock()
             alive = [self._current_self()]
-            alive += [record for key, record in self._records.items()
-                      if now - self._advanced[key] <= self.member_timeout]
+            alive += [record for record, advanced in self._entries.values()
+                      if now - advanced <= self.member_timeout]
             return sorted(alive, key=lambda record: record.key)
 
     def known_addresses(self):
         with self._lock:
             self._purge(self.clock())
-            return sorted(self._records)
+            return sorted(self._entries)
 
     def export(self):
         return [record.to_dict() for record in self.alive_records()]

colony/core/roles/gossip/role.py

@@ -27,6 +27,7 @@ class Gossip(Role):
     BOOTSTRAP_REPROBE = 5
     DENIED_BACKOFF_CAP = 60
     SCORES_TTL = 30
+    MAX_SEEDS = 64
 
     class COMMANDS(Enum):
         HELLO = 'Gossip.Hello'
@@ -41,6 +42,7 @@ class Gossip(Role):
 
         self.reprobe_interval = setting("membership.reprobe_interval")
         self.fanout = setting("membership.fanout")
+        self.hello_timeout = setting("membership.hello_timeout")
         self.max_members = setting("membership.max_members")
         self.discover_port = setting("discovery.discover_port")
         self.probe_timeout = setting("discovery.probe_timeout")
@@ -50,7 +52,6 @@ class Gossip(Role):
         self.random = random.Random()
         self.next_reprobe = 0.0
         self.denied_attempts = 0
-        self.outcome = None
         self._scores = {}
         self._scores_at = None
         self._persisted_peers = None
@@ -91,17 +92,13 @@ class Gossip(Role):
                 seeds.append((entry[0], entry[1]))
         return seeds
 
-    def probe_targets(self):
-        membership = self.membership
-        alive = {record.key for record in membership.alive_records()}
-        remembered = (set(membership.known_addresses()) | set(self.seed_addresses())) - alive
+    def probe_targets(self, alive):
+        remembered = (set(self.membership.known_addresses()) | set(self.seed_addresses())) - alive
 
-        targets = list(self.scan_targets())
+        targets = dict.fromkeys(self.scan_targets())
         for ip, port in sorted(remembered):
-            target = self.discover_target(ip, port)
-            if target not in targets:
-                targets.append(target)
-        return targets
+            targets[self.discover_target(ip, port)] = None
+        return list(targets)
 
     def _probe(self, target):
         host, discover_port = target
@@ -120,7 +117,7 @@ class Gossip(Role):
     def find_peers(self):
         membership = self.membership
         alive = {record.key for record in membership.alive_records()}
-        found = self.scan(self.probe_targets())
+        found = self.scan(self.probe_targets(alive))
         return [address for address in dict.fromkeys(found) if address not in alive]
 
     def exchange(self, ip, port):
@@ -129,7 +126,9 @@ class Gossip(Role):
         known = [record.to_dict() for record in membership.alive_records() if record.key != membership.key]
 
         try:
-            resp = self.context.execute_to_unregistered_node(ip, port, self.COMMANDS.HELLO, sender, known)
+            resp = self.context.execute_to_unregistered_node(
+                ip, port, self.COMMANDS.HELLO, sender, known, timeout=self.hello_timeout
+            )
         except Exception as e:
             self.logger.debug(f"Hello to {ip}:{port} failed: {e}")
             return FAILED
@@ -153,16 +152,23 @@ class Gossip(Role):
         now = self.membership.now()
         if self._scores_at is None or now - self._scores_at >= self.SCORES_TTL:
             self._scores = self.collect_scores()
-            self._scores_at = now
+            if self._scores:
+                self._scores_at = now
         return self._scores
 
     def collect_scores(self):
         return collect_scores()
 
     def refresh_self(self):
-        self.membership.update_self(roles=list(self.context.self().roles), scores=self.scores())
+        self.membership.update_self(scores=self.scores())
 
-    def _after_probe(self, results, now):
+    def exchange_all(self, addresses):
+        if not addresses:
+            return []
+        with ThreadPoolExecutor(max_workers=min(self.probe_workers, len(addresses))) as pool:
+            return list(pool.map(lambda address: self.exchange(*address), addresses))
+
+    def _after_probe(self, results):
         membership = self.membership
         has_peers = any(record.key != membership.key for record in membership.alive_records())
 
@@ -179,28 +185,27 @@ class Gossip(Role):
             outcome, delay = BOOTSTRAPPED, self.BOOTSTRAP_REPROBE
             membership.denied = False
 
-        self.next_reprobe = now + delay
-        self.outcome = outcome
+        self.next_reprobe = membership.now() + delay
         return outcome
 
     def join_or_bootstrap(self):
         self.refresh_self()
-        results = [self.exchange(ip, port) for ip, port in self.find_peers()]
-        outcome = self._after_probe(results, self.membership.now())
+        results = self.exchange_all(self.find_peers())
+        outcome = self._after_probe(results)
         self.logger.info(f"Startup probe finished: {outcome}")
         return outcome
 
     def reprobe(self):
-        results = [self.exchange(ip, port) for ip, port in self.find_peers()]
-        self._after_probe(results, self.membership.now())
+        results = self.exchange_all(self.find_peers())
+        self._after_probe(results)
 
     def tick(self):
         membership = self.membership
         self.refresh_self()
 
         peers = [record for record in membership.alive_records() if record.key != membership.key]
-        for record in self.random.sample(peers, min(self.fanout, len(peers))):
-            self.exchange(*record.key)
+        chosen = self.random.sample(peers, min(self.fanout, len(peers)))
+        self.exchange_all([record.key for record in chosen])
 
         if membership.now() >= self.next_reprobe:
             self.reprobe()
@@ -209,7 +214,9 @@ class Gossip(Role):
         self.release_roles()
 
     def persist_peers(self):
-        addresses = self.membership.known_addresses()
+        known = self.membership.known_addresses()
+        seeds = [address for address in self.seed_addresses() if address not in known]
+        addresses = (known + seeds)[:self.MAX_SEEDS]
         if addresses == self._persisted_peers:
             return
         state = self.context.get_role_manager().state_manager
@@ -234,4 +241,4 @@ class Gossip(Role):
         drop = set(forbidden) | set(over_held)
         if drop:
             self.logger.info(f"Releasing roles {sorted(drop)}")
-            manager.set_roles([name for name in held if name not in drop])
+            manager.drop_roles(drop)

colony/core/roles/manager.py

@@ -1,9 +1,9 @@
 import threading
 
 from colony.core.context import RoleContext
+from colony.core.network.advice import assignable_roles
 from colony.core.roles.policy import DefaultRolePolicy
 from colony.core.state_manager import StateManager
-from colony.role_registry import RolePriority
 from colony.core.logger.logging import get_logger
 
 
@@ -18,10 +18,8 @@ class RoleManager:
         self._lock = threading.RLock()
 
     def ensure_general_roles(self):
-        general = sorted(
-            name for name, stats in self.role_registry.roles.items()
-            if stats.priority == RolePriority.GENERAL
-        )
+        roles = self.role_registry.roles
+        general = sorted(set(roles) - set(assignable_roles(roles)))
         current = self.state_manager.get(StateManager.Data.Roles)
         wanted = list(dict.fromkeys(general + current))
         if wanted != current:
@@ -42,11 +40,14 @@ class RoleManager:
                 self.logger.debug(f"Creating instance of role {role_name}. ")
                 roles[role_name] = role_cls(context=context)
 
-            self.network.self_node.roles = list(roles)
-            self.network.self_node.set_role_instances(roles)
-            self.network.membership.update_self(roles=list(roles))
+            self._publish(roles)
         self.logger.info(f"Roles initialized: {list(roles.keys())}. ")
 
+    def _publish(self, roles):
+        self.network.self_node.set_role_instances(roles)
+        self.network.self_node.roles = list(roles)
+        self.network.membership.update_self(roles=list(roles))
+
     def get_role_instances(self):
         return self.network.self_node.get_role_instances()
 
@@ -92,9 +93,7 @@ class RoleManager:
             }
 
             self.state_manager.set(StateManager.Data.Roles, list(final_roles.keys()))
-            self.network.self_node.set_role_instances(final_roles)
-            self.network.self_node.roles = list(final_roles.keys())
-            self.network.membership.update_self(roles=list(final_roles))
+            self._publish(final_roles)
 
             for role_name in roles_to_remove:
                 self.logger.debug(f"Stopping removed role {role_name}")
@@ -106,6 +105,17 @@ class RoleManager:
 
         self.logger.info(f"Roles updated to: {list(final_roles.keys())}")
 
+    def drop_roles(self, role_names):
+        with self._lock:
+            assignable = assignable_roles(self.role_registry.roles)
+            held = list(self.get_role_instances())
+            remaining = [name for name in held if name not in role_names or name not in assignable]
+            if remaining != held:
+                self.set_roles(remaining)
+
+    def _decline(self, role_name):
+        self.network.membership.decline(role_name, self.role_policy.decline_cooldown)
+
     def request_role(self, role_name):
         membership = self.network.membership
 
@@ -124,9 +134,15 @@ class RoleManager:
             accepted, reason = self.role_policy.decide(role_name, held, scores, self.role_registry.roles)
             if not accepted:
                 self.logger.info(f"Declining role {role_name}: {reason}")
-                membership.decline(role_name, self.role_policy.decline_cooldown)
+                self._decline(role_name)
                 return {"accepted": False, "reason": reason}
 
-            self.set_roles(held + [role_name])
+            try:
+                self.set_roles(held + [role_name])
+            except Exception:
+                self.logger.exception(f"Could not take role {role_name}")
+                self._decline(role_name)
+                return {"accepted": False, "reason": "role could not be started"}
+
             membership.clear_decline(role_name)
             return {"accepted": True, "reason": reason}

colony/core/roles/policy.py

@@ -1,4 +1,4 @@
-from colony.role_registry import RolePriority
+from colony.core.network.advice import assignable_roles
 from colony.settings import setting
 
 
@@ -14,11 +14,11 @@ class RolePolicy:
 
 class DefaultRolePolicy(RolePolicy):
     def __init__(self, max_roles=None, deny=None, min_scores=None, decline_cooldown=None):
-        self.max_roles = setting("role_policy.max_roles") if max_roles is None else max_roles
-        self.deny = set(setting("role_policy.deny") if deny is None else deny)
-        raw_minimums = setting("role_policy.min_scores") if min_scores is None else min_scores
+        self.max_roles = setting("role_policy.max_roles", max_roles)
+        self.deny = set(setting("role_policy.deny", deny))
+        raw_minimums = setting("role_policy.min_scores", min_scores)
         self.min_scores = {role: dict(minimums) for role, minimums in dict(raw_minimums).items()}
-        self.decline_cooldown = setting("role_policy.decline_cooldown") if decline_cooldown is None else decline_cooldown
+        self.decline_cooldown = setting("role_policy.decline_cooldown", decline_cooldown)
 
     def _violation(self, role_name, scores):
         if role_name in self.deny:
@@ -34,11 +34,13 @@ class DefaultRolePolicy(RolePolicy):
             return False, reason
 
         if self.max_roles is not None:
-            assigned = [name for name in held if name in roles and roles[name].priority != RolePriority.GENERAL]
+            assigned = [name for name in held if name in assignable_roles(roles)]
             if len(assigned) >= self.max_roles:
                 return False, "node already holds the maximum number of roles"
 
         return True, "accepted"
 
     def should_keep(self, role_name, scores):
+        if not scores:
+            return role_name not in self.deny
         return self._violation(role_name, scores) is None

colony/core/roles/role.py

@@ -19,6 +19,7 @@ class Role:
         self._stop_event = threading.Event()
 
         self._thread = None
+        self._start_lock = threading.Lock()
 
         if self.CONTINUOUS and self.LOOP_SECONDS_WAIT is None:
             raise NotImplementedError("You made this role to run continuously but have not set seconds to wait. Please set constant variable LOOP_SECONDS_WAIT. ")
@@ -31,11 +32,14 @@ class Role:
         return self.__class__.__name__
 
     def start(self):
-        if not self.CONTINUOUS or self._thread is not None:
+        if not self.CONTINUOUS:
             return
 
-        self._thread = threading.Thread(target=self._run_loop, daemon=True, name=str(self))
-        self._thread.start()
+        with self._start_lock:
+            if self._thread is not None:
+                return
+            self._thread = threading.Thread(target=self._run_loop, daemon=True, name=str(self))
+            self._thread.start()
 
     def _run_loop(self):
         while not self._stop_event.is_set():

colony/settings.py

@@ -17,6 +17,7 @@ DEFAULTS = {
     "membership.reprobe_interval": 15,
     "membership.fanout": 3,
     "membership.forget_after": 600,
+    "membership.hello_timeout": 3,
     "membership.max_members": 1024,
     "membership.admission": "open",
     "foreman.interval": 5,
@@ -30,5 +31,7 @@ DEFAULTS = {
 }
 
 
-def setting(key):
+def setting(key, override=None):
+    if override is not None:
+        return override
     return configuration.get(key, DEFAULTS[key])