Class: RCrewAI::AsyncExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/rcrewai/async_executor.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(**options) ⇒ AsyncExecutor

Returns a new instance of AsyncExecutor.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/rcrewai/async_executor.rb', line 10

def initialize(**options)
  @max_concurrency = options.fetch(:max_concurrency, Concurrent.processor_count)
  @timeout = options.fetch(:timeout, 300) # 5 minutes default
  @logger = Logger.new($stdout)
  @logger.level = options.fetch(:verbose, false) ? Logger::DEBUG : Logger::INFO

  # Create thread pool for task execution
  @thread_pool = Concurrent::ThreadPoolExecutor.new(
    min_threads: 1,
    max_threads: @max_concurrency,
    max_queue: @max_concurrency * 2,
    fallback_policy: :caller_runs
  )

  @futures = {}
  @task_dependencies = {}
  @completed_tasks = Concurrent::Set.new
  @failed_tasks = Concurrent::Set.new
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



8
9
10
 
# File 'lib/rcrewai/async_executor.rb', line 8

def logger
  @logger
end

#max_concurrency ⇒ Object (readonly)

Returns the value of attribute max_concurrency.



8
9
10
 
# File 'lib/rcrewai/async_executor.rb', line 8

def max_concurrency
  @max_concurrency
end

#thread_pool ⇒ Object (readonly)

Returns the value of attribute thread_pool.



8
9
10
 
# File 'lib/rcrewai/async_executor.rb', line 8

def thread_pool
  @thread_pool
end

Instance Method Details

#execute_single_task_async(task) ⇒ Object 



59
60
61
62
63
64
65
66
 
# File 'lib/rcrewai/async_executor.rb', line 59

def execute_single_task_async(task)
  future = Concurrent::Future.execute(executor: @thread_pool) do
    execute_task_with_monitoring(task)
  end

  @futures[task] = future
  future
end

#execute_tasks_async(tasks, dependency_graph = {}) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/rcrewai/async_executor.rb', line 30

def execute_tasks_async(tasks, dependency_graph = {})
  @logger.info "Starting async execution of #{tasks.length} tasks with max #{@max_concurrency} concurrent threads"

  @task_dependencies = dependency_graph
  execution_phases = organize_tasks_by_dependencies(tasks)

  start_time = Time.now
  results = []

  execution_phases.each_with_index do |phase_tasks, phase_index|
    @logger.info "Executing phase #{phase_index + 1}: #{phase_tasks.length} tasks"

    phase_results = execute_phase_concurrently(phase_tasks, phase_index + 1)
    results.concat(phase_results)

    # Check if we should continue
    failed_in_phase = phase_results.count { |r| r[:status] == :failed }
    if failed_in_phase.positive? && should_abort_after_failures?(failed_in_phase, phase_tasks.length)
      @logger.error "Aborting execution due to #{failed_in_phase} failures in phase #{phase_index + 1}"
      break
    end
  end

  total_time = Time.now - start_time
  @logger.info "Async execution completed in #{total_time.round(2)}s"

  format_async_results(results, total_time)
end

#shutdown ⇒ Object 



86
87
88
89
90
91
92
93
 
# File 'lib/rcrewai/async_executor.rb', line 86

def shutdown
  @logger.info 'Shutting down async executor...'
  @thread_pool.shutdown
  return if @thread_pool.wait_for_termination(30)

  @logger.warn 'Thread pool did not shut down gracefully, forcing shutdown'
  @thread_pool.kill
end

#stats ⇒ Object 



95
96
97
98
99
100
101
102
103
104
 
# File 'lib/rcrewai/async_executor.rb', line 95

def stats
  {
    max_concurrency: @max_concurrency,
    active_threads: @thread_pool.length,
    queue_length: @thread_pool.queue_length,
    completed_task_count: @completed_tasks.size,
    failed_task_count: @failed_tasks.size,
    pool_shutdown: @thread_pool.shutdown?
  }
end

#wait_for_completion(futures, timeout = nil) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/rcrewai/async_executor.rb', line 68

def wait_for_completion(futures, timeout = nil)
  timeout ||= @timeout

  results = []
  futures.each do |task, future|
    result = future.value(timeout)
    results << { task: task, result: result, status: :completed }
  rescue Concurrent::TimeoutError
    @logger.error "Task #{task.name} timed out after #{timeout}s"
    results << { task: task, result: 'Task timed out', status: :timeout }
  rescue StandardError => e
    @logger.error "Task #{task.name} failed: #{e.message}"
    results << { task: task, result: e.message, status: :failed }
  end

  results
end