feat: Pipeline-based script loading

Author: marcbachmann

lib/pipeline.ts

@@ -1,5 +1,4 @@
 import * as calculateSlot from "cluster-key-slot";
-import * as pMap from "p-map";
 import { exists, hasFlag } from "redis-commands";
 import asCallback from "standard-as-callback";
 import { deprecate } from "util";
@@ -326,71 +325,9 @@ Pipeline.prototype.exec = function (
     }
   }
 
-  // Check whether scripts exists
-  const scripts = [];
-  for (let i = 0; i < this._queue.length; ++i) {
-    const item = this._queue[i];
-
-    if (item.name !== "evalsha") {
-      continue;
-    }
-
-    const script = this._shaToScript[item.args[0]];
-
-    if (
-      !script ||
-      this.redis._addedScriptHashes[script.sha] ||
-      scripts.includes(script)
-    ) {
-      continue;
-    }
-
-    scripts.push(script);
-  }
-
   const _this = this;
-  if (!scripts.length) {
-    return execPipeline();
-  }
+  execPipeline();
 
-  // In cluster mode, always load scripts before running the pipeline
-  if (this.isCluster) {
-    pMap(scripts, (script) => _this.redis.script("load", script.lua), {
-      concurrency: 10,
-    })
-      .then(function () {
-        for (let i = 0; i < scripts.length; i++) {
-          _this.redis._addedScriptHashes[scripts[i].sha] = true;
-        }
-      })
-      .then(execPipeline, this.reject);
-    return this.promise;
-  }
-
-  this.redis
-    .script(
-      "exists",
-      scripts.map(({ sha }) => sha)
-    )
-    .then(function (results) {
-      const pending = [];
-      for (let i = 0; i < results.length; ++i) {
-        if (!results[i]) {
-          pending.push(scripts[i]);
-        }
-      }
-      return Promise.all(
-        pending.map(function (script) {
-          return _this.redis.script("load", script.lua);
-        })
-      );
-    })
-    .then(function () {
-      for (let i = 0; i < scripts.length; i++) {
-        _this.redis._addedScriptHashes[scripts[i].sha] = true;
-      }
-    })
-    .then(execPipeline, this.reject);
   return this.promise;
 
   function execPipeline() {

lib/script.ts

@@ -39,23 +39,97 @@ export default class Script {
       options.readOnly = true;
     }
 
-    const evalsha = new Command("evalsha", [this.sha].concat(args), options);
-    evalsha.isCustomCommand = true;
-    evalsha.promise = evalsha.promise.catch((err: Error) => {
-      if (err.toString().indexOf("NOSCRIPT") === -1) {
-        throw err;
-      }
-      const command = new Command("eval", [this.lua].concat(args), options);
-      if (container.isPipeline === true) container.redis.sendCommand(command);
-      else container.sendCommand(command);
-      return command.promise;
+    if (container.isPipeline) {
+      return usingPipeline(this, container, args, options, callback);
+    } else if (container.isCluster) {
+      return usingCluster(this, container, args, options, callback);
+    } else {
+      return usingStandalone(this, container, args, options, callback);
+    }
+  }
+}
+
+function createEvalShaWithFallback(script, args, options, client) {
+  const evalsha = createEvalSha(script, args, options);
+  evalsha.promise = evalsha.promise.catch((err: Error) => {
+    if (err.toString().indexOf("NOSCRIPT") === -1) {
+      throw err;
+    }
+
+    // Do an eval as fallback, redis will hash and load it
+    const evalcmd = new Command("eval", [script.lua].concat(args), options);
+    evalcmd.promise = evalcmd.promise.then((r) => {
+      client._addedScriptHashes[script.sha] = true;
+      return r;
     });
+    return client.sendCommand(evalcmd);
+  });
+  return evalsha;
+}
+
+function createEvalSha(script, args, options) {
+  const evalsha = new Command("evalsha", [script.sha].concat(args), options);
+  evalsha.isCustomCommand = true;
+  return evalsha;
+}
+
+// Pipeline mode (cluster and regular)
+function usingPipeline(script, pipeline, args, options, callback) {
+  // The script was loaded explicitly in this pipeline,
+  // so we can directly execute evalsha without loading the script again
+  if (pipeline._addedScriptHashes && pipeline._addedScriptHashes[script.sha]) {
+    const evalsha = createEvalSha(script, args, options);
+    asCallback(evalsha.promise, callback);
+    return pipeline.sendCommand(evalsha);
+  }
 
+  // The script is loaded in redis already, so we try an evalsha
+  // and fallback to loading the script if it fails
+  if (pipeline.redis._addedScriptHashes[script.sha]) {
+    const evalsha = createEvalShaWithFallback(
+      script,
+      args,
+      options,
+      pipeline.redis
+    );
     asCallback(evalsha.promise, callback);
+    return pipeline.sendCommand(evalsha);
+  }
+
+  // If the script is not present on redis or in the pipeline,
+  // we use eval to load the script into the pipeline
+  if (!pipeline._addedScriptHashes) pipeline._addedScriptHashes = {};
+  pipeline._addedScriptHashes[script.sha] = true;
+
+  const evalcmd = new Command("eval", [script.lua].concat(args), options);
+  evalcmd.promise = evalcmd.promise.then((r) => {
+    pipeline.redis._addedScriptHashes[script.sha] = true;
+    return r;
+  });
+  asCallback(evalcmd.promise, callback);
+  return pipeline.sendCommand(evalcmd);
+}
+
+// Standalone mode (cluster)
+function usingCluster(script, cluster, args, options, callback) {
+  const evalsha = createEvalShaWithFallback(script, args, options, cluster);
+  asCallback(evalsha.promise, callback);
+  return cluster.sendCommand(evalsha);
+}
 
-    // The result here is one of
-    // - a Promise when executed on the redis instance
-    // - a pipeline instance in pipeline mode
-    return container.sendCommand(evalsha);
+// Standalone mode (regular)
+function usingStandalone(script, redis, args, options, callback) {
+  if (redis._addedScriptHashes[script.sha]) {
+    const evalsha = createEvalShaWithFallback(script, args, options, redis);
+    return asCallback(redis.sendCommand(evalsha), callback);
   }
+
+  const command = new Command("eval", [script.lua].concat(args), options);
+  command.promise = command.promise.then((r) => {
+    redis._addedScriptHashes[script.sha] = true;
+    return r;
+  });
+
+  asCallback(command.promise, callback);
+  return redis.sendCommand(command);
 }

package-lock.json

@@ -39,7 +39,6 @@
     "lodash.defaults": "^4.2.0",
     "lodash.flatten": "^4.4.0",
     "lodash.isarguments": "^3.1.0",
-    "p-map": "^2.1.0",
     "redis-commands": "1.7.0",
     "redis-errors": "^1.2.0",
     "redis-parser": "^3.0.0",

package.json

@@ -200,7 +200,6 @@ describe("auth", function () {
       let errorEmited = false;
       const redis = new Redis({ port: 17379, password: "pass" });
       redis.on("error", function () {
-        console.log("boop");
         errorEmited = true;
       });
       const stub = sinon.stub(console, "warn").callsFake((warn) => {

test/functional/auth.ts

@@ -76,7 +76,7 @@ describe("autoPipelining for cluster", function () {
         return "bar1";
       }
 
-      if (argv[0] === "evalsha") {
+      if (argv[0] === "evalsha" || argv[0] === "eval") {
         return argv.slice(argv.length - 4);
       }
     });

test/functional/cluster/autopipelining.ts

@@ -1,6 +1,7 @@
 import Redis from "../../lib/Redis";
 import { expect } from "chai";
 import * as sinon from "sinon";
+import { getCommandsFromMonitor } from "../helpers/util";
 
 describe("pipeline", function () {
   it("should return correct result", function (done) {
@@ -275,7 +276,7 @@ describe("pipeline", function () {
       });
     });
 
-    it("should check and load uniq scripts only", function (done) {
+    it("should check and load uniq scripts only", async function () {
       const redis = new Redis();
       redis.defineCommand("test", {
         numberOfKeys: 2,
@@ -286,49 +287,43 @@ describe("pipeline", function () {
         lua: "return {KEYS[1],ARGV[1]}",
       });
 
-      redis.once("ready", function () {
-        const expectedCommands = [
-          ["script", "exists"],
-          ["script", "load", "return {KEYS[1],ARGV[1]}"],
-          ["script", "load", "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}"],
-          ["evalsha"],
-          ["evalsha"],
-          ["evalsha"],
-          ["evalsha"],
-          ["evalsha"],
-          ["evalsha"],
-        ];
-        const expectedResults = [
-          [null, ["a", "1"]],
-          [null, ["b", "2"]],
-          [null, ["k1", "k2", "v1", "v2"]],
-          [null, ["k3", "k4", "v3", "v4"]],
-          [null, ["c", "3"]],
-          [null, ["k5", "k6", "v5", "v6"]],
-        ];
-        redis.monitor(function (err, monitor) {
-          monitor.on("monitor", function (_, command) {
-            const expectedCommand = expectedCommands.shift();
-            expectedCommand.forEach((arg, i) => expect(arg).to.eql(command[i]));
-            if (!expectedCommands.length) {
-              monitor.disconnect();
-              redis.disconnect();
-              done();
-            }
+      const expectedCommands = [
+        ["eval"],
+        ["evalsha"],
+        ["eval"],
+        ["evalsha"],
+        ["evalsha"],
+        ["evalsha"],
+      ];
+
+      const expectedResults = [
+        [null, ["a", "1"]],
+        [null, ["b", "2"]],
+        [null, ["k1", "k2", "v1", "v2"]],
+        [null, ["k3", "k4", "v3", "v4"]],
+        [null, ["c", "3"]],
+        [null, ["k5", "k6", "v5", "v6"]],
+      ];
+
+      const commands = await getCommandsFromMonitor(redis, 6, () => {
+        return redis
+          .pipeline()
+          .echo("a", "1")
+          .echo("b", "2")
+          .test("k1", "k2", "v1", "v2")
+          .test("k3", "k4", "v3", "v4")
+          .echo("c", "3")
+          .test("k5", "k6", "v5", "v6")
+          .exec()
+          .then((results) => {
+            expect(results).to.eql(expectedResults);
           });
-          const pipe = redis.pipeline();
-          pipe
-            .echo("a", "1")
-            .echo("b", "2")
-            .test("k1", "k2", "v1", "v2")
-            .test("k3", "k4", "v3", "v4")
-            .echo("c", "3")
-            .test("k5", "k6", "v5", "v6")
-            .exec(function (err, results) {
-              expect(err).to.eql(null);
-              expect(results).to.eql(expectedResults);
-            });
-        });
+      });
+
+      redis.disconnect();
+
+      expectedCommands.forEach((expectedCommand, j) => {
+        expectedCommand.forEach((arg, i) => expect(arg).to.eql(commands[j][i]));
       });
     });
 
@@ -362,11 +357,10 @@ describe("pipeline", function () {
         lua: `return "Foo"`,
       });
 
-      const [[err, res]] = await redis
-        .pipeline([["execafterreconnect"]])
-        .exec();
-      expect(err).to.equal(null);
-      expect(res).to.equal("Foo");
+      const preloadscript = await redis.pipeline().execafterreconnect().exec();
+
+      expect(preloadscript[0][0]).to.equal(null);
+      expect(preloadscript[0][1]).to.equal("Foo");
 
       const client = await redis.client("list").then((clients) => {
         const myInfo = clients
@@ -379,22 +373,24 @@ describe("pipeline", function () {
 
       await redis2.script("flush");
       await redis2.client("kill", "addr", client);
-
-      const res2 = await redis
-        .pipeline([
-          ["set", "foo", "bar"],
-          ["execafterreconnect"],
-          ["get", "foo"],
-        ])
-        .exec();
-
-      expect(res2).to.deep.equal([
-        [null, "OK"],
-        [null, "Foo"],
-        [null, "bar"],
-      ]);
+      await redis.get("waitforready");
+
+      const commands = await getCommandsFromMonitor(redis2, 3, () => {
+        return redis
+          .pipeline([
+            ["set", "foo", "bar"],
+            ["execafterreconnect"],
+            ["get", "foo"],
+          ])
+          .exec();
+      });
       redis.disconnect();
       redis2.disconnect();
+
+      const expected = ["set", "eval", "get"];
+      commands.forEach((c, i) => {
+        expect(c[0]).to.equal(expected[i]);
+      });
     });
   });