Skip to content

Commit

Permalink
fix(cluster): autopipeline when keyPrefix is used
Browse files Browse the repository at this point in the history
Previously the building of pipelines ignored the key prefix.
It was possible that two keys, foo and bar, might be set into
the same pipeline. However, after being prefixed by a configured
"keyPrefix" value, they may no longer belong to the same
pipeline.

This led to the error:
"All keys in the pipeline should belong to the same slots
allocation group"

Amended version of https://github.com/luin/ioredis/pull/1335/files - see comments on that PR

This may fix redis#1264 and redis#1248.
  • Loading branch information
TysonAndre committed Jul 28, 2021
1 parent beefcc1 commit 2e10f8a
Show file tree
Hide file tree
Showing 6 changed files with 143 additions and 11 deletions.
38 changes: 36 additions & 2 deletions lib/autoPipelining.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import * as PromiseContainer from "./promiseContainer";
import { flatten, isArguments } from "./utils/lodash";
import * as calculateSlot from "cluster-key-slot";
import asCallback from "standard-as-callback";

Expand Down Expand Up @@ -74,11 +75,35 @@ export function shouldUseAutoPipelining(
);
}

/**
* @private
*/
export function getFirstValueInFlattenedArray(
args: (string | string[])[]
): string | undefined {
for (let i = 0; i < args.length; i++) {
const arg = args[i];
if (typeof arg === "string") {
return arg;
} else if (Array.isArray(arg) || isArguments(arg)) {
if (arg.length === 0) {
continue;
}
return arg[0];
}
const flattened = flatten([arg]);
if (flattened.length > 0) {
return flattened[0];
}
}
return undefined;
}

