Skip to content

Make InvocationProcessor more robust against exceptions #3376

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ebr merged 4 commits into from
May 12, 2023
Merged

Make InvocationProcessor more robust against exceptions #3376

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
56fe8ea
fix: queue error should not crash InvocationProcessor
ebr May 10, 2023
591a8b3
Merge branch 'main' into queue-soft-lock
ebr May 12, 2023
b2d752f
Merge branch 'main' into queue-soft-lock
ebr May 12, 2023
988e199
reduce queue check interval to 0.5s
ebr May 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions invokeai/app/services/processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import time
import traceback
from threading import Event, Thread, BoundedSemaphore

Expand All @@ -6,6 +7,7 @@
from .invoker import InvocationProcessorABC, Invoker
from ..models.exceptions import CanceledException

import invokeai.backend.util.logging as logger
class DefaultInvocationProcessor(InvocationProcessorABC):
__invoker_thread: Thread
__stop_event: Event
Expand Down Expand Up @@ -34,8 +36,14 @@ def __process(self, stop_event: Event):
try:
self.__threadLimit.acquire()
while not stop_event.is_set():
queue_item: InvocationQueueItem = self.__invoker.services.queue.get()
try:
queue_item: InvocationQueueItem = self.__invoker.services.queue.get()
except Exception as e:
logger.debug("Exception while getting from queue: %s" % e)

if not queue_item: # Probably stopping
# do not hammer the queue
time.sleep(0.5)
continue

graph_execution_state = (
Expand Down Expand Up @@ -124,7 +132,16 @@ def __process(self, stop_event: Event):
# Queue any further commands if invoking all
is_complete = graph_execution_state.is_complete()
if queue_item.invoke_all and not is_complete:
self.__invoker.invoke(graph_execution_state, invoke_all=True)
try:
self.__invoker.invoke(graph_execution_state, invoke_all=True)
except Exception as e:
logger.error("Error while invoking: %s" % e)
self.__invoker.services.events.emit_invocation_error(
graph_execution_state_id=graph_execution_state.id,
node=invocation.dict(),
source_node_id=source_node_id,
error=traceback.format_exc()
)
elif is_complete:
self.__invoker.services.events.emit_graph_execution_complete(
graph_execution_state.id
Expand Down