[TNTP-2109] Use internal.trigger instead of custom hooks

Author: ita-sammann

crud/common/call.lua

@@ -43,25 +43,22 @@ end
 
 local call_on_storage = rebalance.safe_mode and call_on_storage_safe or call_on_storage_fast
 
-local function safe_mode_enable()
-    call_on_storage = call_on_storage_safe
-
-    for fb_id, fb in pairs(fiber.info()) do
-        local fibers_killed = 0
-        if fb.name:startswith(CRUD_CALL_FUNC_NAME .. '/fast/') and not safe_mode_allowed_fast_methods[fb.name] then
-            fiber.kill(fb_id)
-            fibers_killed = fibers_killed + 1
+rebalance.on_safe_mode_toggle(function(is_enabled)
+    if is_enabled then
+        call_on_storage = call_on_storage_safe
+
+        for fb_id, fb in pairs(fiber.info()) do
+            local fibers_killed = 0
+            if fb.name:startswith(CRUD_CALL_FUNC_NAME .. '/fast/') and not safe_mode_allowed_fast_methods[fb.name] then
+                fiber.kill(fb_id)
+                fibers_killed = fibers_killed + 1
+            end
+            log.debug('Killed %d fibers with fast-mode crud requests.', fibers_killed)
         end
-        log.debug('Killed %d fibers with fast-mode crud requests.', fibers_killed)
+    else
+        call_on_storage = call_on_storage_fast
     end
-end
-
-local function safe_mode_disable()
-    call_on_storage = call_on_storage_fast
-end
-
-rebalance.register_safe_mode_enable_hook(safe_mode_enable)
-rebalance.register_safe_mode_disable_hook(safe_mode_disable)
+end)
 
 call.storage_api = {[CALL_FUNC_NAME] = call_on_storage}
 

crud/common/rebalance.lua

@@ -1,5 +1,6 @@
 local fiber = require('fiber')
 local log = require('log')
+local trigger = require('internal.trigger')
 local vshard_consts = require('vshard.consts')
 local utils = require('crud.common.utils')
 local has_metrics_module, metrics = pcall(require, 'metrics')
@@ -8,8 +9,8 @@ local SAFE_MODE_SPACE = '_crud_rebalance_safe_mode_local'
 
 local rebalance = {
     safe_mode = false,
-    safe_mode_enable_hooks = {},
-    safe_mode_disable_hooks = {},
+    -- Trigger is run with one argument: true if safe mode is enabled and false if disabled.
+    on_safe_mode_toggle = trigger.new('_crud.safe_mode_toggle'),
     _router_cache_last_clear_ts = 0 -- On module load we don't know when (and if) route cache was cleared.
 }
 
@@ -39,9 +40,8 @@ local function _safe_mode_enable()
     end
     rebalance.safe_mode = true
 
-    for hook, _ in pairs(rebalance.safe_mode_enable_hooks) do
-        pcall(hook)
-    end
+    -- This function is running inside on_commit trigger, need pcall to protect from errors in external code.
+    pcall(rebalance.on_safe_mode_toggle.run, rebalance.on_safe_mode_toggle, true)
 
     log.info('Rebalance safe mode enabled')
 end
@@ -52,9 +52,8 @@ local function _safe_mode_disable()
     box.space._bucket:on_replace(safe_mode_bucket_trigger)
     rebalance.safe_mode = false
 
-    for hook, _ in pairs(rebalance.safe_mode_disable_hooks) do
-        pcall(hook)
-    end
+    -- This function is running inside on_commit trigger, need pcall to protect from errors in external code.
+    pcall(rebalance.on_safe_mode_toggle.run, rebalance.on_safe_mode_toggle, false)
 
     log.info('Rebalance safe mode disabled')
 end
@@ -130,22 +129,6 @@ function rebalance.init()
     end
 end
 
-function rebalance.register_safe_mode_enable_hook(func)
-    rebalance.safe_mode_enable_hooks[func] = true
-end
-
-function rebalance.remove_safe_mode_enable_hook(func)
-    rebalance.safe_mode_enable_hooks[func] = nil
-end
-
-function rebalance.register_safe_mode_disable_hook(func)
-    rebalance.safe_mode_disable_hooks[func] = true
-end
-
-function rebalance.remove_safe_mode_disable_hook(func)
-    rebalance.safe_mode_disable_hooks[func] = nil
-end
-
 function rebalance.safe_mode_status()
     return rebalance.safe_mode
 end

crud/common/sharding/bucket_ref_unref.lua

@@ -172,8 +172,14 @@ function M.set_safe_mode_status(safe_mode_status)
     end
 
     if not hooks_registered then
-        rebalance.register_safe_mode_enable_hook(set_safe_mode)
-        rebalance.register_safe_mode_disable_hook(set_fast_mode)
+        rebalance.on_safe_mode_toggle(function(is_enabled)
+            if is_enabled then
+                set_safe_mode()
+            else
+                set_fast_mode()
+            end
+        end)
+
         hooks_registered = true
     end
 end