Provides per tube consumer channels and put+wait draft

Author: ochaton

example-tube/init.lua

@@ -1,6 +1,4 @@
-box.cfg{
-	-- wal_mode = 'none',
-}
+box.cfg{}
 
 require 'strict'.on()
 
@@ -17,7 +15,6 @@ local format = {
 -- utube stands for unordered tubes
 box.schema.create_space('utube', {
 	format = format,
-	temporary = true,
 	if_not_exists = true,
 })
 
@@ -38,6 +35,8 @@ box.space.utube:create_index('run', {
 	if_not_exists = true,
 })
 
+box.space.utube:truncate()
+
 do
 	local fio = require 'fio'
 	local root = fio.abspath(debug.getinfo(1).source:gsub("@","")):gsub("[^/]+$", "")
@@ -57,21 +56,24 @@ require 'xqueue'.upgrade(box.space.utube, {
 		id      = 'time64',
 		delayed = true,
 		buried  = true,
-		zombie  = 60,
 
 		retval  = 'tuple',
 	},
 })
 
+local fiber = require 'fiber'
+local test = require 'tap'.test("tubes")
+
 for i = 1, 10 do
-	box.space.utube:put({
+	local task = box.space.utube:put({
 		tube    = "tube-" .. i,
 		payload = {
 			id   = i,
 			ctime = require 'clock'.realtime64(),
 		},
 		nice = i,
 	})
+	test:ok(task, "task was returned by :put")
 end
 
 for i = 10, 1, -1 do
@@ -80,22 +82,82 @@ for i = 10, 1, -1 do
 		tube = tube,
 	})
 
-	assert(task, "task not taken from tube: " .. tube)
-	assert(task.tube == tube, "task must be taken from specific tube")
+	test:ok(task, "task was taken from tube: " .. tube)
+	test:is(task.tube, tube, "task is from the same tube consumer requested")
 
 	local notask = box.space.utube:take({
 		tube = tube,
 		timeout = 0.5,
 	})
 
-	assert(not notask, "task was not taken because tube is empty")
+	test:isnil(notask, "no task because tube " .. tube .. " is empty")
 
 	if i % 2 == 0 then
-		box.space.utube:ack(task)
+		local ret = box.space.utube:ack(task)
+		test:ok(ret, ":ack returned processed task")
+		test:is(ret.id, task.id, ":ack returned same task with same task.id")
 	else
-		box.space.utube:release(task)
+		local ret = box.space.utube:release(task)
+		test:ok(ret, ":release returned processed task")
+		test:is(ret.id, task.id, ":release returned same task with same task.id")
+		test:is(ret.status, 'R', ":release retutned task in status 'R'")
+	end
+end
+
+do
+	-- common worker:
+	for i = 1, box.space.utube:len() do
+		local task = box.space.utube:take(0)
+		test:ok(task, ":take returned task from queue")
+		local tube = 'tube-' .. (2*i-1)
+		test:is(task.tube, tube, "task must be from tube: " .. tube)
+
+		local ret = box.space.utube:release(task, { delay = 1 })
+		test:is(ret.id, task.id, ":release successfully returned task back to queue")
+		test:is(ret.status, "W", "task was returned to status 'W' because was released with delay")
+	end
+end
+
+-- clears queue:
+box.space.utube:truncate()
+
+-- Let's check how take+put works:
+do
+	local first = fiber.channel(1)
+	local second = fiber.channel(1)
+
+	local tubename = 'tube'
+
+	fiber.create(function()
+		-- creating this fiber we initiated new session
+		-- so only this fiber takes ownership for this task
+		second:put(box.space.utube:take(1), 0)
+	end)
+
+	fiber.create(function()
+		first:put(box.space.utube:take({
+			tube = tubename,
+			timeout = 1,
+		}), 0)
+	end)
+
+	fiber.sleep(0.1)
+
+	for i = 1, 2 do
+		box.space.utube:put({
+			tube = tubename,
+			payload = { id = i },
+			nice = 512,
+		})
 	end
+
+	local task1 = first:get(0.1)
+	test:ok(task1, "tube consumer must receive task")
+	test:is(task1.payload.id, 1, "tube consumer must receive task first")
+
+	local task2 = second:get(0.1)
+	test:ok(task2, "common consumer must receive task")
+	test:is(task2.payload.id, 2, "common consumer must receive task second")
 end
 
-require 'console'.start()
 os.exit()

xqueue.lua

@@ -245,6 +245,7 @@ function M.upgrade(space,opts,depth)
 		self.bysid = space.xq.bysid
 		self._lock = space.xq._lock
 		self.take_wait = space.xq.take_wait
+		self.take_chans = space.xq.take_chans or {}
 		self._on_repl  = space.xq._on_repl
 		self._on_dis = space.xq._on_dis
 	else
