Class: Taski::Execution::TaskWrapper

Inherits:
Object
  • Object
show all
Defined in:
lib/taski/execution/task_wrapper.rb

Overview

Manages state and synchronization for a single task instance. Does NOT start threads or fibers — Executor and WorkerPool control scheduling.

State transitions (both run and clean phases):

pending -> running -> completed | failed
pending -> skipped (run-phase only)

Constant Summary collapse

STATE_PENDING = 
:pending
STATE_RUNNING = 
:running
STATE_COMPLETED = 
:completed
STATE_FAILED = 
:failed
STATE_SKIPPED = 
:skipped

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(task, registry:, execution_facade: nil, args: nil) ⇒ TaskWrapper

Returns a new instance of TaskWrapper.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/taski/execution/task_wrapper.rb', line 22

def initialize(task, registry:, execution_facade: nil, args: nil)
  @task = task
  @registry = registry
  @execution_facade = execution_facade
  @args = args
  @result = nil
  @clean_result = nil
  @error = nil
  @clean_error = nil
  @monitor = Monitor.new
  @condition = @monitor.new_cond
  @clean_condition = @monitor.new_cond
  @state = STATE_PENDING
  @clean_state = STATE_PENDING
  @waiters = []
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(method_name, *args, &block) ⇒ Object 



208
209
210
211
212
213
214
 
# File 'lib/taski/execution/task_wrapper.rb', line 208

def method_missing(method_name, *args, &block)
  if @task.class.exported_methods.include?(method_name)
    get_exported_value(method_name)
  else
    super
  end
end

Instance Attribute Details

#clean_error ⇒ Object (readonly)

Returns the value of attribute clean_error.



14
15
16
 
# File 'lib/taski/execution/task_wrapper.rb', line 14

def clean_error
  @clean_error
end

#error ⇒ Object (readonly)

Returns the value of attribute error.



14
15
16
 
# File 'lib/taski/execution/task_wrapper.rb', line 14

def error
  @error
end

#result ⇒ Object (readonly)

Returns the value of attribute result.



14
15
16
 
# File 'lib/taski/execution/task_wrapper.rb', line 14

def result
  @result
end

#task ⇒ Object (readonly)

Returns the value of attribute task.



14
15
16
 
# File 'lib/taski/execution/task_wrapper.rb', line 14

def task
  @task
end

Instance Method Details

#clean ⇒ Object 



69
70
71
72
73
74
 
# File 'lib/taski/execution/task_wrapper.rb', line 69

def clean
  with_args_lifecycle do
    trigger_clean_and_wait
    @clean_result
  end
end

#completed? ⇒ Boolean

Returns:

  • (Boolean) 


44
 
# File 'lib/taski/execution/task_wrapper.rb', line 44

def completed? = state == STATE_COMPLETED

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


45
 
# File 'lib/taski/execution/task_wrapper.rb', line 45

def failed? = state == STATE_FAILED

#get_exported_value(method_name) ⇒ Object 



92
93
94
95
96
97
98
 
# File 'lib/taski/execution/task_wrapper.rb', line 92

def get_exported_value(method_name)
  with_args_lifecycle do
    trigger_execution_and_wait
    raise @error if @error # steep:ignore
    @task.public_send(method_name)
  end
end

#mark_clean_completed(result) ⇒ Object 



178
179
180
181
182
183
184
185
 
# File 'lib/taski/execution/task_wrapper.rb', line 178

def mark_clean_completed(result)
  @monitor.synchronize do
    @clean_result = result
    @clean_state = STATE_COMPLETED
    @clean_condition.broadcast
  end
  update_clean_progress(:completed)
end

#mark_clean_failed(error) ⇒ Object 



187
188
189
190
191
192
193
194
 
# File 'lib/taski/execution/task_wrapper.rb', line 187

def mark_clean_failed(error)
  @monitor.synchronize do
    @clean_error = error
    @clean_state = STATE_FAILED
    @clean_condition.broadcast
  end
  update_clean_progress(:failed)
end

#mark_clean_running ⇒ Object 



170
171
172
173
174
175
176
 
# File 'lib/taski/execution/task_wrapper.rb', line 170

def mark_clean_running
  @monitor.synchronize do
    return false unless @clean_state == STATE_PENDING
    @clean_state = STATE_RUNNING
    true
  end
end

#mark_completed(result) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/taski/execution/task_wrapper.rb', line 134

def mark_completed(result)
  waiters_to_notify = nil
  @monitor.synchronize do
    @result = result
    @state = STATE_COMPLETED
    @condition.broadcast
    waiters_to_notify = @waiters.dup
    @waiters.clear
  end
  notify_fiber_waiters_completed(waiters_to_notify)
  update_progress(:completed)
end

#mark_failed(error) ⇒ Object 



147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/taski/execution/task_wrapper.rb', line 147

