90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
import sys
|
|
import threading
|
|
|
|
from colony.core.logger.logging import get_logger
|
|
from colony.core.network.core import Network
|
|
from colony.core.dispatcher import Dispatcher
|
|
from colony.core.roles.foreman.role import Foreman
|
|
from colony.core.roles.gossip.role import Gossip, BOOTSTRAPPED
|
|
from colony.core.roles.manager import RoleManager
|
|
from colony.core.state_manager import StateManager
|
|
from colony.role_registry import RoleRegistry
|
|
from colony.settings import configuration, setting
|
|
from colony.transport.discover import Discover
|
|
from colony.transport.listener import Listener
|
|
from colony.utils.network import hosts_in_range
|
|
from role_settings import available_roles
|
|
|
|
|
|
def main():
|
|
logger = get_logger("main")
|
|
|
|
try:
|
|
hosts_in_range(setting("discovery.range_start"), setting("discovery.range_end"))
|
|
except ValueError as e:
|
|
logger.error(str(e))
|
|
sys.exit(1)
|
|
|
|
state_manager = StateManager()
|
|
network = Network()
|
|
if network.self_node.ip.startswith("127."):
|
|
logger.warning(f"This node's address is {network.self_node.ip}, other nodes will not be able to reach it")
|
|
|
|
role_registry = RoleRegistry(role_list=available_roles)
|
|
dispatcher = Dispatcher(network=network, role_registry=role_registry)
|
|
role_manager = RoleManager(
|
|
network=network,
|
|
role_registry=role_registry,
|
|
state_manager=state_manager,
|
|
)
|
|
|
|
role_manager.ensure_general_roles()
|
|
role_manager.ensure_on_demand_roles(setting("roles.on_demand"))
|
|
role_manager.initialize_roles()
|
|
|
|
listener = Listener(dispatcher=dispatcher)
|
|
try:
|
|
listener.bind()
|
|
except OSError as e:
|
|
logger.error(f"Cannot listen on {listener.host}:{listener.port}: {e}")
|
|
sys.exit(1)
|
|
|
|
threads = []
|
|
|
|
t = threading.Thread(
|
|
target=listener.start,
|
|
daemon=True,
|
|
name="Listener"
|
|
)
|
|
t.start()
|
|
threads.append(t)
|
|
|
|
t = threading.Thread(
|
|
target=Discover.listen,
|
|
args=(configuration.listener.port, "0.0.0.0", setting("discovery.discover_port")),
|
|
daemon=True,
|
|
name="Discover"
|
|
)
|
|
t.start()
|
|
threads.append(t)
|
|
|
|
roles = role_manager.get_role_instances()
|
|
gossip: Gossip = roles["Gossip"]
|
|
outcome = gossip.join_or_bootstrap()
|
|
gossip.release_roles()
|
|
|
|
if outcome == BOOTSTRAPPED:
|
|
foreman: Foreman = roles["Foreman"]
|
|
for _ in role_registry.roles:
|
|
if foreman.tick() is None:
|
|
break
|
|
|
|
role_manager.start_all()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|