Skip to content

Experiment with concurrency levels #425

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

Open
adespawn wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Experiment with concurrency levels #425

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fd4bd84
Refactor benchmarking
adespawn Mar 30, 2026
221c343
Add benchmark that are parametrized on concurrency
adespawn Mar 30, 2026
a3a795c
Update batch benchmark to use common utils
adespawn Apr 9, 2026
b90d2fd
Add rust version of concurrency parameterized benchmarks
adespawn Apr 10, 2026
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
2 changes: 2 additions & 0 deletions benchmark/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@ Result will be saved to `out.svg`. You can read more in the benchmarker document

## Legacy python script

This script is no longer updated, so it may be broken.

A file that runs all benchmarks: `runner.py`

The script compares benchmark results for our driver, [Cassandra driver](https://github.com/apache/cassandra-nodejs-driver) and [Rust driver](https://github.com/scylladb/scylla-rust-driver). Parameters for the benchmarks can be modified inside it. The result is a `graph.png` file that presents a graph of time on a logarithmic scale. The graphs are uploaded to the provided discord webhook.
Expand Down
75 changes: 33 additions & 42 deletions benchmark/logic/batch.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,50 +1,41 @@
"use strict";
const async = require("async");
// Possible values of argv[2] (driver) are scylladb-driver-alpha and cassandra-driver.
const cassandra = require(process.argv[2]);
const utils = require("./utils");
const { exit } = require("process");
const assert = require("assert");

const client = new cassandra.Client(utils.getClientArgs());
const iterCnt = parseInt(process.argv[3]);
// Expectantly determined max batch size, that doesn't cause database error.
const step = 3971;
module.exports = function (cassandra, client, stepCount, _concurrencyLevel) {
Copy link
Copy Markdown
Contributor

@wprzytula wprzytula Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⛏️ Here you use underscore in _concurrencyLevel, and elsewhere not.

Copy link
Copy Markdown
Contributor Author

@adespawn adespawn Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In non concurrent benchmarks I ignore this parameter. For this reason I use underscore to indicate that

// REMEMBER: update benchmark config.yml when changing the constant value.
const iterCnt = stepCount || 3000000;
// Experimentally determined max batch size that doesn't cause database error.
const batchSize = 3971;

async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
// Limit batch size to step size
for (let z = 0; z * step < iterCnt; z++) {
let queries = [];
for (let i = 0; i < Math.min(iterCnt - (z * step), step); i++) {
queries.push({
query: 'INSERT INTO benchmarks.basic (id, val) VALUES (?, ?)',
params: [cassandra.types.Uuid.random(), 10]
});
async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
for (let z = 0; z * batchSize < iterCnt; z++) {
let queries = [];
for (let i = 0; i < Math.min(iterCnt - (z * batchSize), batchSize); i++) {
queries.push({
query: 'INSERT INTO benchmarks.basic (id, val) VALUES (?, ?)',
params: [cassandra.types.Uuid.random(), 10]
});
}
try {
await client.batch(queries, { prepare: true });
} catch (err) {
return next(err);
}
}
try {
await client.batch(queries, { prepare: true });
} catch (err) {
return next(err);
}
}
next();
},
async function test(next) {
const query = "SELECT COUNT(1) FROM benchmarks.basic USING TIMEOUT 120s;";
try {
let res = await client.execute(query);
assert(res.rows[0].count == iterCnt);
} catch (err) {
return next(err);
next();
},
async function test(next) {
utils.checkRowCount(client, iterCnt, next);
},
function r() {
exit(0);
}
next();
},
function r() {
exit(0);
}
], utils.onError);
], utils.onError);
};
27 changes: 27 additions & 0 deletions benchmark/logic/benchmark.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
// Single entry point for all benchmarks.
// This file takes the name of the benchmark, including driver name, step count and concurrency level
// and runs the corresponding benchmark logic, loaded at runtime.

"use strict";
const { exit } = require("process");
const utils = require("./utils");

const defaultConcurrencyLevel = 100;

const driverName = process.argv[2];
const benchmarkName = process.argv[3];
// Each individual benchmark has a default step count defined.
// See the explanation in config.yml for more details.
const stepCount = process.argv[4] !== "default" ? parseInt(process.argv[4], 10) : undefined;
const concurrencyLevel = process.argv[5] !== "default" ? parseInt(process.argv[5], 10) : defaultConcurrencyLevel;

if (!driverName || !benchmarkName) {
console.error("Usage: node benchmark.js <driver> <benchmark-name> [step-count] [concurrency]");
exit(1);
}

const cassandra = require(driverName);
const client = new cassandra.Client(utils.getClientArgs());

const benchmark = require(`./${benchmarkName}`);
benchmark(cassandra, client, stepCount, concurrencyLevel);
79 changes: 39 additions & 40 deletions benchmark/logic/concurrent_deser.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,52 +1,51 @@
"use strict";
const async = require("async");
// Possible values of argv[2] (driver) are scylladb-driver-alpha and cassandra-driver.
const cassandra = require(process.argv[2]);
const utils = require("./utils");
const { exit } = require("process");
const assert = require("assert");

