Update peer_discovery.py

peer_discovery.py

@@ -8,18 +8,16 @@ import requests
 from zeroconf import Zeroconf, ServiceInfo, ServiceBrowser
 import random
 
-# منفذ عشوائي مكوَّن من 4 أرقام (مثلاً 0007، 8321…)
-PORT = int(os.getenv("CPU_PORT", random.randint(1, 9999)))
+# ---------- المتغيرات الأساسية ----------
+PORT = None
+CURRENT_SERVER = None
+CONNECTED = threading.Event()
+PEERS = set()
 
-# إذا كان باقي الكود يحتاج متغيّر rport iterable
-# أنشئ مجموعة تحتوي هذا المنفذ فقط (اختياري):
-rport = {f"{PORT:04}"}
-# 👇 إعداد الـ peer discovery عبر LAN وInternet
 SERVICE = "_tasknode._tcp.local."
-PORT = int(os.getenv("CPU_PORT", random.choice(list(rport))))
-PEERS = set()  # مجموعة URLs للأقران (/run)
+LAN_PORT = 7520  # منفذ LAN الداخلي
 
-# 🌐 قائمة السيرفرات (Failover List)
+# ---------- قائمة السيرفرات ----------
 BASES = [
     "https://cv4790811.regru.cloud",
     "https://amaloffload.onrender.com",
@@ -28,8 +26,8 @@ BASES = [
     "http://10.229.228.178",
 ]
 
-CENTRAL_REGISTRY_SERVERS = [f"{base}:{port}" for base in BASES for port in rport]
-
+RPORTS = [f"{i:04}" for i in range(1, 10000)]
+CENTRAL_REGISTRY_SERVERS = [f"{base}:{port}" for base in BASES for port in RPORTS]
 current_server_index = 0
 
 
@@ -50,6 +48,7 @@ def get_local_ip():
             try: s.close()
             except: pass
 
+
 # 🔄 اختيار السيرفر المتاح
 def get_active_central_server():
     global current_server_index
@@ -64,6 +63,7 @@ def get_active_central_server():
             continue
     return None
 
+
 # ❶ تسجيل الخدمة في شبكة LAN
 def register_service_lan():
     zc = Zeroconf()
@@ -81,6 +81,7 @@ def register_service_lan():
     except Exception as e:
         print(f"❌ LAN registration failed: {e}")
 
+
 # ❷ مستمع اكتشاف LAN
 class Listener:
     def add_service(self, zc, t, name):
@@ -103,37 +104,53 @@ def discover_lan_loop():
     while True:
         time.sleep(5)
 
-# ❸ التسجيل في الخادم المركزي (مع Failover)
-def register_with_central():
-    node_id = os.getenv("NODE_ID", socket.gethostname())
-    info = {"node_id": node_id, "ip": get_local_ip(), "port": PORT}
-    server = get_active_central_server()
-    if not server:
-        print("❌ لا يوجد أي سيرفر مركزي متاح حالياً")
-        return
-    try:
-        resp = requests.post(f"{server}/register", json=info, timeout=5)
-        resp.raise_for_status()
-        peers_list = resp.json()
-        for p in peers_list:
-            peer_url = f"http://{p['ip']}:{p['port']}/run"
-            if peer_url not in PEERS:
-                PEERS.add(peer_url)
-                print(f"🌐 Registered and discovered central peer: {peer_url}")
-    except Exception as e:
-        print(f"❌ Central registration failed on {server}: {e}")
 
-# ❹ مزامنة الأقران مع السيرفر المركزي (مع Failover)
+# ❸ الاتصال بسيرفر مركزي (مع محاولات حتى النجاح)
+def connect_until_success():
+    """
+    يحاول الاتصال بكل سيرفر وبورت بالتسلسل، ويتوقف عند أول نجاح
+    ويضبط PORT وCONNECTED وCENTRAL_REGISTRY
+    """
+    global PORT, current_server_index, CURRENT_SERVER
+
+    for port_int in range(1, 10000):
+        port = f"{port_int:04}"
+        for idx, server_base in enumerate(BASES):
+            server = f"{server_base}:{port}"
+            info = {
+                "node_id": os.getenv("NODE_ID", socket.gethostname()),
+                "ip": get_local_ip(),
+                "port": int(port)
+            }
+            try:
+                resp = requests.post(f"{server}/register", json=info, timeout=5)
+                resp.raise_for_status()
+                PORT = int(port)
+                CURRENT_SERVER = server
+                current_server_index = idx
+                CONNECTED.set()
+                print(f"✅ Connected: {server} on port {PORT}")
+                peers = resp.json()
+                for p in peers:
+                    peer_url = f"http://{p['ip']}:{p['port']}/run"
+                    if peer_url not in PEERS:
+                        PEERS.add(peer_url)
+                        print(f"🌐 Central peer: {peer_url}")
+                return
+            except Exception as e:
+                logging.info("❌ %s:%s -> %s", server_base, port, e)
+        time.sleep(0.1)
+
+
+# ❹ مزامنة مستمرة مع السيرفر المركزي
 def fetch_central_loop():
     print("🔄 Central registry sync loop started")
     while True:
-        server = get_active_central_server()
-        if not server:
-            print("⚠️ لا يوجد سيرفر مركزي متاح للمزامنة")
-            time.sleep(30)
+        if not CONNECTED.is_set():
+            time.sleep(5)
             continue
         try:
-            resp = requests.get(f"{server}/peers", timeout=5)
+            resp = requests.get(f"{CURRENT_SERVER}/peers", timeout=5)
             resp.raise_for_status()
             peers_list = resp.json()
             for p in peers_list:
@@ -142,51 +159,17 @@ def fetch_central_loop():
                     PEERS.add(peer_url)
                     print(f"🌐 Central peer discovered: {peer_url}")
         except Exception as e:
-            print(f"⚠️ Fetch central peers failed on {server}: {e}")
+            print(f"⚠️ Fetch central peers failed: {e}")
         time.sleep(300)
-def connect_until_success():
-    """
-    يحاول الاتصال بكل سيرفر في CENTRAL_REGISTRY_SERVERS
-    على كل منفذ في RPORTS بالتتابع (مع تأخير بسيط).
-    لا يخرج إلا بعد نجاح التسجيل، ويُعيد السيرفر المختار وقائمة الأقران.
-    """
-    global PORT, current_server_index
-    while True:
-        for port in RPORTS:                          # جرّب كل المنافذ
-            for idx, server in enumerate(CENTRAL_REGISTRY_SERVERS):
-                info = {
-                    "node_id": os.getenv("NODE_ID", socket.gethostname()),
-                    "ip": get_local_ip(),
-                    "port": port
-                }
-                try:
-                    resp = requests.post(f"{server}/register",
-                                          json=info, timeout=5)
-                    resp.raise_for_status()          # نجاح
-                    PORT = port                      # ثبّت المنفذ الناجح
-                    current_server_index = idx       # حدّث المؤشّر
-                    print(f"✅ Connected: {server} on port {PORT}")
-                    return server, resp.json()       # peers_list
-                except Exception as e:
-                    logging.info("❌ %s:%s -> %s", server, port, e)
-        time.sleep(5)  # انتظر قليلاً ثم أَعِد الكرّة
-
-# 🚀 Main
-def main():
-    logging.basicConfig(level=logging.INFO)
-    print("🚀 Peer Discovery System starting...")
-
-    threading.Thread(target=register_service_lan, daemon=True).start()
-    threading.Thread(target=discover_lan_loop, daemon=True).start()
-
-    register_with_central()
-    threading.Thread(target=fetch_central_loop, daemon=True).start()
 
-    try:
-        while True:
-            time.sleep(60)
-    except KeyboardInterrupt:
-        print("🛑 Exiting...")
 
-if __name__ == "__main__":
-    main()
+# 🚀 تشغيل تلقائي عند الاستيراد
+def _start():
+    if not CONNECTED.is_set():
+        print("🚀 Starting peer discovery...")
+        connect_until_success()
+        threading.Thread(target=register_service_lan, daemon=True).start()
+        threading.Thread(target=discover_lan_loop, daemon=True).start()
+        threading.Thread(target=fetch_central_loop, daemon=True).start()
+
+_start()  # ← تشغيل تلقائي