select: yield on storage tuple lookup

Yield on `select`/`pairs` tuple lookup on storage. Fiber yields each
`yield_every` records (same as in `count`). `yield_every` should be a
positive integer, default is 1000.

crud code contains several `while true` loops. Three of them are related
to the retry strategy. Since retries wraps around net box calls which
yield, it shouldn't be dangerous. The other two are in storage select
procedure: on after tuple scroll and records filtering. If there are
a lot of records which not satisfy any conditions, storage will stuck
with 100% CPU load, like in #312. This patch covers these two cases.

There are pairs loops without any yields on router: fields extraction.
Since the count of records is not expected to be that big there
(we already work with the final set which would be sent to user),
this patch doesn't change the behavior on router.

Closes #312

Author: DifferentialOrange

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Changed
+* Yield on select/pairs storage tuple lookup (#312).
+
 ## [1.1.0] - 13-03-23
 
 ### Added

README.md

@@ -959,6 +959,8 @@ where:
   * `vshard_router` (`?string|table`) - Cartridge vshard group name or
     vshard router instance. Set this parameter if your space is not
     a part of the default vshard cluster
+  * `yield_every` (`?number`) - number of tuples processed on storage to yield after,
+    `yield_every` should be > 0, default value is 1000
 
 
 Returns metadata and array of rows, error.

crud/select.lua

@@ -42,6 +42,7 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         sharding_key_hash = '?number',
         sharding_func_hash = '?number',
         skip_sharding_hash_check = '?boolean',
+        yield_every = '?number',
     })
 
     local space = box.space[space_name]
@@ -77,6 +78,7 @@ local function select_on_storage(space_name, index_id, conditions, opts)
         after_tuple = opts.after_tuple,
         tarantool_iter = opts.tarantool_iter,
         limit = opts.limit,
+        yield_every = opts.yield_every,
     })
     if err ~= nil then
         return nil, SelectError:new("Failed to execute select: %s", err)

crud/select/compat/select.lua

@@ -27,6 +27,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         field_names = '?table',
+        yield_every = '?number',
         call_opts = 'table',
     })
 
@@ -36,6 +37,12 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         return nil, SelectError:new("batch_size should be > 0")
     end
 
+    if opts.yield_every ~= nil and opts.yield_every < 1 then
+        return nil, SelectError:new("yield_every should be > 0")
+    end
+
+    local yield_every = opts.yield_every or const.DEFAULT_YIELD_EVERY
+
     -- check conditions
     local conditions, err = compare_conditions.parse(user_conditions)
     if err ~= nil then
@@ -163,6 +170,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         sharding_func_hash = sharding_func_hash,
         sharding_key_hash = sharding_key_data.hash,
         skip_sharding_hash_check = skip_sharding_hash_check,
+        yield_every = yield_every,
     }
 
     local merger = Merger.new(vshard_router, replicasets_to_select, space, plan.index_id,
@@ -202,6 +210,8 @@ function select_module.pairs(space_name, user_conditions, opts)
         timeout = '?number',
 
         vshard_router = '?string|table',
+
+        yield_every = '?number',
     })
 
     opts = opts or {}
@@ -222,6 +232,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         bucket_id = opts.bucket_id,
         force_map_call = opts.force_map_call,
         field_names = opts.fields,
+        yield_every = opts.yield_every,
         call_opts = {
             mode = opts.mode,
             prefer_replica = opts.prefer_replica,
@@ -273,6 +284,8 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         timeout = '?number',
 
         vshard_router = '?string|table',
+
+        yield_every = '?number',
     })
 
     if opts.first ~= nil and opts.first < 0 then
@@ -288,6 +301,7 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         bucket_id = opts.bucket_id,
         force_map_call = opts.force_map_call,
         field_names = opts.fields,
+        yield_every = opts.yield_every,
         call_opts = {
             mode = opts.mode,
             prefer_replica = opts.prefer_replica,

crud/select/compat/select_old.lua

@@ -30,6 +30,7 @@ local function select_iteration(space_name, plan, opts)
         call_opts = 'table',
         sharding_hash = 'table',
         vshard_router = 'table',
+        yield_every = 'number',
     })
 
     local call_opts = opts.call_opts
@@ -45,6 +46,7 @@ local function select_iteration(space_name, plan, opts)
         sharding_func_hash = opts.sharding_hash.sharding_func_hash,
         sharding_key_hash = opts.sharding_hash.sharding_key_hash,
         skip_sharding_hash_check = opts.sharding_hash.skip_sharding_hash_check,
+        yield_every = opts.yield_every,
     }
 
     local storage_select_args = {
@@ -90,6 +92,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         bucket_id = '?number|cdata',
         force_map_call = '?boolean',
         field_names = '?table',
+        yield_every = '?number',
         call_opts = 'table',
     })
 
@@ -99,6 +102,12 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         return nil, SelectError:new("batch_size should be > 0")
     end
 
+    if opts.yield_every ~= nil and opts.yield_every < 1 then
+        return nil, SelectError:new("yield_every should be > 0")
+    end
+
+    local yield_every = opts.yield_every or const.DEFAULT_YIELD_EVERY
+
     local batch_size = opts.batch_size or common.DEFAULT_BATCH_SIZE
 
     -- check conditions