def mark_failed(error)
  waiters_to_notify = nil
  @monitor.synchronize do
    @error = error
    @state = STATE_FAILED
    @condition.broadcast
    waiters_to_notify = @waiters.dup
    @waiters.clear
  end
  notify_fiber_waiters_failed(waiters_to_notify, error)
  update_progress(:failed)
end

#mark_running ⇒ Object 



126
127
128
129
130
131
132
 
# File 'lib/taski/execution/task_wrapper.rb', line 126

def mark_running
  @monitor.synchronize do
    return false unless @state == STATE_PENDING
    @state = STATE_RUNNING
    true
  end
end

#mark_skipped ⇒ Object 



160
161
162
163
164
165
166
167
168
 
# File 'lib/taski/execution/task_wrapper.rb', line 160

def mark_skipped
  @monitor.synchronize do
    return false unless @state == STATE_PENDING
    @state = STATE_SKIPPED
    @condition.broadcast
  end
  notify_skipped
  true
end

#pending? ⇒ Boolean

Returns:

  • (Boolean) 


43
 
# File 'lib/taski/execution/task_wrapper.rb', line 43

def pending? = state == STATE_PENDING

#request_value(method, thread_queue, fiber) ⇒ Object

Atomically resolve the dependency value for a waiting Fiber. Returns a status tuple indicating how the caller should proceed:

  • [:completed, value] → dependency already done, resume immediately
  • [:failed, error] → dependency failed, propagate error
  • [:wait] → dependency running, Fiber parked (will be resumed via thread_queue)
  • [:start] → dependency was PENDING, now RUNNING (caller must drive it)


106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/taski/execution/task_wrapper.rb', line 106

def request_value(method, thread_queue, fiber)
  @monitor.synchronize do
    case @state
    when STATE_COMPLETED
      value = @task.public_send(method)
      [:completed, value]
    when STATE_FAILED
      [:failed, @error]
    when STATE_RUNNING
      @waiters << [thread_queue, fiber, method]
      [:wait]
    else
      # PENDING → atomically transition to RUNNING
      @state = STATE_RUNNING
      @waiters << [thread_queue, fiber, method]
      [:start]
    end
  end
end

#reset! ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/taski/execution/task_wrapper.rb', line 48

def reset!
  @monitor.synchronize do
    @state = STATE_PENDING
    @clean_state = STATE_PENDING
    @result = nil
    @clean_result = nil
    @error = nil
    @clean_error = nil
  end
  @task.reset! if @task.respond_to?(:reset!)
  @registry.reset!
end

#respond_to_missing?(method_name, include_private = false) ⇒ Boolean

Returns:

  • (Boolean) 


216
217
218
 
# File 'lib/taski/execution/task_wrapper.rb', line 216

def respond_to_missing?(method_name, include_private = false)
  @task.class.exported_methods.include?(method_name) || super
end

#run ⇒ Object 



61
62
63
64
65
66
67
 
# File 'lib/taski/execution/task_wrapper.rb', line 61

def run
  with_args_lifecycle do
    trigger_execution_and_wait
    raise @error if @error # steep:ignore
    @result
  end
end

#run_and_clean(clean_on_failure: false, &block) ⇒ Object

Runs execution followed by cleanup. Block is called between phases.

Parameters:

  • clean_on_failure (Boolean) (defaults to: false) —

    When true, clean runs even if run raises. Default is false (clean is skipped on run failure).



79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/taski/execution/task_wrapper.rb', line 79

def run_and_clean(clean_on_failure: false, &block)
  facade = ensure_facade
  facade.notify_start # Pre-increment nest_level to prevent double rendering
  run_succeeded = false
  result = run
  run_succeeded = true
  block&.call
  result
ensure
  clean if run_succeeded || clean_on_failure
  facade&.notify_stop # Final decrement and render
end

#skipped? ⇒ Boolean

Returns:

  • (Boolean) 


46
 
# File 'lib/taski/execution/task_wrapper.rb', line 46

def skipped? = state == STATE_SKIPPED

#state ⇒ Object 



39
40
41
 
# File 'lib/taski/execution/task_wrapper.rb', line 39

def state
  @monitor.synchronize { @state }
end

#wait_for_clean_completion ⇒ Object 



202
203
204
205
206
 
# File 'lib/taski/execution/task_wrapper.rb', line 202

def wait_for_clean_completion
  @monitor.synchronize do
    @clean_condition.wait_until { @clean_state == STATE_COMPLETED || @clean_state == STATE_FAILED }
  end
end

#wait_for_completion ⇒ Object 



196
197
198
199
200
 
# File 'lib/taski/execution/task_wrapper.rb', line 196

def wait_for_completion
  @monitor.synchronize do
    @condition.wait_until { @state == STATE_COMPLETED || @state == STATE_FAILED || @state == STATE_SKIPPED }
  end
end