utube: fix slow take on busy utubes

If some of the utube for tasks at the top of the queue were busy
most of the time, `take` would slow down for every other task.
This problem is fixed by creating a new space `space_ready_buffer`. It
contains first task with `READY` status from each utube.

This solution shows great results for the stated problem, with the cost
of slowing the `put` method. Thus, this workaround is disabled by default.
To enable it, user should set the
`storage_mode = queue.driver.utube.STORAGE_MODE_READY_BUFFER` as an option
while creating the tube. As example:
```lua
local test_queue = queue.create_tube('test_queue', 'utube',
        {temporary = true, storage_mode = queue.driver.utube.STORAGE_MODE_READY_BUFFER})
```

Part of #228

CHANGELOG.md

@@ -7,10 +7,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+- `storage_mode` option for creating a `utube` tube (#228). It enables the
+  workaround for slow takes while working with busy tubes.
+
 ### Fixed
 
 - Stuck in `INIT` state if an instance failed to enter the `running` mode
   in time (#226). This fix works only for Tarantool versions >= 2.10.0.
+- Slow takes on busy `utube` tubes (#228). The workaround could be enabled by
+  passing the `storage_mode = "ready_buffer"` option while creating
+  the tube.
 
 ## [1.3.3] - 2023-09-13
 

README.md

@@ -165,13 +165,49 @@ 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.
 
+It is advised not to use `utube` methods inside transactions with
+`read-confirmed` isolation level. It can lead to errors when trying to make
+parallel tube methods calls with mvcc enabled.
+
 The following options can be specified when creating a `utube` queue:
   * `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
+  * `storage_mode` - string - one of
+    * `queue.driver.utube.STORAGE_MODE_DEFAULT` ("default") - default
+      implementation of `utube`
+    * `queue.driver.utube.STORAGE_MODE_READY_BUFFER`
+      ("ready_buffer") - allows processing `take` requests faster, but
+      by the cost of `put` operations speed. Right now this option is supported
+      only for `memtx` engine.
+      WARNING: this is an experimental storage mode.
+
+    Here is a benchmark comparison of these two modes:
+    * Benchmark for simple `put` and `take` methods. 30k utubes are created
+      with a single task each. Task creation time is calculated. After that
+      30k consumers are calling `take` + `ack`, each in the separate fiber.
+      Time to ack all tasks is calculated. The results are as follows:
+
+      |         | put (30k) | take+ack |
+      |---------|-----------|----------|
+      | default | 180ms     | 1.6s     |
+      | ready   | 270ms     | 1.7s     |
+    * Benchmark for the busy utubes. 10 tubes are created.
+      Each contains 1000 tasks. After that, 10 consumers are created (each works
+      on his tube only, one tube — one consumer). Each consumer will 
+      `take`, then `yield` and then `ack` every task from their utube 
+      (1000 tasks each).
+      After that, we can also run this benchmark with 10k tasks on each utube,
+      100k tasks and 150k tasks. But all that with 10 utubes and 10 consumers.
+      The results are as follows:
+
+      |         | 1k    | 10k  | 50k  | 150k  |
+      |---------|-------|------|------|-------|
+      | default | 53s   | 1.5h | 100h | 1000h |
+      | ready   | 450ms | 4.7s | 26s  | 72s   |
 
 The following options can be specified when putting a task in a `utube`
 queue:

queue/abstract.lua

@@ -295,9 +295,12 @@ function tube.drop(self)
         error("There are taken tasks in the tube")
     end
 
-    local space_name = tube[3]
-
-    box.space[space_name]:drop()
+    if self.raw.drop ~= nil then
+        self.raw:drop()
+    else
+        local space_name = tube[3]
+        box.space[space_name]:drop()
+    end
     box.space._queue:delete{tube_name}
     -- drop queue
     queue.tube[tube_name] = nil