share retries throughout pipeline
This commit is contained in:
parent
d418cb1bc2
commit
7761aea28b
|
@ -150,7 +150,7 @@ class ChainPipeline:
|
||||||
tile_mask: Image.Image,
|
tile_mask: Image.Image,
|
||||||
dims: Tuple[int, int, int],
|
dims: Tuple[int, int, int],
|
||||||
) -> Image.Image:
|
) -> Image.Image:
|
||||||
for i in range(3):
|
for i in range(job.retries):
|
||||||
try:
|
try:
|
||||||
output_tile = stage_pipe.run(
|
output_tile = stage_pipe.run(
|
||||||
job,
|
job,
|
||||||
|
@ -167,6 +167,7 @@ class ChainPipeline:
|
||||||
if is_debug():
|
if is_debug():
|
||||||
save_image(server, "last-tile.png", output_tile)
|
save_image(server, "last-tile.png", output_tile)
|
||||||
|
|
||||||
|
job.retries = job.retries - i
|
||||||
return output_tile
|
return output_tile
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception(
|
logger.exception(
|
||||||
|
@ -188,9 +189,9 @@ class ChainPipeline:
|
||||||
stage_sources = stage_outputs
|
stage_sources = stage_outputs
|
||||||
else:
|
else:
|
||||||
logger.debug("image within tile size of %s, running stage", tile)
|
logger.debug("image within tile size of %s, running stage", tile)
|
||||||
for i in range(3):
|
for i in range(job.retries):
|
||||||
try:
|
try:
|
||||||
stage_sources = stage_pipe.run(
|
stage_outputs = stage_pipe.run(
|
||||||
job,
|
job,
|
||||||
server,
|
server,
|
||||||
stage_params,
|
stage_params,
|
||||||
|
@ -199,6 +200,10 @@ class ChainPipeline:
|
||||||
callback=callback,
|
callback=callback,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
)
|
)
|
||||||
|
# doing this on the same line as stage_pipe.run can leave sources as None, which the pipeline
|
||||||
|
# does not like, so it throws
|
||||||
|
stage_sources = stage_outputs
|
||||||
|
job.retries = job.retries - i
|
||||||
break
|
break
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception(
|
logger.exception(
|
||||||
|
|
|
@ -22,6 +22,7 @@ class WorkerContext:
|
||||||
last_progress: Optional[ProgressCommand]
|
last_progress: Optional[ProgressCommand]
|
||||||
idle: "Value[bool]"
|
idle: "Value[bool]"
|
||||||
timeout: float
|
timeout: float
|
||||||
|
retries: int
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
@ -44,6 +45,7 @@ class WorkerContext:
|
||||||
self.last_progress = None
|
self.last_progress = None
|
||||||
self.idle = idle
|
self.idle = idle
|
||||||
self.timeout = 1.0
|
self.timeout = 1.0
|
||||||
|
self.retries = 3 # TODO: get from env
|
||||||
|
|
||||||
def start(self, job: str) -> None:
|
def start(self, job: str) -> None:
|
||||||
self.job = job
|
self.job = job
|
||||||
|
|
Loading…
Reference in New Issue