debug logs

Signed-off-by: arun696 <[email protected]>

unidist/core/backends/mpi/core/controller/api.py

@@ -380,6 +380,7 @@ def submit(task, *args, num_returns=1, **kwargs):
 
     # dest_rank = RoundRobin.get_instance().schedule_rank()
     dest_rank = choose_destination_rank(collected_data_ids)
+    print(dest_rank, collected_data_ids)
     push_data_owners(dest_rank, collected_data_ids)
     output_ids = object_store.generate_output_data_id(
         dest_rank, garbage_collector, num_returns

unidist/core/backends/mpi/core/controller/common.py

@@ -282,7 +282,8 @@ def choose_destination_rank(data_ids):
     """
     Choose destination rank considering which worker has the maximum share of data.
 
-    Get the data shares in each worker process
+    Get the data shares in each worker process and choose rank with the most data share.
+    If data_ids are empty choses the rank using the round robin approach.
 
     Parameters
     ----------
@@ -294,13 +295,16 @@ def choose_destination_rank(data_ids):
     int
         A rank number.
     """
+    if not data_ids:
+        chosen_rank = RoundRobin.get_instance().schedule_rank()
+        return chosen_rank
     data_share = defaultdict(lambda: 0)
     for data_id in data_ids:
         data_share[object_store.get_data_owner(data_id)] += object_store.get_data_size(
             data_id
         )
-    rank_with_max_data = max(data_share, key=data_share.get)
-    return rank_with_max_data
+    chosen_rank = max(data_share, key=data_share.get)
+    return chosen_rank
 
 
 def collect_all_data_id_from_args(value, collected_data_ids=[]):

unidist/core/backends/mpi/core/worker/loop.py

@@ -4,6 +4,7 @@
 
 """Worker MPI process task processing functionality."""
 import asyncio
+import sys
 from functools import wraps, partial
 
 try:
@@ -29,7 +30,7 @@
 # When building documentation we do not have MPI initialized so
 # we use the condition to set "worker_0.log" in order to build it succesfully.
 log_file = "worker_{}.log".format(mpi_state.rank if mpi_state is not None else 0)
-w_logger = common.get_logger("worker", log_file)
+w_logger = common.get_logger("worker", sys.stdout, True)
 
 # Actors map {handle : actor}
 actor_map = {}