@@ -214,6 +223,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         call_opts = opts.call_opts,
         sharding_hash = sharding_hash,
         vshard_router = vshard_router,
+        yield_every = yield_every,
     })
 
     return iter
@@ -235,6 +245,8 @@ function select_module.pairs(space_name, user_conditions, opts)
         timeout = '?number',
 
         vshard_router = '?string|table',
+
+        yield_every = '?number',
     })
 
     opts = opts or {}
@@ -255,6 +267,7 @@ function select_module.pairs(space_name, user_conditions, opts)
         bucket_id = opts.bucket_id,
         force_map_call = opts.force_map_call,
         field_names = opts.fields,
+        yield_every = opts.yield_every,
         call_opts = {
             mode = opts.mode,
             prefer_replica = opts.prefer_replica,
@@ -326,6 +339,7 @@ local function select_module_call_xc(vshard_router, space_name, user_conditions,
         bucket_id = opts.bucket_id,
         force_map_call = opts.force_map_call,
         field_names = opts.fields,
+        yield_every = opts.yield_every,
         call_opts = {
             mode = opts.mode,
             prefer_replica = opts.prefer_replica,
@@ -387,6 +401,8 @@ function select_module.call(space_name, user_conditions, opts)
         timeout = '?number',
 
         vshard_router = '?string|table',
+
+        yield_every = '?number',
     })
 
     opts = opts or {}

crud/select/executor.lua

@@ -1,4 +1,5 @@
 local errors = require('errors')
+local fiber = require('fiber')
 local fun = require('fun')
 
 local dev_checks = require('crud.common.dev_checks')
@@ -17,17 +18,23 @@ local ExecuteSelectError = errors.new_class('ExecuteSelectError')
 
 local executor = {}
 
-local function scroll_to_after_tuple(gen, space, scan_index, tarantool_iter, after_tuple)
+local function scroll_to_after_tuple(gen, space, scan_index, tarantool_iter, after_tuple, yield_every)
     local primary_index = space.index[0]
 
     local scroll_key_parts = utils.merge_primary_key_parts(scan_index.parts, primary_index.parts)
 
     local cmp_operator = select_comparators.get_cmp_operator(tarantool_iter)
     local scroll_comparator = select_comparators.gen_tuples_comparator(cmp_operator, scroll_key_parts)
 
+    local looked_up_tuples = 0
     while true do
         local tuple
         gen.state, tuple = gen(gen.param, gen.state)
+        looked_up_tuples = looked_up_tuples + 1
+
+        if yield_every ~= nil and looked_up_tuples % yield_every == 0 then
+            fiber.yield()
+        end
 
         if tuple == nil then
             return nil
@@ -73,6 +80,7 @@ function executor.execute(space, index, filter_func, opts)
         after_tuple = '?table',
         tarantool_iter = 'number',
         limit = '?number',
+        yield_every = '?number',
     })
 
     opts = opts or {}
@@ -137,7 +145,7 @@ function executor.execute(space, index, filter_func, opts)
 
     if opts.after_tuple ~= nil then
         local err
-        tuple, err = scroll_to_after_tuple(gen, space, index, opts.tarantool_iter, opts.after_tuple)
+        tuple, err = scroll_to_after_tuple(gen, space, index, opts.tarantool_iter, opts.after_tuple, opts.yield_every)
         if err ~= nil then
             return nil, ExecuteSelectError:new("Failed to scroll to the after_tuple: %s", err)
         end
@@ -151,6 +159,7 @@ function executor.execute(space, index, filter_func, opts)
         gen.state, tuple = gen(gen.param, gen.state)
     end
 
+    local looked_up_tuples = 0
     while true do
         if tuple == nil then
             break
@@ -170,6 +179,11 @@ function executor.execute(space, index, filter_func, opts)
         end
 
         gen.state, tuple = gen(gen.param, gen.state)
+        looked_up_tuples = looked_up_tuples + 1
+
+        if opts.yield_every ~= nil and looked_up_tuples % opts.yield_every == 0 then
+            fiber.yield()
+        end
     end
 
     return resp

crud/select/iterator.lua

@@ -28,6 +28,8 @@ function Iterator.new(opts)
 
         call_opts = 'table',
         sharding_hash = 'table',
+
+        yield_every = '?number',
     })
 
     local iter = {
@@ -57,6 +59,8 @@ function Iterator.new(opts)
         wait_for_update = false,
 
         sharding_hash = opts.sharding_hash,
+
+        yield_every = opts.yield_every,
     }
 
     setmetatable(iter, Iterator)
@@ -108,6 +112,7 @@ local function update_replicasets_tuples(iter, after_tuple, replicaset_uuid)
         call_opts = iter.call_opts,
         sharding_hash = iter.sharding_hash,
         vshard_router = iter.vshard_router,
+        yield_every = iter.yield_every,
     })
     if err ~= nil then
         if sharding.result_needs_sharding_reload(err) then