How LiteRT-LM Handles Task Cancellation and Sequential Execution in the ExecutionQueue

LiteRT-LM uses a single-threaded ExecutionQueue class that guarantees strict FIFO task ordering through monotonic IDs and supports safe cancellation via the Remove() API, which only succeeds for tasks that have not yet started executing.

The LiteRT-LM runtime provides deterministic task scheduling through its ExecutionQueue implementation found in runtime/framework/execution_queue.h and execution_queue.cc. This class delivers sequential execution while allowing callers to cancel pending work, making it ideal for inference pipelines that require ordered operations with optional cleanup.

Sequential Execution Architecture

The ExecutionQueue guarantees that tasks run in the exact order they were submitted using a combination of monotonic identifiers and a dedicated worker thread.

Single-Threaded Worker Loop

The constructor spawns a background thread running WorkerThread() [lines 27-30 in execution_queue.cc]. This thread exclusively owns task consumption, ensuring that only one callable executes at a time regardless of how many threads enqueue work. The worker repeatedly waits on mutex_.Await until task_order_ contains work or the queue is shutting down [lines 78-80].

FIFO Ordering with Task IDs

Each Enqueue() call atomically increments next_id_ and stores the callable in pending_tasks_ while pushing the ID onto task_order_ [lines 58-62]. The worker thread pops IDs from the front of task_order_, guaranteeing that lower-numbered tasks execute before higher-numbered ones. The unit test ExecuteTasksInOrder verifies this behavior, confirming that tasks producing values [1,2,3] complete in strict sequence [execution_queue_test.cc, lines 33-47].

Lock-Free Task Execution

After extracting a task ID from task_order_, the worker looks up the callable in pending_tasks_, immediately erases the entry from the map, and releases the mutex before invoking the function [lines 84-95, 97-101]. Running tasks outside the critical section prevents deadlocks and allows tasks to safely interact with the queue during their execution.

Task Cancellation Mechanism

LiteRT-LM supports cooperative cancellation through the Remove() method, which eliminates pending work before the worker thread reaches it.

The Remove API

Clients receive an integer ID when calling Enqueue() and can later pass this ID to Remove(int id) [execution_queue.h]. The method attempts to erase the ID from pending_tasks_ while holding mutex_ [lines 59-66 in execution_queue.cc]. If the erase succeeds, Remove returns absl::OkStatus(); if the ID is absent, it returns absl::NotFoundError() [line 66].

Skipping Cancelled Tasks

During the worker loop, after popping an ID from task_order_, the code checks whether that ID still exists in pending_tasks_ [lines 87-94]. If Remove() has already erased the entry, the worker simply continues to the next iteration, effectively dropping the cancelled task without execution. The test RemoveTask demonstrates this: task 2 is enqueued and then removed while task 1 runs, resulting in task 2's callback never firing [execution_queue_test.cc, lines 55-80].

Edge Cases and Thread Safety

The ExecutionQueue handles shutdown and race conditions through explicit state management.

Cancelling Running Tasks

Once the worker extracts a task from pending_tasks_ [line 90], the entry no longer exists in the map. Consequently, any subsequent Remove() call returns NOT_FOUND, as demonstrated by the RemoveRunningTaskFails test [lines 87-100]. This design prevents cancellation of in-flight work, ensuring that running tasks always complete.

Graceful Shutdown

The destructor sets stop_ = true and notifies the condition variable, causing WorkerThread() to exit its loop cleanly after completing any remaining tasks [lines 31-38]. This guarantees that all enqueued work either executes or is safely discarded during destruction.

Implementation Examples

Enqueue Tasks for Sequential Execution

#include "runtime/framework/execution_queue.h"

litert::lm::ExecutionQueue queue;

queue.Enqueue([] { LOG(INFO) << "First task"; });
queue.Enqueue([] { LOG(INFO) << "Second task"; });
queue.Enqueue([] { LOG(INFO) << "Third task"; });

// Output order is guaranteed: First, Second, Third

Cancel a Pending Task

litert::lm::ExecutionQueue queue;

// Enqueue a cancellable task and retain its ID.
auto id = queue.Enqueue([] { 
  LOG(INFO) << "This task may be cancelled"; 
}).value();

// Later, before the worker reaches it:
absl::Status status = queue.Remove(id);
if (status.ok()) {
  LOG(INFO) << "Task cancelled successfully";
} else {
  LOG(ERROR) << "Cancellation failed: " << status;
}

Safe Recursive Enqueue

litert::lm::ExecutionQueue queue;

queue.Enqueue([&queue] {
  LOG(INFO) << "Outer task";
  // Safe to enqueue from within a running task because
  // the mutex is released during execution.
  queue.Enqueue([] { LOG(INFO) << "Inner task"; });
});

Summary

• The ExecutionQueue class in runtime/framework/execution_queue.cc provides single-threaded FIFO execution through a dedicated worker thread and monotonic task IDs stored in task_order_.
• Tasks are stored in pending_tasks_ and looked up by ID during the worker loop, with execution occurring outside the mutex lock to prevent deadlocks.
• The Remove() API cancels tasks only if they have not started executing, returning OK for successful cancellation or NOT_FOUND for missing or already-running tasks.
• Once a task is popped from task_order_ and removed from pending_tasks_, it cannot be cancelled and will run to completion.
• Shutdown is handled by the stop_ flag, which allows the worker to exit cleanly after finishing current work without leaving dangling threads.

Frequently Asked Questions

Can I cancel a task that is currently running in the LiteRT-LM ExecutionQueue?

No. Once the worker thread begins executing a task, it removes the ID from pending_tasks_ before invoking the callable [line 90 in execution_queue.cc]. Any Remove() call at this point returns NOT_FOUND, ensuring that running tasks complete without interruption. This behavior is verified by the RemoveRunningTaskFails unit test [execution_queue_test.cc, lines 87-100].

How does the ExecutionQueue maintain strict ordering across multiple producer threads?

The queue uses an atomic next_id_ counter and a mutex-protected task_order_ queue. All Enqueue() operations acquire the mutex to insert tasks in chronological order, while the single worker thread exclusively consumes from the front of task_order_, guaranteeing FIFO execution regardless of how many threads submit work.

Is it safe to call Enqueue from within a task running on the ExecutionQueue?

Yes. Because WorkerThread() releases mutex_ before executing the callable [lines 97-101 in execution_queue.cc], tasks can safely enqueue additional work or remove other pending tasks without risking deadlock. The task ID lookup and removal from pending_tasks_ happens while holding the lock, but the actual invocation happens after release.

What happens to pending tasks when the ExecutionQueue is destroyed?

The destructor sets stop_ = true and wakes the worker thread [lines 31-38]. The worker exits its loop cleanly, and any tasks remaining in task_order_ are discarded without execution since their corresponding entries in pending_tasks_ are destroyed along with the queue object. The destructor joins the worker thread to ensure clean shutdown.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}