export function executeWithAutoPipelining(
client,
functionName: string,
commandName: string,
args: string[],
args: (string | string[])[],
callback
) {
const CustomPromise = PromiseContainer.get();
Expand All @@ -104,7 +129,16 @@ export function executeWithAutoPipelining(
}

// If we have slot information, we can improve routing by grouping slots served by the same subset of nodes
const slotKey = client.isCluster ? client.slots[calculateSlot(args[0])].join(",") : 'main';
// Note that the first value in args may be a (possibly empty) array.
// ioredis will only flatten one level of the array, in the Command constructor.
const prefix = client.options.keyPrefix || "";
const slotKey = client.isCluster
? client.slots[
calculateSlot(
`${client.options.keyPrefix}${getFirstValueInFlattenedArray(args)}`
)
].join(",")
: "main";

if (!client._autoPipelines.has(slotKey)) {
const pipeline = client.pipeline();
Expand Down
3 changes: 2 additions & 1 deletion lib/utils/lodash.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import defaults = require("lodash.defaults");
import flatten = require("lodash.flatten");
import isArguments = require("lodash.isarguments");

export function noop() {}

export { defaults, flatten };
export { defaults, flatten, isArguments };
14 changes: 14 additions & 0 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions package.json
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
"denque": "^1.1.0",
"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",
Expand All @@ -53,6 +54,7 @@
"@types/debug": "^4.1.5",
"@types/lodash.defaults": "^4.2.6",
"@types/lodash.flatten": "^4.4.6",
"@types/lodash.isarguments": "^3.1.6",
"@types/mocha": "^7.0.2",
"@types/node": "^13.11.0",
"@types/redis-errors": "1.2.0",
Expand Down
58 changes: 50 additions & 8 deletions test/functional/cluster/autopipelining.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ describe("autoPipelining for cluster", function () {
if (argv[0] === "get" && argv[1] === "foo6") {
return "bar6";
}

if (argv[0] === "get" && argv[1] === "baz:foo10") {
return "bar10";
}
});

new MockServer(30002, function (argv) {
Expand Down Expand Up @@ -68,6 +72,10 @@ describe("autoPipelining for cluster", function () {
return "bar5";
}

if (argv[0] === "get" && argv[1] === "baz:foo1") {
return "bar1";
}

if (argv[0] === "evalsha") {
return argv.slice(argv.length - 4);
}
Expand Down Expand Up @@ -177,6 +185,40 @@ describe("autoPipelining for cluster", function () {
cluster.disconnect();
});

it("should support building pipelines when a prefix is used", async () => {
const cluster = new Cluster(hosts, {
enableAutoPipelining: true,
keyPrefix: "baz:",
});
await new Promise((resolve) => cluster.once("connect", resolve));

await cluster.set("foo1", "bar1");
await cluster.set("foo10", "bar10");

expect(
await Promise.all([cluster.get("foo1"), cluster.get("foo10")])
).to.eql(["bar1", "bar10"]);

cluster.disconnect();
});

it("should support building pipelines when a prefix is used with arrays to flatten", async () => {
const cluster = new Cluster(hosts, {
enableAutoPipelining: true,
keyPrefix: "baz:",
});
await new Promise((resolve) => cluster.once("connect", resolve));

await cluster.set(["foo1"], "bar1");
await cluster.set(["foo10"], "bar10");

expect(
await Promise.all([cluster.get(["foo1"]), cluster.get(["foo10"])])
).to.eql(["bar1", "bar10"]);

cluster.disconnect();
});

it("should support commands queued after a pipeline is already queued for execution", (done) => {
const cluster = new Cluster(hosts, { enableAutoPipelining: true });

Expand Down Expand Up @@ -407,9 +449,9 @@ describe("autoPipelining for cluster", function () {
const promise4 = cluster.set("foo6", "bar");

// Override slots to induce a failure
const key1Slot = calculateKeySlot('foo1');
const key2Slot = calculateKeySlot('foo2');
const key5Slot = calculateKeySlot('foo5');
const key1Slot = calculateKeySlot("foo1");
const key2Slot = calculateKeySlot("foo2");
const key5Slot = calculateKeySlot("foo5");

changeSlot(cluster, key1Slot, key2Slot);
changeSlot(cluster, key2Slot, key5Slot);
Expand Down Expand Up @@ -498,9 +540,9 @@ describe("autoPipelining for cluster", function () {
expect(cluster.autoPipelineQueueSize).to.eql(4);

// Override slots to induce a failure
const key1Slot = calculateKeySlot('foo1');
const key2Slot = calculateKeySlot('foo2');
const key5Slot = calculateKeySlot('foo5');
const key1Slot = calculateKeySlot("foo1");
const key2Slot = calculateKeySlot("foo2");
const key5Slot = calculateKeySlot("foo5");
changeSlot(cluster, key1Slot, key2Slot);
changeSlot(cluster, key2Slot, key5Slot);
});
Expand Down Expand Up @@ -547,8 +589,8 @@ describe("autoPipelining for cluster", function () {

expect(cluster.autoPipelineQueueSize).to.eql(3);

const key1Slot = calculateKeySlot('foo1');
const key2Slot = calculateKeySlot('foo2');
const key1Slot = calculateKeySlot("foo1");
const key2Slot = calculateKeySlot("foo2");
changeSlot(cluster, key1Slot, key2Slot);
});
});
Expand Down
39 changes: 39 additions & 0 deletions test/unit/autoPipelining.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
import * as sinon from "sinon";
import { expect } from "chai";
import { getFirstValueInFlattenedArray } from "../../lib/autoPipelining";
import { flatten } from "../../lib/utils/lodash";

describe("autoPipelining", function () {
const expectGetFirstValueIs = (values, expected) => {
expect(getFirstValueInFlattenedArray(values)).to.eql(expected);
// getFirstValueInFlattenedArray should behave the same way as flatten(args)[0]
// but be much more efficient.
expect(flatten(values)[0]).to.eql(expected);
};

it("should be able to efficiently get array args", function () {
expectGetFirstValueIs([], undefined);
expectGetFirstValueIs([null, "key"], null);
expectGetFirstValueIs(["key", "value"], "key");
expectGetFirstValueIs([[], "key"], "key");
expectGetFirstValueIs([["key"]], "key");
// @ts-ignore
expectGetFirstValueIs([[["key"]]], ["key"]);
// @ts-ignore
expectGetFirstValueIs([0, 1, 2, 3, 4], 0);
// @ts-ignore
expectGetFirstValueIs([[true]], true);
// @ts-ignore
expectGetFirstValueIs([Buffer.from("test")], Buffer.from("test"));
// @ts-ignore
expectGetFirstValueIs([{}], {});
// lodash.isArguments is true for this legacy js way to get argument lists
const createArguments = function () {
return arguments;
};
// @ts-ignore
expectGetFirstValueIs([createArguments(), createArguments("key")], "key");
// @ts-ignore
expectGetFirstValueIs([createArguments("")], "");
});
});

0 comments on commit 2e10f8a

Please sign in to comment.