diff --git a/ventis/controller/local_controller.py b/ventis/controller/local_controller.py index 9f53a3e..3d74341 100644 --- a/ventis/controller/local_controller.py +++ b/ventis/controller/local_controller.py @@ -61,6 +61,9 @@ def __init__(self, port=50051): self.server, self.servicer = start_server(port, my_endpoint=self._my_endpoint) self.request_queue = self.servicer.request_queue + # Let the gRPC servicer fan a result out to this node's consumers as + # soon as it arrives via WriteResult (see _fan_out_to_consumers). + self.servicer.on_result = self._fan_out_to_consumers # Connect to Redis and report healthy status redis_host = os.environ.get("VENTIS_REDIS_HOST", "localhost") @@ -314,6 +317,10 @@ def _mark_future_failed(self, future_id, error, origin=None): {"failed": 1, "error_message": error_message}, ) + # Unblock any consumers waiting on this future so a failed dependency + # surfaces as an error instead of a 300s timeout. + self._fan_out_to_consumers(future_id, failed=1, error_message=error_message) + if origin and origin != self._my_endpoint: self._send_result_callback( origin, @@ -427,7 +434,9 @@ def _process_request(self, data): ) # If the result is already available, push it immediately. - # This handles the race where _notify_consumers already ran. + # This handles the race where the producer resolved the + # future before this consumer registered (and thus before + # _fan_out_to_consumers could see it). existing_result = self.redis.hget(future_key, "result") if existing_result is not None and existing_result != "": logger.info( @@ -586,6 +595,9 @@ def _execute_locally( # Write result to local Redis self.redis.hset(f"future:{future_id}", "result", serialized) + # Push the result to any consumers registered on this node. + self._fan_out_to_consumers(future_id, result=serialized) + # If the request came from another node, send result back to origin if origin and origin != self._my_endpoint: self._send_result_callback( @@ -700,6 +712,33 @@ def _send_result_callback( logger.error("Failed to send result callback to %s: %s", origin, e) self._mark_future_failed(future_id, f"Result callback failed: {e}") + def _fan_out_to_consumers(self, future_id, result=None, failed=0, error_message=""): + """Push a completed future (result or failure) to every endpoint registered + as a consumer on THIS node's Redis. + + Called at every site that writes a terminal value for a future -- local + production (_execute_locally), failure (_mark_future_failed), and remote + arrival (WriteResult) -- so propagation is event-driven and never depends + on anyone calling Future.value(). Whichever node holds the consumer set is + always either the producer or the origin, so one of those sites fires; the + value then re-fans-out at each node it lands on, walking the graph. + + Delivery is intentionally not deduped: a consumer's WriteResult just + re-writes the same value, so a redundant push (e.g. racing the immediate + push in _process_request) is idempotent and harmless. + """ + if not future_id: + return + for endpoint in self.redis.smembers(f"future:{future_id}:consumers"): + if endpoint and endpoint != self._my_endpoint: + self._send_result_callback( + endpoint, + future_id, + result=result, + failed=failed, + error_message=error_message, + ) + # ------------------------------------------------------------------ # # Shutdown # # ------------------------------------------------------------------ # diff --git a/ventis/controller/local_controller_frontend.py b/ventis/controller/local_controller_frontend.py index 0117c4a..2e2de59 100644 --- a/ventis/controller/local_controller_frontend.py +++ b/ventis/controller/local_controller_frontend.py @@ -36,6 +36,9 @@ def __init__(self, my_endpoint="unknown"): except ImportError: from redis_client import RedisClient self.redis = RedisClient(host=redis_host, port=redis_port) + # Set by LocalController: fans a just-arrived result out to this node's + # consumers. Signature: on_result(future_id, result, failed, error_message). + self.on_result = None def Execute(self, request, context): """Accept an Execute request and push it into the queue.""" @@ -78,6 +81,16 @@ def WriteResult(self, request, context): future_id, result, ) + # Relay the just-arrived value to any consumers registered on + # this node (the origin is where consumer sets live). This is + # what walks the value hop-by-hop through the graph. + if self.on_result: + self.on_result( + future_id, + result=result, + failed=failed, + error_message=error_message, + ) else: logger.error("WriteResult: missing future_id in %s", data) except Exception as e: diff --git a/ventis/future.py b/ventis/future.py index 598bc4f..6c0cfa1 100644 --- a/ventis/future.py +++ b/ventis/future.py @@ -176,9 +176,9 @@ def value(self, timeout=None): ) self.calculated = True - # Push result to all consumers - self._notify_consumers() - + # Note: consumer fan-out is handled entirely by the local controllers at + # result-write time (see LocalController._fan_out_to_consumers). value() + # is now purely a top-level pull for whoever holds the root future. return self.result # def __call__(self, timeout=None): @@ -191,39 +191,6 @@ def is_available(self): """ return self.result is not None - def _get_consumers(self): - """Return the list of consumers from Redis.""" - return self.redis.smembers(self._consumers_key()) - - def _notify_consumers(self): - """Push this future's result to all registered consumer endpoints via gRPC WriteResult.""" - consumers = self._get_consumers() - if not consumers: - return - for endpoint in consumers: - try: - if not self.result: - logger.warning( - "Future %s is notifying consumer %s with an empty/None result", - self.id, - endpoint, - ) - channel = grpc.insecure_channel(endpoint) - stub = local_controler_pb2_grpc.LocalControllerStub(channel) - payload = json.dumps({"future_id": self.id, "result": self.result}) - request = local_controler_pb2.JsonResponse(resonse=payload) - stub.WriteResult(request) - logger.info( - "Notified consumer %s with result for future %s", endpoint, self.id - ) - except Exception as e: - logger.error( - "Failed to notify consumer %s for future %s: %s", - endpoint, - self.id, - e, - ) - def _add_consumer(self, consumer): """Add a consumer.""" self.consumers.append(consumer)