Skip to content

perf(zero-cache): improve rm fanout flow control #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Karavil wants to merge 1 commit into from
Closed

perf(zero-cache): improve rm fanout flow control #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions packages/zero-cache/src/config/zero-config.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,11 +406,11 @@ test('zero-cache --help', () => {
rather, it protects the system when the upstream throughput exceeds the downstream
throughput.

--change-streamer-flow-control-consensus-padding-seconds number default: 1
--change-streamer-flow-control-consensus-padding-seconds number default: 0.1
ZERO_CHANGE_STREAMER_FLOW_CONTROL_CONSENSUS_PADDING_SECONDS env
During periodic flow control checks (every 64kb), the amount of time to wait after the
majority of subscribers have acked, after which replication will continue even if
some subscribers have yet to ack. (Note that this is not a timeout for the entire send,
During flow control flushes, the amount of time to wait after the majority of
subscribers have acked, after which replication will continue even if some subscribers
have yet to ack. (Note that this is not a timeout for the entire send,
but a timeout that starts after the majority of receivers have acked.)

This allows a bounded amount of time for backlogged subscribers to catch up on each flush
Expand Down
8 changes: 4 additions & 4 deletions packages/zero-cache/src/config/zero-config.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -651,11 +651,11 @@ export const zeroOptions = {
},

flowControlConsensusPaddingSeconds: {
type: v.number().default(1),
type: v.number().default(0.1),
desc: [
`During periodic flow control checks (every 64kb), the amount of time to wait after the`,
`majority of subscribers have acked, after which replication will continue even if`,
`some subscribers have yet to ack. (Note that this is not a timeout for the {italic entire} send,`,
`During flow control flushes, the amount of time to wait after the majority of`,
`subscribers have acked, after which replication will continue even if some subscribers`,
`have yet to ack. (Note that this is not a timeout for the {italic entire} send,`,
`but a timeout that starts {italic after} the majority of receivers have acked.)`,
``,
`This allows a bounded amount of time for backlogged subscribers to catch up on each flush`,
Expand Down
152 changes: 151 additions & 1 deletion packages/zero-cache/src/services/change-streamer/broadcast.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,70 @@
import {describe, expect, test} from 'vitest';
import {afterEach, describe, expect, test, vi} from 'vitest';
import {BigIntJSON} from '../../../../shared/src/bigint-json.ts';
import {createSilentLogContext} from '../../../../shared/src/logging-test-utils.ts';
import {sleep} from '../../../../shared/src/sleep.ts';
import type {Subscription} from '../../types/subscription.ts';
import {ReplicationMessages} from '../replicator/test-utils.ts';
import {Broadcast} from './broadcast.ts';
import type {Subscriber} from './subscriber.ts';
import {createSubscriber} from './test-utils.ts';

const json = BigIntJSON.stringify;

describe('change-streamer/broadcast', () => {
const messages = new ReplicationMessages({issues: 'id'});
const lc = createSilentLogContext();
const change: [string, 'begin', string] = [
'11',
'begin',
json(['begin', messages.begin(), {commitWatermark: '13'}]),
];

afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});

async function consumeOne(downstream: Subscription<string>) {
const pipeline = downstream.pipeline;
if (pipeline === undefined) {
throw new Error('Expected pipelined subscription');
}
const next = await pipeline[Symbol.asyncIterator]().next();
if (next.done) {
throw new Error('Expected subscription message');
}
next.value.consumed();
}

async function flushCompletions() {
await Promise.resolve();
await Promise.resolve();
}

async function makeSubscribers(count: number) {
const subscribers: Subscriber[] = [];
const downstreams: Subscription<string>[] = [];
for (let i = 0; i < count; i++) {
const [subscriber, , downstream] = createSubscriber('00', true);
subscribers.push(subscriber);
downstreams.push(downstream);
}
await Promise.all(downstreams.map(consumeOne));
await flushCompletions();
return {subscribers, downstreams};
}

async function consumeChange(downstream: Subscription<string>) {
await consumeOne(downstream);
await flushCompletions();
}

async function closeAll(subscribers: readonly Subscriber[]) {
for (const sub of subscribers) {
sub.close();
}
await flushCompletions();
}

test('without tracking', () => {
const [sub1, stream1] = createSubscriber('00', true);
Expand Down Expand Up @@ -76,6 +130,102 @@ describe('change-streamer/broadcast', () => {
}
});

test('flow control releases after majority plus padding', async () => {
vi.useFakeTimers();
const {subscribers, downstreams} = await makeSubscribers(4);
const broadcast = new Broadcast(subscribers, change, {
lc,
flowControlConsensusPaddingMs: 50,
});

await consumeChange(downstreams[0]);
await consumeChange(downstreams[1]);
await vi.advanceTimersByTimeAsync(100);
expect(broadcast.isDone).toBe(false);

await consumeChange(downstreams[2]);
await vi.advanceTimersByTimeAsync(49);
expect(broadcast.isDone).toBe(false);

await vi.advanceTimersByTimeAsync(1);
await broadcast.done;
expect(broadcast.isDone).toBe(true);

await closeAll(subscribers);
});

test('flow control timer resets after post-majority progress', async () => {
vi.useFakeTimers();
const {subscribers, downstreams} = await makeSubscribers(5);
const broadcast = new Broadcast(subscribers, change, {
lc,
flowControlConsensusPaddingMs: 50,
});

await consumeChange(downstreams[0]);
await consumeChange(downstreams[1]);
await consumeChange(downstreams[2]);
await vi.advanceTimersByTimeAsync(30);
await consumeChange(downstreams[3]);

await vi.advanceTimersByTimeAsync(49);
expect(broadcast.isDone).toBe(false);

await vi.advanceTimersByTimeAsync(1);
await broadcast.done;
expect(broadcast.isDone).toBe(true);

await closeAll(subscribers);
});

test('flow control timer is cleared when all subscribers complete', async () => {
vi.useFakeTimers();
const clearTimeoutSpy = vi.spyOn(globalThis, 'clearTimeout');
const {subscribers, downstreams} = await makeSubscribers(4);
const broadcast = new Broadcast(subscribers, change, {
lc,
flowControlConsensusPaddingMs: 50,
});

await consumeChange(downstreams[0]);
await consumeChange(downstreams[1]);
await consumeChange(downstreams[2]);
expect(broadcast.isDone).toBe(false);

await consumeChange(downstreams[3]);
await broadcast.done;
expect(broadcast.isDone).toBe(true);
expect(clearTimeoutSpy).toHaveBeenCalled();

await vi.advanceTimersByTimeAsync(50);
expect(broadcast.isDone).toBe(true);
await closeAll(subscribers);
});

test('negative flow control padding disables early release', async () => {
vi.useFakeTimers();
const {subscribers, downstreams} = await makeSubscribers(4);
const broadcast = new Broadcast(subscribers, change, {
lc,
flowControlConsensusPaddingMs: -1,
});

await consumeChange(downstreams[0]);
await consumeChange(downstreams[1]);
await consumeChange(downstreams[2]);
expect(broadcast.checkProgress(lc, -1, performance.now() + 1000)).toBe(
false,
);
await vi.advanceTimersByTimeAsync(1000);
expect(broadcast.isDone).toBe(false);

await consumeChange(downstreams[3]);
await broadcast.done;
expect(broadcast.isDone).toBe(true);

await closeAll(subscribers);
});

test('checkProgress', async () => {
const [sub1] = createSubscriber('00', true);
const [sub2] = createSubscriber('00', true);
Expand Down
Loading