@@ -253,6 +254,7 @@ function M.upgrade(space,opts,depth)
 		-- byfid = {};
 		self._lock = {}
 		self.take_wait = fiber.channel(0)
+		self.take_chans = setmetatable({}, { __mode = 'v' })
 	end
 	setmetatable(self.bysid, {
 		__serialize='map',
@@ -808,6 +810,15 @@ function M.upgrade(space,opts,depth)
 
 	function self:wakeup(t)
 		if t[self.fieldmap.status] ~= 'R' then return end
+		if self.fieldmap.tube then
+			-- we may have consumers in the tubes:
+			local tube_chan = self.take_chans[t[self.fieldmap.tube]]
+			if tube_chan and tube_chan:has_readers() and tube_chan:put(true, 0) then
+				-- we have successfully notified consumer
+				return
+			end
+			-- otherwise fallback to default channel:
+		end
 		if self.take_wait:has_readers() then
 			self.take_wait:put(true,0)
 		end
@@ -953,6 +964,9 @@ end
 			* if set, task will become `W` instead of `R` for `delay` seconds
 		+ `ttl` - number of seconds
 			* if set, task will be discarded after ttl seconds unless was taken
+		+ `wait` - number of seconds
+			* if set, callee fiber will be blocked up to `wait` seconds until task won't
+			be processed or timeout reached.
 
 ```lua
 box.space.myqueue:put{ name="xxx"; data="yyy"; }
@@ -1002,6 +1016,10 @@ function methods:put(t, opts)
 		end
 		t[ xq.fieldmap.status ] = 'W'
 		t[ xq.fieldmap.runat ]  = xq.timeoffset(opts.delay)
+
+		if opts.wait then
+			error("Are you crazy? Call of :put({...}, { wait = <>, delay = <> }) looks weird", 2)
+		end
 	elseif opts.ttl then
 		if not xq.features.ttl then
 			error("Feature ttl is not enabled",2)
@@ -1021,11 +1039,23 @@ function methods:put(t, opts)
 	local tuple = xq.tuple(t)
 	local key = tuple[ xq.key.no ]
 
-	xq._lock[ key ] = true
-	t = self:insert( tuple )
-	xq._lock[ key ] = nil
+	local chan
+	if opts.wait then
+		chan = fiber.channel(1)
+		xq.put_wait[key] = chan
+	end
 
+	xq:atomic(key, function()
+		t = self:insert(tuple)
+	end)
 	xq:wakeup(t)
+
+	if chan then
+		local res = chan:get(opts.wait)
+		if res then
+			return xq.retwrap(res), true
+		end
+	end
 	return xq.retwrap(t)
 end
 
@@ -1036,6 +1066,7 @@ end
 	- `timeout` - number of seconds to wait for new task
 		+ choose reasonable time
 		+ beware of **readahead** size (see tarantool docs)
+	- `tube` - name of the tube worker wants to take task from (feature tube must be enabled)
 	- returns task tuple or table (see retval) or nothing on timeout
 	- *TODO*: ttr must be there
 ]]
@@ -1065,12 +1096,18 @@ function methods:take(timeout, opts)
 	local index
 	local start_with
 
+	local tube_chan
 	if opts.tube then
 		if not xq.features.tube then
 			error("Feature tube is not enabled", 2)
 		end
+
+		assert(type(opts.tube) == 'string', "opts.tube must be a string")
+
 		index = xq.tube_index
 		start_with = {opts.tube, 'R'}
+		tube_chan = xq.take_chans[opts.tube] or fiber.channel()
+		xq.take_chans[opts.tube] = tube_chan
 	else
 		index = xq.index
 		start_with = {'R'}
@@ -1082,9 +1119,7 @@ function methods:take(timeout, opts)
 	while not found do
 		for _,t in index:pairs(start_with, { iterator = box.index.EQ }) do
 			key = t[ xq.key.no ]
-			if xq._lock[ key ] then
-				-- continue
-			else
+			if not xq._lock[ key ] then
 				-- found key
 				xq._lock[ key ] = true
 				found = t
@@ -1093,13 +1128,22 @@ function methods:take(timeout, opts)
 		end
 		if not found then
 			local left = (now + timeout) - fiber.time()
-			if left <= 0 then return end
-			xq.take_wait:get(left)
-			if box.session.storage.destroyed then return end
-		else
-			break
+			if left <= 0 then goto finish end
+
+			(tube_chan or xq.take_wait):get(left)
+			if box.session.storage.destroyed then goto finish end
 		end
 	end
+	::finish::
+
+	-- If we were last reader from the tube
+	-- we remove channel. Writers will get false
+	-- on :put if they exists.
+	if tube_chan and not tube_chan:has_readers() then
+		tube_chan:close()
+		xq.take_chans[opts.tube] = nil
+	end
+	if not found then return end
 
 	local r,e = pcall(function()
 		local sid = box.session.id()