Optimizations and fixes for clearing clients

SGrondin · Feb 25, 2019 · 71debfb · 71debfb
1 parent 582db61
commit 71debfb
Show file tree

Hide file tree

Showing 18 changed files with 311 additions and 119 deletions.
diff --git a/es5.js b/es5.js
diff --git a/lib/Bottleneck.js b/lib/Bottleneck.js
@@ -671,6 +671,7 @@ Bottleneck = function () {
     Promise: Promise,
     timeout: null,
     heartbeatInterval: 5000,
+    clientTimeout: 10000,
     clientOptions: {},
     clusterNodes: null,
     clearDatastore: false,

diff --git a/lib/RedisDatastore.js b/lib/RedisDatastore.js
@@ -49,7 +49,7 @@ RedisDatastore = class RedisDatastore {
     }).then(() => {
       return this.connection.__addLimiter__(this.instance);
     }).then(() => {
-      return this.runScript("heartbeat", []);
+      return this.runScript("register_client", [this.instance.queued()]);
     }).then(() => {
       var base;
 
@@ -152,13 +152,13 @@ RedisDatastore = class RedisDatastore {
     var _this3 = this;
 
     return _asyncToGenerator(function* () {
-      if (!(name === "init" || name === "heartbeat")) {
+      if (!(name === "init" || name === "register_client")) {
         yield _this3.ready;
       }
 
       return new _this3.Promise((resolve, reject) => {
         var all_args, arr;
-        all_args = [Date.now(), _this3.clientId, _this3.instance.queued()].concat(args);
+        all_args = [Date.now(), _this3.clientId].concat(args);
 
         _this3.instance.Events.trigger("debug", `Calling Redis script: ${name}.lua`, all_args);
 
@@ -171,12 +171,18 @@ RedisDatastore = class RedisDatastore {
         });
         return _this3.connection.__scriptFn__(name)(...arr);
       }).catch(e => {
-        if (e.message === "SETTINGS_KEY_NOT_FOUND" && name !== "heartbeat") {
-          return _this3.runScript("init", _this3.prepareInitSettings(false)).then(() => {
+        if (e.message === "SETTINGS_KEY_NOT_FOUND") {
+          if (name === "heartbeat") {
+            return _this3.Promise.resolve();
+          } else {
+            return _this3.runScript("init", _this3.prepareInitSettings(false)).then(() => {
+              return _this3.runScript(name, args);
+            });
+          }
+        } else if (e.message === "UNKNOWN_CLIENT") {
+          return _this3.runScript("register_client", [_this3.instance.queued()]).then(() => {
             return _this3.runScript(name, args);
           });
-        } else if (name === "heartbeat") {
-          return _this3.Promise.resolve();
         } else {
           return _this3.Promise.reject(e);
         }
@@ -213,7 +219,8 @@ RedisDatastore = class RedisDatastore {
     args = this.prepareObject(Object.assign({}, this.storeOptions, {
       id: this.originalId,
       version: this.instance.version,
-      groupTimeout: this.timeout
+      groupTimeout: this.timeout,
+      clientTimeout: this.clientTimeout
     }));
     args.unshift(clear ? 1 : 0, this.instance.version);
     return args;

diff --git a/lib/Scripts.js b/lib/Scripts.js
@@ -5,6 +5,7 @@ lua = require("./lua.json");
 headers = {
   refs: lua["refs.lua"],
   validate_keys: lua["validate_keys.lua"],
+  validate_client: lua["validate_client.lua"],
   refresh_expiration: lua["refresh_expiration.lua"],
   process_tick: lua["process_tick.lua"],
   conditions_check: lua["conditions_check.lua"],
@@ -67,75 +68,81 @@ templates = {
     refresh_expiration: false,
     code: lua["group_check.lua"]
   },
-  blacklist_client: {
+  register_client: {
     keys: exports.allKeys,
     headers: ["validate_keys"],
     refresh_expiration: false,
+    code: lua["register_client.lua"]
+  },
+  blacklist_client: {
+    keys: exports.allKeys,
+    headers: ["validate_keys", "validate_client"],
+    refresh_expiration: false,
     code: lua["blacklist_client.lua"]
   },
   heartbeat: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: false,
     code: lua["heartbeat.lua"]
   },
   update_settings: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: true,
     code: lua["update_settings.lua"]
   },
   running: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: false,
     code: lua["running.lua"]
   },
   queued: {
     keys: exports.allKeys,
-    headers: ["validate_keys"],
+    headers: ["validate_keys", "validate_client"],
     refresh_expiration: false,
     code: lua["queued.lua"]
   },
   done: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: false,
     code: lua["done.lua"]
   },
   check: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick", "conditions_check"],
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"],
     refresh_expiration: false,
     code: lua["check.lua"]
   },
   submit: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick", "conditions_check"],
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"],
     refresh_expiration: true,
     code: lua["submit.lua"]
   },
   register: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick", "conditions_check"],
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"],
     refresh_expiration: true,
     code: lua["register.lua"]
   },
   free: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: true,
     code: lua["free.lua"]
   },
   current_reservoir: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: false,
     code: lua["current_reservoir.lua"]
   },
   increment_reservoir: {
     keys: exports.allKeys,
-    headers: ["validate_keys", "process_tick"],
+    headers: ["validate_keys", "validate_client", "process_tick"],
     refresh_expiration: true,
     code: lua["increment_reservoir.lua"]
   }

diff --git a/lib/lua.json b/lib/lua.json
diff --git a/light.js b/light.js
@@ -1364,6 +1364,7 @@
 	    Promise: Promise,
 	    timeout: null,
 	    heartbeatInterval: 5000,
+	    clientTimeout: 10000,
 	    clientOptions: {},
 	    clusterNodes: null,
 	    clearDatastore: false,

diff --git a/src/Bottleneck.coffee b/src/Bottleneck.coffee
@@ -41,6 +41,7 @@ class Bottleneck
     Promise: Promise
     timeout: null
     heartbeatInterval: 5000
+    clientTimeout: 10000
     clientOptions: {}
     clusterNodes: null
     clearDatastore: false

diff --git a/src/RedisDatastore.coffee b/src/RedisDatastore.coffee
@@ -21,7 +21,7 @@ class RedisDatastore
     @ready = @connection.ready
     .then (@clients) => @runScript "init", @prepareInitSettings @clearDatastore
     .then => @connection.__addLimiter__ @instance
-    .then => @runScript "heartbeat", []
+    .then => @runScript "register_client", [@instance.queued()]
     .then =>
       (@heartbeat = setInterval =>
         @runScript "heartbeat", []
@@ -72,19 +72,23 @@ class RedisDatastore
       @connection.disconnect flush
 
   runScript: (name, args) ->
-    await @ready unless name == "init" or name == "heartbeat"
+    await @ready unless name == "init" or name == "register_client"
     new @Promise (resolve, reject) =>
-      all_args = [Date.now(), @clientId, @instance.queued()].concat args
+      all_args = [Date.now(), @clientId].concat args
       @instance.Events.trigger "debug", "Calling Redis script: #{name}.lua", all_args
       arr = @connection.__scriptArgs__ name, @originalId, all_args, (err, replies) ->
         if err? then return reject err
         return resolve replies
       @connection.__scriptFn__(name) arr...
     .catch (e) =>
-      if e.message == "SETTINGS_KEY_NOT_FOUND" and name != "heartbeat"
-        @runScript("init", @prepareInitSettings(false))
+      if e.message == "SETTINGS_KEY_NOT_FOUND"
+        if name == "heartbeat" then @Promise.resolve()
+        else
+          @runScript("init", @prepareInitSettings(false))
+          .then => @runScript(name, args)
+      else if e.message == "UNKNOWN_CLIENT"
+        @runScript("register_client", [@instance.queued()])
         .then => @runScript(name, args)
-      else if name == "heartbeat" then @Promise.resolve()
       else @Promise.reject e
 
   prepareArray: (arr) -> (if x? then x.toString() else "") for x in arr
@@ -96,9 +100,10 @@ class RedisDatastore
 
   prepareInitSettings: (clear) ->
     args = @prepareObject Object.assign({}, @storeOptions, {
-      id: @originalId,
-      version: @instance.version,
+      id: @originalId
+      version: @instance.version
       groupTimeout: @timeout
+      @clientTimeout
     })
     args.unshift (if clear then 1 else 0), @instance.version
     args

diff --git a/src/Scripts.coffee b/src/Scripts.coffee
@@ -3,6 +3,7 @@ lua = require "./lua.json"
 headers =
   refs: lua["refs.lua"]
   validate_keys: lua["validate_keys.lua"]
+  validate_client: lua["validate_client.lua"]
   refresh_expiration: lua["refresh_expiration.lua"]
   process_tick: lua["process_tick.lua"]
   conditions_check: lua["conditions_check.lua"]
@@ -68,64 +69,69 @@ templates =
     headers: []
     refresh_expiration: false
     code: lua["group_check.lua"]
-  blacklist_client:
+  register_client:
     keys: exports.allKeys
     headers: ["validate_keys"]
     refresh_expiration: false
+    code: lua["register_client.lua"]
+  blacklist_client:
+    keys: exports.allKeys
+    headers: ["validate_keys", "validate_client"]
+    refresh_expiration: false
     code: lua["blacklist_client.lua"]
   heartbeat:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: false
     code: lua["heartbeat.lua"]
   update_settings:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: true
     code: lua["update_settings.lua"]
   running:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: false
     code: lua["running.lua"]
   queued:
     keys: exports.allKeys
-    headers: ["validate_keys"]
+    headers: ["validate_keys", "validate_client"]
     refresh_expiration: false
     code: lua["queued.lua"]
   done:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: false
     code: lua["done.lua"]
   check:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick", "conditions_check"]
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"]
     refresh_expiration: false
     code: lua["check.lua"]
   submit:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick", "conditions_check"]
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"]
     refresh_expiration: true
     code: lua["submit.lua"]
   register:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick", "conditions_check"]
+    headers: ["validate_keys", "validate_client", "process_tick", "conditions_check"]
     refresh_expiration: true
     code: lua["register.lua"]
   free:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: true
     code: lua["free.lua"]
   current_reservoir:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: false
     code: lua["current_reservoir.lua"]
   increment_reservoir:
     keys: exports.allKeys
-    headers: ["validate_keys", "process_tick"]
+    headers: ["validate_keys", "validate_client", "process_tick"]
     refresh_expiration: true
     code: lua["increment_reservoir.lua"]
 

diff --git a/src/redis/blacklist_client.lua b/src/redis/blacklist_client.lua
@@ -1,5 +1,8 @@
 local blacklist = ARGV[num_static_argv + 1]
 
-redis.call('zadd', client_last_seen_key, 0, blacklist)
+if redis.call('zscore', client_last_seen_key, blacklist) then
+  redis.call('zadd', client_last_seen_key, 0, blacklist)
+end
+
 
 return {}
diff --git a/src/redis/init.lua b/src/redis/init.lua
@@ -24,11 +24,6 @@ if redis.call('exists', settings_key) == 0 then
     'capacityPriorityCounter', 0
   )
 
-  if clear == 1 then
-    -- Force all connected clients to re-register
-    process_tick(now, true)
-  end
-
 else
   -- Apply migrations
   local settings = redis.call('hmget', settings_key,
@@ -54,7 +49,7 @@ else
     end
 
     -- 2.11.1
-    if version_digits[2] < 11 and version_digits[3] < 1 then
+    if version_digits[2] < 11 or (version_digits[2] == 11 and version_digits[3] < 1) then
       if redis.call('hstrlen', settings_key, 'lastReservoirRefresh') == 0 then
         redis.call('hmset', settings_key,
           'lastReservoirRefresh', now,
@@ -78,11 +73,17 @@ else
     end
 
     -- 2.15.2
-    if version_digits[2] < 15 and version_digits[3] < 2 then
+    if version_digits[2] < 15 or (version_digits[2] == 15 and version_digits[3] < 2) then
       redis.call('hsetnx', settings_key, 'capacityPriorityCounter', 0)
       redis.call('hset', settings_key, 'version', '2.15.2')
     end
 
+    -- 2.17.0
+    if version_digits[2] < 17 then
+      redis.call('hsetnx', settings_key, 'clientTimeout', 10000)
+      redis.call('hset', settings_key, 'version', '2.17.0')
+    end
+
   end
 
   process_tick(now, false)