added check for pending get requests after put data

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

@@ -118,7 +118,8 @@ def __del__(self):
     def __getstate__(self):
         """Remove a reference to garbage collector for correct `pickle` serialization."""
         attributes = self.__dict__.copy()
-        del attributes["_gc"]
+        if hasattr(self, "_gc"):
+            del attributes["_gc"]
         return attributes
 
     def base_data_id(self):

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

@@ -4,7 +4,6 @@
 
 """Worker MPI process task processing functionality."""
 import asyncio
-import sys
 from functools import wraps, partial
 
 try:
@@ -126,6 +125,8 @@ async def worker_loop():
             # Check pending actor requests also.
             task_store.check_pending_actor_tasks()
 
+            RequestStore.get_instance().check_pending_get_requests(request["id"])
+
         elif operation_type == common.Operation.PUT_OWNER:
             request = communication.recv_simple_operation(mpi_state.comm, source_rank)
             object_store.put_data_owner(request["id"], request["owner"])

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

@@ -180,7 +180,7 @@ def unwrap_local_data_id(self, arg):
                         self.request_worker_data(owner_rank, arg)
                 return arg, True
             else:
-                raise ValueError("DataID is missing!")
+                raise ValueError("DataID is missing! {}".format(arg))
         else:
             return arg, False
 
@@ -357,6 +357,9 @@ def process_task_request(self, request):
         w_logger.debug("Is pending - {}".format(is_pending))
 
         if is_pending or is_kw_pending:
+            current_rank = communication.MPIState.get_instance().rank
+            for output_id in output_ids:
+                ObjectStore.get_instance().put_data_owner(output_id, current_rank)
             request["args"] = args
             request["kwargs"] = kwargs
             return request