update dags

dags/test-train-pytorch.py

@@ -78,7 +78,29 @@ def run_training_node_func(rank, world_size):
   # Small delay to ensure all workers see the ready state
   time.sleep(3)
 
-  # STEP 3: Configure distributed environment
+  # STEP 3: Configure distributed Environment
+
+  import socket
+
+  if rank == 0:
+      # Rank 0 becomes master
+      master_addr = socket.gethostbyname(socket.gethostname())
+      Variable.set("MASTER_ADDR_DYNAMIC", master_addr)
+      print(f"[{rank}] Acting as MASTER at {master_addr}")
+  else:
+      # Other ranks wait for master
+      print(f"[{rank}] Waiting for MASTER_ADDR...")
+      while True:
+          try:
+              master_addr = Variable.get("MASTER_ADDR_DYNAMIC", default_var=None)
+              if master_addr:
+                  break
+          except:
+              pass
+          time.sleep(1)
+
+      print(f"[{rank}] Found MASTER at {master_addr}")
+
   os.environ['MASTER_ADDR'] = MASTER_ADDR
   os.environ['MASTER_PORT'] = MASTER_PORT
   os.environ['WORLD_SIZE'] = str(world_size)