feat: add concurrent option to utube

Also adds support for taking from a specific sub-queue by passing `{ utube = '...' }` to the `:take` method.

Author: aleclarson

README.md

@@ -127,23 +127,24 @@ options are equal.
 
 The main idea of this queue backend is the same as in a `fifo` queue:
 the tasks are executed in FIFO order.
-However, tasks may be grouped into sub-queues.
+
+However, tasks may be grouped into sub-queues. Sub-queues split the task
+stream according to the sub-queue name. Each sub-queue is executed in strict
+FIFO order. By default, only one task per sub-queue may be taken at any time.
 
 The following options can be specified when creating a `utube` queue:
+ * `concurrent` - number - the number of concurrent tasks per sub-queue
+(defaults to 1)
  * `temporary` - boolean - if true, the contents of the queue do not persist
 on disk
  * `if_not_exists` - boolean - if true, no error will be returned if the tube
 already exists
  * `on_task_change` - function name - a callback to be executed on every
 operation
 
-The following options can be specified when putting a task in a `utube`
-queue:
+The following options are supported for the `:put` and `:take` methods of
+a `utube` queue:
  * `utube` - the name of the sub-queue.
-Sub-queues split the task stream according to the sub-queue name: it is
-not possible to take two tasks
-out of a sub-queue concurrently, each sub-queue is executed in strict
-FIFO order, one task at a time.
 
 `utube` queue does not support:
  * task priorities (`pri`)

queue/abstract/driver/utube.lua

@@ -42,10 +42,11 @@ function tube.create_space(space_name, opts)
 end
 
 -- start tube on space
-function tube.new(space, on_task_change)
+function tube.new(space, on_task_change, opts)
  on_task_change = on_task_change or (function() end)
  local self = setmetatable({
  space = space,
+ concurrent = opts.concurrent or 1,
  on_task_change = on_task_change,
  }, { __index = method })
  return self
@@ -65,21 +66,46 @@ function method.put(self, data, opts)
  return task
 end
 
+-- check concurrency of a sub-queue
+function method.is_throttled(self, utube)
+ if self.concurrent == 1 then
+ local taken = self.space.index.utube:min{state.TAKEN, utube}
+ return taken ~= nil and taken[3] == utube
+ elseif self.concurrent ~= 1/0 then
+ local num_taken = self.space.index.utube:count{state.TAKEN, utube}
+ return num_taken == self.concurrent
+ end
+ return false
+end
+
 -- take task
-function method.take(self)
- for s, task in self.space.index.status:pairs(state.READY,
- { iterator = 'GE' }) do
- if task[2] ~= state.READY then
- break
+function method.take(self, opts)
+ local task
+ if opts and opts.utube then
+ if not self:is_throttled(opts.utube) then
+ local t = self.space.index.utube:min{state.READY, opts.utube}
+ if t and t[2] == state.READY and t[3] == opts.utube then
+ task = t
+ end
  end
-
- local taken = self.space.index.utube:min{state.TAKEN, task[3]}
- if taken == nil or taken[2] ~= state.TAKEN then
- task = self.space:update(task[1], { { '=', 2, state.TAKEN } })
- self.on_task_change(task, 'take')
- return task
+ else
+ local utubes = {}
+ for s, t in self.space.index.status:pairs(state.READY) do
+ local utube = t[3]
+ if not utubes[utube] then
+ if not self:is_throttled(utube) then
+ task = t
+ break
+ end
+ utubes[utube] = true
+ end
  end
  end
+ if task then
+ task = self.space:update(task[1], { { '=', 2, state.TAKEN } })
+ self.on_task_change(task, 'take')
+ return task
+ end
 end
 
 -- touch task