Class: Disbatch::Task

Inherits:
Object
  • Object
show all
Defined in:
lib/disbatch/task.rb

Overview

Represents a Disbatch task

Defined Under Namespace

Modules: Status

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queue, parameters, id) ⇒ Task

Create a task object

Parameters:



24
25
26
27
28
 
# File 'lib/disbatch/task.rb', line 24

def initialize(queue, parameters, id)
  @queue  = queue
  @parameters = parameters
  @id     = id
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



6
7
8
 
# File 'lib/disbatch/task.rb', line 6

def id
  @id
end

#parameters ⇒ Object (readonly) Also known as: params

Returns the value of attribute parameters.



6
7
8
 
# File 'lib/disbatch/task.rb', line 6

def parameters
  @parameters
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



6
7
8
 
# File 'lib/disbatch/task.rb', line 6

def queue
  @queue
end

Class Method Details

.create(queue, parameters) ⇒ Object

Create a new task

Parameters:



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/disbatch/task.rb', line 50

def self.create(queue, parameters)
  id = Mongo.try do
    Disbatch.db[:tasks].insert({
    :queue => queue.id,
    :parameters => parameters,
    :ctime => Time.now,
    :mtime => Time.now,
    :node => -1,
    :status => Status::CREATED,
    :stdout => '',
    :stderr => '',
    :log => []
    })
  end

  unless id.nil?
    new(queue, parameters, id)
  end
end

.take(queue) ⇒ Object

Take a task

Parameters:



33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/disbatch/task.rb', line 33

def self.take(queue)
  doc = Mongo.try do
    Disbatch.db[:tasks].find_and_modify({
      :query => { :queue => queue.id, :status => Status::CREATED },
      :update => { :$set => { :node => Disbatch.node.id, :status => Status::CLAIMED, } }
    })
  end

  unless doc.nil?
    new(queue, doc['parameters'], doc['_id'])
  end
end

Instance Method Details

#conclude ⇒ Object

Conclude task



140
141
142
 
# File 'lib/disbatch/task.rb', line 140

def conclude
  self.status=Status::CONCLUDED
end

#execute! ⇒ Object

Execute the task



145
146
147
 
# File 'lib/disbatch/task.rb', line 145

def execute!
  Disbatch::Plugin[queue.plugin].execute(self)
end

#fail ⇒ Object

Fail task



130
131
132
 
# File 'lib/disbatch/task.rb', line 130

def fail
  self.status=Status::TERMINATED
end

#log(message) ⇒ Object

Log a message

Parameters:

  • message (String) 


73
74
75
 
# File 'lib/disbatch/task.rb', line 73

def log(message)
  Disbatch.db[:tasks].update({:_id => @id}, {'$push' => {'log'=>message} })
end

#query(attribute) ⇒ Object

Query a task attribute

Parameters:

  • attribute (String) 


80
81
82
83
 
# File 'lib/disbatch/task.rb', line 80

def query(attribute)
  doc = Mongo.try { Disbatch.db[:tasks].find_one({:_id => @id}, { :fields => { attribute => 1 } }) }
  doc.nil? ? nil : doc[attribute]
end

#release ⇒ Object

Release task



135
136
137
 
# File 'lib/disbatch/task.rb', line 135

def release
  self.status=Status::CREATED
end

#status ⇒ Object

Query task status



94
95
96
 
# File 'lib/disbatch/task.rb', line 94

def status
  query('status')
end

#status=(status) ⇒ Object

Set task status

Parameters:

  • status (Integer) 


101
102
103
 
# File 'lib/disbatch/task.rb', line 101

def status=(status)
  update(:status,status)
end

#stderr ⇒ Object

Query stderr



118
119
120
 
# File 'lib/disbatch/task.rb', line 118

def stderr 
  query('stderr')
end

#stderr=(message) ⇒ Object

Set stdout

Parameters:

  • message (String) 


125
126
127
 
# File 'lib/disbatch/task.rb', line 125

def stderr=(message)
  update(:stdout,message)
end

#stdout ⇒ Object

Query stdout



106
107
108
 
# File 'lib/disbatch/task.rb', line 106

def stdout 
  query('stdout')
end

#stdout=(message) ⇒ Object

Set stdout

Parameters:

  • message (String) 


113
114
115
 
# File 'lib/disbatch/task.rb', line 113

def stdout=(message)
  update(:stdout,message)
end

#update(attribute, value) ⇒ Object

Update a task attribute

Parameters:

  • attribute (Symbol)
  • value (Object) 


89
90
91
 
# File 'lib/disbatch/task.rb', line 89

def update(attribute, value)
  Disbatch.db[:tasks].update({:_id => @id}, {:$set => { attribute => value } })
end