perf: speed up xqueue.upgrade() for large spaces (#22)

* perf: speed up xqueue.upgrade() for large spaces
* test: fix flaky test

Author: ochaton

.env

@@ -0,0 +1,4 @@
+BENCHER_PROJECT=xqueue
+BENCHER_ADAPTER=json
+BENCHER_TESTBED=localhost
+LUABENCH_USE_BMF=true

.github/workflows/benchmark.yml

@@ -0,0 +1,27 @@
+name: Benchmarking code
+
+on:
+  - push
+
+jobs:
+  run-unit-benchmarks:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        version: ["2.10.6", "2.11.0", "2.11.2"]
+    steps:
+    - uses: actions/checkout@v4
+    - uses: tarantool/setup-tarantool@v3
+      with:
+        tarantool-version: '${{matrix.version}}'
+    - name: install argparse
+      run: tarantoolctl rocks install argparse
+    - name: install luabench
+      run: tarantoolctl rocks --server=https://moonlibs.org install luabench 0.3.0
+    - name: run benchmarks
+      env:
+        LUABENCH_USE_BMF: false
+        LUABENCH_TIMEOUT: 60
+        LUABENCH_DURATION: '10s'
+      run: |
+        .rocks/bin/luabench

.luabench

@@ -0,0 +1,2 @@
+jit = 'on'
+duration = '10s'

benchmarks/001_put_take_bench.lua

@@ -40,7 +40,7 @@ xqueue.upgrade(box.space.queue, {
 
 local queue = box.space.queue --[[@as xqueue.space]]
 lb.before_all(function() queue:truncate() end)
-lb.after_all(function() queue:truncate() box.snapshot() end)
+lb.after_all(function() box.space.queue:truncate() box.snapshot() end)
 
 local M = {}
 

benchmarks/002_initial_stat_bench.lua

@@ -0,0 +1,62 @@
+local lb = require 'luabench'
+local fiber = require 'fiber'
+local fun   = require 'fun'
+
+local M = {}
+
+local STATUS = { 'R', 'T', 'W', 'B', 'Z', 'D' }
+
+lb.before_all(function()
+	box.cfg{ memtx_memory = 2^30 }
+
+	box.schema.space.create('q1', {
+		if_not_exists = true,
+		format = {
+			{ name = 'id',     type = 'unsigned' },
+			{ name = 'status', type = 'string' },
+		},
+	})
+
+	box.space.q1:create_index('primary', { parts = {'id'}, if_not_exists = true })
+	box.space.q1:create_index('xq', { parts = {'status', 'id'}, if_not_exists = true })
+
+	if fiber.extend_slice then
+		fiber.extend_slice({ err = 3600, warn = 3600 })
+	end
+
+	box.begin()
+		local tab = {}
+		for no = 1, 4e6 do
+			tab[1], tab[2] = no, STATUS[math.random(#STATUS)]
+			box.space.q1:replace(tab)
+		end
+	box.commit()
+end)
+
+lb.after_all(function()
+	box.space.q1:drop()
+	box.snapshot()
+end)
+
+function M.bench_iterate_all(b)
+	local limit = b.N
+	local scanned = 0
+	local stats = {}
+	for _, t in box.space.q1:pairs({}, { iterator = "ALL" }) do
+		stats[t.status] = (stats[t.status] or 0ULL) + 1
+		scanned = scanned + 1
+		if limit == scanned then break end
+	end
+	b.N = scanned
+end
+
+function M.bench_count(b)
+	local total = 0
+	for _, s in pairs(STATUS) do
+		total = total + box.space.q1.index.xq:count(s)
+		if b.N < total then break end
+	end
+	b.N = total
+end
+
+return M

test/basic_test.lua

@@ -178,10 +178,10 @@ function g.test_delayed_queue()
 	t.assert_equals(queue:get({taken.id}).status, 'W', 'queue:release(..., {delay=<>}) must put task in W')
 	t.assert_le(queue:get({task_put_delay_500ms.id}).runat, queue:get({taken.id}).runat, "first task must wakeup earlier")
 
-	local taken_delayed = queue:take({ timeout = 0.13 })
+	local taken_delayed = queue:take({ timeout = 3 })
 	t.assert(taken_delayed, 'delayed task must be taken after timeout')
 
-	local taken_put = queue:take({ timeout = 0.1 })
+	local taken_put = queue:take({ timeout = 3 })
 	t.assert(taken_put, 'released task must be taken after timeout')
 
 	t.assert_equals(taken_delayed.id, task_put_delay_500ms.id, "firstly delayed task must be taken")

xqueue.lua

@@ -215,6 +215,15 @@ local function _table2tuple ( qformat )
 	return dostring(fun)
 end
 
+local pretty_st = {
+	R = "Ready",
+	T = "Taken",
+	W = "Waiting",
+	B = "Buried",
+	Z = "Zombie",
+	D = "Done",
+}
+
 ---@class xqueue.space
 local methods = {}
 
@@ -583,19 +592,17 @@ function M.upgrade(space,opts,depth)
 			tube = stat_tube;
 		}
 		if self.fields.tube then
-			for _, t in space:pairs(nil, { iterator = box.index.ALL }) do
-				local s = t[self.fields.status]
-				self._stat.counts[s] = (self._stat.counts[s] or 0LL) + 1
-
-				local tube = t[self.fields.tube]
-				if stat_tube[tube] then
-					stat_tube[tube].counts[s] = (stat_tube[tube].counts[s] or 0LL) + 1
+			for status in pairs(pretty_st) do
+				self._stat.counts[status] = 0LL+self.index:count(status)
+			end
+			for tube in pairs(stat_tube) do
+				for status in pairs(pretty_st) do
+					stat_tube[tube].counts[status] = 0LL+self.tube_index:count({ tube, status })
 				end
 			end
 		else
-			for _, t in space:pairs(nil, { iterator = box.index.ALL }) do
-				local s = t[self.fields.status]
-				self._stat.counts[s] = (self._stat.counts[s] or 0LL) + 1
+			for status in pairs(pretty_st) do
+				self._stat.counts[status] = 0LL+self.index:count(status)
 			end
 		end
 	else
@@ -1748,15 +1755,6 @@ function methods:truncate()
 	return ret
 end
 
-local pretty_st = {
-	R = "Ready",
-	T = "Taken",
-	W = "Waiting",
-	B = "Buried",
-	Z = "Zombie",
-	D = "Done",
-}
-
 local shortmap = { __serialize = 'map' }
 
 ---@param pretty? boolean