Class: OsCtld::ContinuousExecutor

Inherits:
Object
  • Object
show all
Includes:
OsCtl::Lib::Utils::Exception, OsCtl::Lib::Utils::Log
Defined in:
lib/osctld/continuous_executor.rb

Overview

Continuous execution queue

This class handles parallel execution of queued commands with priorities. Commands can be added, the queue can be cleared and the pool of workers can be resized at runtime.

Defined Under Namespace

Classes: Command

Instance Method Summary collapse

Constructor Details

#initialize(size) ⇒ ContinuousExecutor

Returns a new instance of ContinuousExecutor

Parameters:

  • size (Integer) —

    initial number of workers



60
61
62
63
64
65
66
67
68
69
 
# File 'lib/osctld/continuous_executor.rb', line 60

def initialize(size)
  @mutex = Mutex.new
  @size = size
  @workers = []
  @front_queue = OsCtl::Lib::Queue.new
  @exec_queue = []
  @counter = Concurrent::AtomicFixnum.new(0)

  start
end

Instance Method Details

#<<(cmd) ⇒ Object

Enqueue command for execution

Parameters:



79
80
81
 
# File 'lib/osctld/continuous_executor.rb', line 79

def <<(cmd)
  enqueue(cmd)
end

#clear ⇒ Object

Clear the execution queue



94
95
96
 
# File 'lib/osctld/continuous_executor.rb', line 94

def clear
  @front_queue << :clear
end

#enqueue(cmd) ⇒ Object

Enqueue command for execution

Parameters:



73
74
75
 
# File 'lib/osctld/continuous_executor.rb', line 73

def enqueue(cmd)
  @front_queue << cmd
end

#exec(cmd) ⇒ Object (protected) 



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/osctld/continuous_executor.rb', line 182

def exec(cmd)
  t = Thread.new do
    begin
      ret = cmd.send(:exec)

    rescue Exception => e
      log(:warn, 'cont', "Exception raised during command execution: #{e.message}")
      puts denixstorify(e.backtrace).join("\n")

    ensure
      @front_queue << Thread.current
      cmd.send(:done, ret)
    end
  end

  @workers << t
end

#execute(cmd, timeout: nil) ⇒ any

Enqueue command for execution and wait until it is executed

Parameters:

  • cmd (Command)
  • timeout (Integer, nil) —

    how long to wait for the command to execute

Returns:

  • (any) —

    return value from the executed command



87
88
89
90
91
 
# File 'lib/osctld/continuous_executor.rb', line 87

def execute(cmd, timeout: nil)
  q = cmd.send(:return_queue)
  enqueue(cmd)
  q.shift(timeout: timeout)
end

#queue ⇒ Array<Command>

Return contents of the current queue

Returns:



111
112
113
 
# File 'lib/osctld/continuous_executor.rb', line 111

def queue
  sync { @exec_queue.clone }
end

#resize(new_size) ⇒ Object

Parameters:

  • new_size (Integer) —

    new number of workers



105
106
107
 
# File 'lib/osctld/continuous_executor.rb', line 105

def resize(new_size)
  @front_queue << new_size
end

#start ⇒ Object (protected) 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/osctld/continuous_executor.rb', line 116

def start
  @main = Thread.new do
    loop do
      v = @front_queue.pop

      if v.is_a?(Command)
        v.send(:'order=', @counter.increment)

        sync do
          if @workers.size < @size
            exec(v)
          else
            @exec_queue << v
            @exec_queue.sort!
          end
        end

      elsif v.is_a?(Array)
        v.each { |cmd| cmd.send(:'order=', @counter.increment) }

        sync do
          @exec_queue.concat(v)
          @exec_queue.sort!

          while @workers.size < @size && @exec_queue.any?
            exec(@exec_queue.shift)
          end
        end

      elsif v.is_a?(Thread)
        sync do
          @workers.delete(v)

          while @workers.size < @size && @exec_queue.any?
            exec(@exec_queue.shift)
          end
        end

      elsif v == :clear
        @front_queue.clear
        sync { @exec_queue.clear }

      elsif v == :stop
        sync do
          @exec_queue.clear
          @workers.each(&:join)
        end

        break

      elsif v.is_a?(Integer)
        sync do
          @size = v

          while @workers.size < @size && @exec_queue.any?
            exec(@exec_queue.shift)
          end
        end

      else
        # unknown
      end
    end
  end
end

#stop ⇒ Object

Stop execution and wait for all workers to finish



99
100
101
102
 
# File 'lib/osctld/continuous_executor.rb', line 99

def stop
  @front_queue << :stop
  @main.join
end

#sync ⇒ Object (protected) 



200
201
202
 
# File 'lib/osctld/continuous_executor.rb', line 200

def sync
  @mutex.synchronize { yield }
end