cleveragents-core/.opencode/skills/programming-patterns/references/concurrency/active-object.md

Active Object Pattern

Problem

A client calls methods on an object, but those methods involve expensive or blocking operations. Executing them synchronously blocks the caller. Manually managing threads for each call is complex and error-prone.

Solution

The Active Object pattern decouples method invocation from method execution. Method calls are turned into request objects placed on a queue. A scheduler (running in its own thread) dequeues requests and executes them. Results are returned via futures/promises. The caller never blocks on execution.

When to Use

• When method calls should be non-blocking for the caller
• When execution order must be controlled (priority queues, scheduling)
• When you want to serialize access to a resource without explicit locking
• Middleware, proxy layers, or command-based architectures

When to Avoid

• Simple synchronous operations with negligible latency
• When the indirection and queuing overhead is not justified
• Real-time systems where queuing introduces unacceptable delay

Pseudocode

class ActiveObject:
    queue = RequestQueue()
    scheduler = Thread(run=process_loop)

    function async_method(args):
        future = Future()
        queue.put(Request(method, args, future))
        return future

    function process_loop():
        while running:
            request = queue.get()
            result = request.method(request.args)
            request.future.set_result(result)

// Usage
obj = ActiveObject()
f1 = obj.async_method("compute", data)
f2 = obj.async_method("transform", data)
print(f1.get())
print(f2.get())

Python

import threading
import queue
from concurrent.futures import Future
from typing import Callable, Any

class ActiveObject:
    def __init__(self, name: str):
        self.name = name
        self._queue: queue.Queue = queue.Queue()
        self._running = True
        self._thread = threading.Thread(target=self._scheduler, daemon=True)
        self._thread.start()

    def _scheduler(self) -> None:
        """Continuously dequeue and execute requests."""
        while self._running:
            try:
                func, args, future = self._queue.get(timeout=0.1)
                try:
                    result = func(*args)
                    future.set_result(result)
                except Exception as e:
                    future.set_exception(e)
            except queue.Empty:
                continue

    def invoke(self, func: Callable, *args: Any) -> Future:
        """Submit a method request; returns a Future."""
        future: Future = Future()
        self._queue.put((func, args, future))
        return future

    def shutdown(self) -> None:
        self._running = False
        self._thread.join()

# Domain methods (executed by scheduler thread)
def compute_sum(a: int, b: int) -> int:
    import time
    time.sleep(0.1)  # simulate work
    result = a + b
    print(f"[Scheduler] compute_sum({a}, {b}) = {result}")
    return result

def format_greeting(name: str) -> str:
    import time
    time.sleep(0.05)
    result = f"Hello, {name}!"
    print(f"[Scheduler] format_greeting({name}) = {result}")
    return result

def main() -> None:
    active = ActiveObject("Worker")

    # Non-blocking invocations
    f1 = active.invoke(compute_sum, 10, 20)
    f2 = active.invoke(compute_sum, 3, 7)
    f3 = active.invoke(format_greeting, "World")

    print("[Main] Requests submitted, waiting for results...")

    # Retrieve results (blocks only when result not yet ready)
    print(f"[Main] Result 1: {f1.result()}")
    print(f"[Main] Result 2: {f2.result()}")
    print(f"[Main] Result 3: {f3.result()}")

    active.shutdown()
    print("[Main] Active object shut down.")

if __name__ == "__main__":
    main()

Go

package main

import (
	"fmt"
	"time"
)

// Request holds a function to execute and a channel for the result
type Request struct {
	Execute func() interface{}
	Result  chan interface{}
}

// ActiveObject processes requests sequentially
type ActiveObject struct {
	queue chan Request
	done  chan struct{}
}

func NewActiveObject(bufferSize int) *ActiveObject {
	ao := &ActiveObject{
		queue: make(chan Request, bufferSize),
		done:  make(chan struct{}),
	}
	go ao.scheduler()
	return ao
}

func (ao *ActiveObject) scheduler() {
	for req := range ao.queue {
		result := req.Execute()
		req.Result <- result
	}
	close(ao.done)
}

// Invoke submits a request and returns a channel for the result
func (ao *ActiveObject) Invoke(fn func() interface{}) chan interface{} {
	resultCh := make(chan interface{}, 1)
	ao.queue <- Request{Execute: fn, Result: resultCh}
	return resultCh
}

func (ao *ActiveObject) Shutdown() {
	close(ao.queue)
	<-ao.done
}

func main() {
	ao := NewActiveObject(10)

	// Submit requests (non-blocking)
	f1 := ao.Invoke(func() interface{} {
		time.Sleep(100 * time.Millisecond)
		result := 10 + 20
		fmt.Printf("[Scheduler] compute_sum(10, 20) = %d\n", result)
		return result
	})

	f2 := ao.Invoke(func() interface{} {
		time.Sleep(50 * time.Millisecond)
		result := 3 + 7
		fmt.Printf("[Scheduler] compute_sum(3, 7) = %d\n", result)
		return result
	})

	f3 := ao.Invoke(func() interface{} {
		time.Sleep(50 * time.Millisecond)
		result := "Hello, World!"
		fmt.Printf("[Scheduler] format_greeting = %s\n", result)
		return result
	})

	fmt.Println("[Main] Requests submitted, waiting for results...")

	fmt.Printf("[Main] Result 1: %v\n", <-f1)
	fmt.Printf("[Main] Result 2: %v\n", <-f2)
	fmt.Printf("[Main] Result 3: %v\n", <-f3)

	ao.Shutdown()
	fmt.Println("[Main] Active object shut down.")
}

JavaScript

class ActiveObject {
  constructor(name) {
    this.name = name;
    this.queue = [];
    this.processing = false;
  }

  invoke(fn) {
    return new Promise((resolve, reject) => {
      this.queue.push({ fn, resolve, reject });
      this._processNext();
    });
  }

  async _processNext() {
    if (this.processing || this.queue.length === 0) return;
    this.processing = true;

    while (this.queue.length > 0) {
      const { fn, resolve, reject } = this.queue.shift();
      try {
        const result = await fn();
        resolve(result);
      } catch (err) {
        reject(err);
      }
    }

    this.processing = false;
  }
}

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function computeSum(a, b) {
  await sleep(100);
  const result = a + b;
  console.log(`[Scheduler] computeSum(${a}, ${b}) = ${result}`);
  return result;
}

async function formatGreeting(name) {
  await sleep(50);
  const result = `Hello, ${name}!`;
  console.log(`[Scheduler] formatGreeting(${name}) = ${result}`);
  return result;
}

async function main() {
  const active = new ActiveObject("Worker");

  // Submit requests (returns promises immediately)
  const f1 = active.invoke(() => computeSum(10, 20));
  const f2 = active.invoke(() => computeSum(3, 7));
  const f3 = active.invoke(() => formatGreeting("World"));

  console.log("[Main] Requests submitted, waiting for results...");

  const [r1, r2, r3] = await Promise.all([f1, f2, f3]);
  console.log(`[Main] Result 1: ${r1}`);
  console.log(`[Main] Result 2: ${r2}`);
  console.log(`[Main] Result 3: ${r3}`);
  console.log("[Main] Active object shut down.");
}

main();

Related Patterns

• Future/Promise — active object method calls return futures; callers await results without blocking on execution.
• Thread Pool — the active object's scheduler runs on a thread pool worker thread.
• Command — each method invocation is turned into a command object enqueued in the active object's queue.
• Proxy — the active object proxy intercepts synchronous method calls and converts them to async requests.