tube: fix slow take on busy utubes

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` and `utubettl` 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` and `utubettl` 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:
@@ -208,16 +244,53 @@ in strict FIFO order.
 
 This queue type is effectively a combination of `fifottl` and `utube`.
 
+It is advised not to use `utubettl` 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 `utubettl` 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.utubettl.STORAGE_MODE_DEFAULT` ("default") - default
+      implementation of `utubettl`
+    * `queue.driver.utubettl.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 | 200ms     | 1.7s     |
+      | ready   | 320ms     | 1.8s     |
+    * 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 140k tasks. But all that with 10 utubes and 10 consumers.
+      The results are as follows:
+
+      |         | 1k    | 10k  | 50k  | 140k  |
+      |---------|-------|------|------|-------|
+      | default | 80s   | 1.6h | 100h | 1000h |
+      | ready   | 520ms | 5.4s | 28s  | 83s   |
 
 The following options can be specified when putting a task in a
 `utubettl` queue:
+  * `pri` - task priority (`0` is the highest priority and is the default)
   * `utube` - the name of the sub-queue
   * `ttl` - numeric - time to live for a task put into the queue, in
 seconds. if `ttl` is not specified, it is set to infinity

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