Class: Celluloid::Task

Inherits:
Object
  • Object
show all
Defined in:
lib/celluloid/task.rb,
lib/celluloid/task/fibered.rb,
lib/celluloid/task/threaded.rb

Overview

Tasks are interruptable/resumable execution contexts used to run methods

Direct Known Subclasses

Fibered, Threaded

Defined Under Namespace

Classes: Fibered, Threaded

Constant Summary collapse

TerminatedError = 
TaskTerminated
TimeoutError = 
TaskTimeout

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(type, meta) ⇒ Task

Create a new task



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/celluloid/task.rb', line 18

def initialize(type, meta)
  @type     = type
  @meta     = meta
  @status   = :new

  @exclusive         = false
  @dangerous_suspend = @meta ? @meta.dup.delete(:dangerous_suspend) : false
  @guard_warnings    = false

  actor     = Thread.current[:celluloid_actor]
  @chain_id = Internals::CallChain.current_id

  fail NotActorError, "can't create tasks outside of actors" unless actor
  guard "can't create tasks inside of tasks" if Thread.current[:celluloid_task]

  create do
    begin
      @status = :running
      actor.setup_thread

      name_current_thread thread_metadata

      Thread.current[:celluloid_task] = self
      Internals::CallChain.current_id = @chain_id

      actor.tasks << self
      yield
    rescue TaskTerminated
      # Task was explicitly terminated
    ensure
      name_current_thread nil
      @status = :dead
      actor.tasks.delete self
    end
  end
end

Instance Attribute Details

#chain_id ⇒ Object

Returns the value of attribute chain_id.



15
16
17
 
# File 'lib/celluloid/task.rb', line 15

def chain_id
  @chain_id
end

#guard_warnings ⇒ Object

Returns the value of attribute guard_warnings.



15
16
17
 
# File 'lib/celluloid/task.rb', line 15

def guard_warnings
  @guard_warnings
end

#meta ⇒ Object (readonly)

Returns the value of attribute meta.



14
15
16
 
# File 'lib/celluloid/task.rb', line 14

def meta
  @meta
end

#status ⇒ Object (readonly)

Returns the value of attribute status.



14
15
16
 
# File 'lib/celluloid/task.rb', line 14

def status
  @status
end

#type ⇒ Object (readonly)

Returns the value of attribute type.



14
15
16
 
# File 'lib/celluloid/task.rb', line 14

def type
  @type
end

Class Method Details

.current ⇒ Object

Obtain the current task



5
6
7
 
# File 'lib/celluloid/task.rb', line 5

def self.current
  Thread.current[:celluloid_task] || fail(NotTaskError, "not within a task context")
end

.suspend(status) ⇒ Object

Suspend the running task, deferring to the scheduler



10
11
12
 
# File 'lib/celluloid/task.rb', line 10

def self.suspend(status)
  Task.current.suspend(status)
end

Instance Method Details

#backtrace ⇒ Object 



128
129
 
# File 'lib/celluloid/task.rb', line 128

def backtrace
end

#create(&_block) ⇒ Object 



55
56
57
 
# File 'lib/celluloid/task.rb', line 55

def create(&_block)
  fail "Implement #{self.class}#create"
end

#exclusive ⇒ Object

Execute a code block in exclusive mode.



91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/celluloid/task.rb', line 91

def exclusive
  if @exclusive
    yield
  else
    begin
      @exclusive = true
      yield
    ensure
      @exclusive = false
    end
  end
end

#exclusive? ⇒ Boolean

Is this task running in exclusive mode?

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/celluloid/task.rb', line 124

def exclusive?
  @exclusive
end

#guard(message) ⇒ Object 



141
142
143
144
145
146
147
 
# File 'lib/celluloid/task.rb', line 141

def guard(message)
  if @guard_warnings
    Internals::Logger.warn message if $CELLULOID_DEBUG
  else
    fail message if $CELLULOID_DEBUG
  end
end

#inspect ⇒ Object

Nicer string inspect for tasks



137
138
139
 
# File 'lib/celluloid/task.rb', line 137

def inspect
  "#<#{self.class}:0x#{object_id.to_s(16)} @type=#{@type.inspect}, @meta=#{@meta.inspect}, @status=#{@status.inspect}>"
end

#resume(value = nil) ⇒ Object

Resume a suspended task, giving it a value to return if needed



80
81
82
83
84
85
86
87
88
 
# File 'lib/celluloid/task.rb', line 80

def resume(value = nil)
  guard "Cannot resume a task from inside of a task" if Thread.current[:celluloid_task]
  if running?
    deliver(value)
  else
    Internals::Logger.warn "Attempted to resume a dead task: type=#{@type.inspect}, meta=#{@meta.inspect}, status=#{@status.inspect}"
  end
  nil
end

#running? ⇒ Boolean

Is the current task still running?

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/celluloid/task.rb', line 132

def running?
  @status != :dead
end

#suspend(status) ⇒ Object

Suspend the current task, changing the status to the given argument



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/celluloid/task.rb', line 60

def suspend(status)
  fail "Cannot suspend while in exclusive mode" if exclusive?
  fail "Cannot suspend a task from outside of itself" unless Task.current == self

  @status = status

  if $CELLULOID_DEBUG && @dangerous_suspend
    Internals::Logger.with_backtrace(caller[2...8]) do |logger|
      logger.warn "Dangerously suspending task: type=#{@type.inspect}, meta=#{@meta.inspect}, status=#{@status.inspect}"
    end
  end

  value = signal

  @status = :running
  fail value if value.is_a?(Celluloid::Interruption)
  value
end

#terminate ⇒ Object

Terminate this task



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/celluloid/task.rb', line 105

def terminate
  fail "Cannot terminate an exclusive task" if exclusive?

  if running?
    if $CELLULOID_DEBUG
      Internals::Logger.with_backtrace(backtrace) do |logger|
        type = @dangerous_suspend ? :warn : :debug
        logger.send(type, "Terminating task: type=#{@type.inspect}, meta=#{@meta.inspect}, status=#{@status.inspect}")
      end
    end
    exception = TaskTerminated.new("task was terminated")
    exception.set_backtrace(caller)
    resume exception
  else
    fail DeadTaskError, "task is already dead"
  end
end