diff --git a/lwr/lwr_client/client.py b/lwr/lwr_client/client.py index a4ee845b710d203c7f42c753e8a9d90104667a02..0aef72a597744ab1d92d21929b7e2b2af8887218 100644 --- a/lwr/lwr_client/client.py +++ b/lwr/lwr_client/client.py @@ -308,13 +308,15 @@ class InputCachingClient(Client): cache_required = self.cache_required(input_path) if cache_required: self.client_manager.queue_transfer(self, input_path) - while True: + while not event_holder.failed: available = self.file_available(input_path) if available['ready']: token = available['token'] args["cache_token"] = token return self._raw_execute(action, args) event_holder.event.wait(30) + if event_holder.failed: + raise Exception("Failed to transfer file %s" % input_path) @parseJson() def cache_required(self, path): diff --git a/lwr/lwr_client/manager.py b/lwr/lwr_client/manager.py index 738a971bbd24d3ebb4af3808855c93e39418236c..cb74c6022338f126a9ab337e158d3c823111820b 100644 --- a/lwr/lwr_client/manager.py +++ b/lwr/lwr_client/manager.py @@ -51,8 +51,13 @@ class ClientManager(object): def __perform_transfer(self, transfer_info): (client, path) = transfer_info event_holder = self.event_manager.acquire_event(path, force_clear=True) - client.cache_insert(path) - event_holder.event.set() + failed = True + try: + client.cache_insert(path) + failed = False + finally: + event_holder.failed = failed + event_holder.release() def __init_transfer_threads(self, num_transfer_threads): self.num_transfer_threads = num_transfer_threads diff --git a/lwr/lwr_client/util.py b/lwr/lwr_client/util.py index 001db19c2a946f2baeef8e179377c675c5657411..6a610a58c02b78e65e0dc8eb6d7ed9eb7d7fa987 100644 --- a/lwr/lwr_client/util.py +++ b/lwr/lwr_client/util.py @@ -26,6 +26,10 @@ class EventHolder(object): self.event = event self.path = path self.condition_manager = condition_manager + self.failed = False def release(self): self.event.set() + + def fail(self): + self.failed = True