const client = new cassandra.Client(utils.getClientArgs());
const iterCnt = parseInt(process.argv[3]);
module.exports = function (cassandra, client, stepCount, concurrencyLevel) {
// REMEMBER: update benchmark config.yml when changing the constant value.
const iterCnt = stepCount || 2000;

async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaDeSer, next);
},
async function insert(next) {
let allParameters = utils.insertConcurrentDeSer(cassandra, iterCnt);
try {
const _result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true });
} catch (err) {
return next(err);
}
next();
},
async function select(next) {
let remaining = iterCnt;
while (remaining > 0) {
let currentStep = Math.min(remaining, 500);
remaining -= currentStep;
let allParameters = [];
for (let i = 0; i < currentStep; i++) {
allParameters.push({
query: 'SELECT * FROM benchmarks.basic',
});
}
async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaDeSer, next);
},
async function insert(next) {
let allParameters = utils.insertConcurrentDeSer(cassandra, iterCnt);
try {
const result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true, collectResults: true });
for (let singleResult of result.resultItems) {
assert.equal(singleResult.rowLength, iterCnt);
}
const _result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true, concurrencyLevel });
} catch (err) {
return next(err);
}
next();
},
async function select(next) {
let remaining = iterCnt;
while (remaining > 0) {
let currentStep = Math.min(remaining, 500);
remaining -= currentStep;
let allParameters = [];
for (let i = 0; i < currentStep; i++) {
allParameters.push({
query: 'SELECT * FROM benchmarks.basic',
});
}
try {
const result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true, collectResults: true, concurrencyLevel });
for (let singleResult of result.resultItems) {
assert.equal(singleResult.rowLength, iterCnt);
}
} catch (err) {
return next(err);
}
}
next();
},
function r() {
exit(0);
}
next();
},
function r() {
exit(0);
}
], utils.onError);

], utils.onError);
};
67 changes: 33 additions & 34 deletions benchmark/logic/concurrent_insert.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,42 +1,41 @@
"use strict";
const async = require("async");
// Possible values of argv[2] (driver) are scylladb-driver-alpha and cassandra-driver.
const cassandra = require(process.argv[2]);
const utils = require("./utils");
const { exit } = require("process");

const client = new cassandra.Client(utils.getClientArgs());
const iterCnt = parseInt(process.argv[3]);
module.exports = function (cassandra, client, stepCount, concurrencyLevel) {
// REMEMBER: update benchmark config.yml when changing the constant value.
const iterCnt = stepCount || 4000000;

async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
let limited = async function (steps) {
let allParameters = [];
for (let i = 0; i < steps; i++) {
allParameters.push({
query: 'INSERT INTO benchmarks.basic (id, val) VALUES (?, ?)',
params: [cassandra.types.Uuid.random(), 10]
});
async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
let limited = async function (steps) {
let allParameters = [];
for (let i = 0; i < steps; i++) {
allParameters.push({
query: 'INSERT INTO benchmarks.basic (id, val) VALUES (?, ?)',
params: [cassandra.types.Uuid.random(), 10]
});
}
try {
const _result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true, concurrencyLevel });
} catch (err) {
return next(err);
}
}
try {
const _result = await cassandra.concurrent.executeConcurrent(client, allParameters, { prepare: true });
} catch (err) {
return next(err);
}
}
await utils.repeatCapped(limited, iterCnt);

next();
},
async function test(next) {
utils.checkRowCount(client, iterCnt, next);
},
function r() {
exit(0);
}
], utils.onError);
await utils.repeatCapped(limited, iterCnt);

next();
},
async function test(next) {
utils.checkRowCount(client, iterCnt, next);
},
function r() {
exit(0);
}
], utils.onError);
};
74 changes: 36 additions & 38 deletions benchmark/logic/concurrent_paging.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,48 +1,46 @@
"use strict";
const async = require("async");
// Possible values of argv[2] (driver) are scylladb-nodejs-rs-driver and cassandra-driver.
const cassandra = require(process.argv[2]);
const utils = require("./utils");
const { exit } = require("process");
const { assert } = require("console");
const assert = require("assert");

const client = new cassandra.Client(utils.getClientArgs());
const iterCnt = parseInt(process.argv[3]);
const concurrencyLevel = 20;
module.exports = function (cassandra, client, stepCount, concurrencyLevel) {
// REMEMBER: update benchmark config.yml when changing the constant value.
const iterCnt = stepCount || 1280;

async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
utils.insertSimple(client, 50, next);
},
async function select(next) {
let limited = async function (steps) {
for (let i = 0; i < steps; i++) {
try {
let s = 0;
let q = await client.execute('SELECT * FROM benchmarks.basic', [], { prepare: true, fetchSize: 1 });
for await (const row of q) {
s += row['val'];
async.series(
[
function initialize(next) {
utils.prepareDatabase(client, utils.tableSchemaBasic, next);
},
async function insert(next) {
utils.insertSimple(client, 50, next);
},
async function select(next) {
let limited = async function (steps) {
for (let i = 0; i < steps; i++) {
try {
let s = 0;
let q = await client.execute('SELECT * FROM benchmarks.basic', [], { prepare: true, fetchSize: 1 });
for await (const row of q) {
s += row['val'];
}
assert(s === 5000);
} catch (err) {
return next(err);
}
assert(s === 5000);
} catch (err) {
return next(err);
}

}
await utils.executeMultipleRepeatCapped(limited, iterCnt, concurrencyLevel);
next();
},
function r() {
exit(0);
}
], function (err) {
if (err) {
console.error("Error: ", err.message, err.stack);
exit(1);
}
await utils.executeMultipleRepeatCapped(limited, iterCnt, concurrencyLevel);
next();
},
function r() {
exit(0);
}
], function (err) {
if (err) {
console.error("Error: ", err.message, err.stack);
exit(1);
}
},);
});
};
Loading
Loading