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`. 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 (it is ~3 times slower). Thus, this workaround is
disabled by default. To enable it, user should set the `policy = "utube_ready_space"`
as an option while creating the tube. As example:
```lua
local test_queue = queue.create_tube('test_queue', 'utube',
        {temporary = true, policy = "utube_ready_space"})
```

Part of #228

CHANGELOG.md

@@ -7,10 +7,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Added
+- `v2` boolean 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 `v2 = true` option while creating the tube.
 
 ## [1.3.3] - 2023-09-13
 

README.md

@@ -172,6 +172,11 @@ on disk
 already exists
   * `on_task_change` - function name - a callback to be executed on every
 operation
+  * `policy` - string - one of
+    * `"default"` - default implementation of `utube`
+    * `"utube_ready_space"` - allows processing `take` requests faster, but
+      by the cost of `put` operations speed. Right now this option is enabled
+      only for `memtx` engine
 
 The following options can be specified when putting a task in a `utube`
 queue:

queue/abstract.lua

@@ -297,7 +297,11 @@ function tube.drop(self)
 
     local space_name = tube[3]
 
-    box.space[space_name]:drop()
+    if self.raw.drop ~= nil then
+        self.raw:drop()
+    else
+        box.space[space_name]:drop()
+    end
     box.space._queue:delete{tube_name}
     -- drop queue
     queue.tube[tube_name] = nil

queue/abstract/driver/policy.lua

@@ -0,0 +1,5 @@
+-- tube policies
+return {
+    POLICY_DEFAULT = "default",
+    POLICY_UTUBE_READY_SPACE = "utube_ready_space",
+}