Compare throughput against jsmpp and Cloudhopper on one sink #15

Merged
lilleman merged 2 commits from peer-throughput into main 2026-09-20 23:52:16 +02:00
8 changed files with 270 additions and 1 deletions
+30
View File
@@ -62,6 +62,36 @@ Against real peers, same driver, window 50, 20,000 messages:
| Jasmin 0.10 | 2,207 | 0 | | Jasmin 0.10 | 2,207 | 0 |
| SMPPSim 3.0.0 | — | 19,000 of 20,000 | | SMPPSim 3.0.0 | — | 19,000 of 20,000 |
## Against the other client libraries
Same sink, same 100,000 single-segment messages, same host. This is the comparison that means
something: every client is measured pushing into *our* server, so the server's work is common to all
three and only the client differs.
```bash
docker compose -f compose.yaml -f benchmarks/compose.jsmpp.yaml up -d --build
docker compose -f compose.yaml -f benchmarks/compose.jsmpp.yaml run --rm node \
node benchmarks/peer-load.ts --driver=http://jsmpp:8080 --count=100000 --concurrency=50
```
| Window | this library | jsmpp 3.0.3 | Cloudhopper 5.0.10 |
| --- | --- | --- | --- |
| 10 | 25,358 | 30,771 | 27,945 |
| 50 | 38,675 | 40,934 | 32,384 |
| 200 | 40,046 | 42,105 | 25,497 |
**We are slowest at the default window**, which is the setting most callers will ever run — 25,358
against jsmpp's 30,771. That is the throughput work worth doing, and it is worth doing there.
Two things the table does not show. This library does it on one event loop where both Java peers
spend one OS thread per in-flight request, which is why Cloudhopper falls off at 200 threads and we
do not. And all three are pushing into the same Node sink, whose own cost is in every number, so the
differences between clients are compressed rather than exaggerated here.
Kannel is absent deliberately: it is a gateway rather than a client library, wired here as an ESME
that forwards from its own spool, so loading it would measure its HTTP frontend and queue rather
than an SMPP client. The number would not belong in this table.
Jasmin routes and persists where the sink does neither, so the gap is not an efficiency ratio Jasmin routes and persists where the sink does neither, so the gap is not an efficiency ratio
between two comparable things — what it establishes is that this library is not the bottleneck between two comparable things — what it establishes is that this library is not the bottleneck
against a production SMSC, by more than an order of magnitude. SMPPSim's store fills at roughly a against a production SMSC, by more than an order of magnitude. SMPPSim's store fills at roughly a
+25
View File
@@ -0,0 +1,25 @@
x-log-limits: &log-limits
logging:
driver: json-file
options:
max-file: "3"
max-size: 20m
# The peer dials the host it was given at build time, "node", so the sink answers under that name.
services:
cloudhopper:
build: ./interop-tests/peers/cloudhopper
image: interop-cloudhopper-load:5.0.10-ae6485a
command: ["node", "2775"]
<<: *log-limits
healthcheck:
test: ["CMD-SHELL", "bash -c 'exec 3<>/dev/tcp/127.0.0.1/8080'"]
interval: 1s
retries: 30
timeout: 2s
node:
command: ["node", "benchmarks/smsc-sink.ts"]
environment:
NPM_CONFIG_CACHE: /tmp/npm-cache
PORT: "2775"
+25
View File
@@ -0,0 +1,25 @@
x-log-limits: &log-limits
logging:
driver: json-file
options:
max-file: "3"
max-size: 20m
# The peer dials the host it was given at build time, "node", so the sink answers under that name.
services:
jsmpp:
build: ./interop-tests/peers/jsmpp
image: interop-jsmpp-load:3.0.3-a24db96
command: ["node", "2775"]
<<: *log-limits
healthcheck:
test: ["CMD-SHELL", "bash -c 'exec 3<>/dev/tcp/127.0.0.1/8080'"]
interval: 1s
retries: 30
timeout: 2s
node:
command: ["node", "benchmarks/smsc-sink.ts"]
environment:
NPM_CONFIG_CACHE: /tmp/npm-cache
PORT: "2775"
+33
View File
@@ -0,0 +1,33 @@
/**
* Drives a peer's HTTP control surface through the same load the local driver runs, so the number
* that comes back is that library's own rate against our sink rather than ours against theirs.
*/
function arg(name: string, fallback: string): string {
const found = process.argv.find(one => one.startsWith(`--${name}=`));
return found === undefined ? fallback : found.slice(name.length + 3);
}
const driver = arg('driver', 'http://jsmpp:8080');
const count = arg('count', '20000');
const concurrency = arg('concurrency', '50');
async function call(path: string): Promise<unknown> {
const response = await fetch(`${driver}${path}`);
return response.json();
}
// Cloudhopper's window defaults to 1 and is set at bind, so threads alone would serialise it.
const bound = await call(`/bind?systemId=bench&password=benchpw&windowSize=${concurrency}`);
if (typeof bound !== 'object' || bound === null || !('ok' in bound) || bound.ok !== true) {
process.stdout.write(`${JSON.stringify({ bind: bound })}\n`);
process.exit(1);
}
const loaded = await call(`/load?count=${count}&concurrency=${concurrency}`);
process.stdout.write(`${JSON.stringify(loaded)}\n`);
await call('/unbind');
+22 -1
View File
@@ -39,7 +39,28 @@ Traced as far as `oserl`'s `smpp_pdu_syntax:pack/2` (the `trx_deadlock_fix_1` br
`rebar.config` pins), which builds the header as plain 32-bit bit-syntax `rebar.config` pins), which builds the header as plain 32-bit bit-syntax
(`<<Len:32, CmdId:32, 0:32, SeqNum:32>>`) - correct on inspection, so the corruption happens (`<<Len:32, CmdId:32, 0:32, SeqNum:32>>`) - correct on inspection, so the corruption happens
somewhere between that call and the socket write, not chased further given the time-box. Reproduced somewhere between that call and the socket write, not chased further given the time-box. Reproduced
identically on three separate runs (byte-for-byte). Recorded as **blocked**; `smppload.test.ts` Re-examined 2026-09-20 to see whether it could be unblocked for the throughput comparison in
`benchmarks/`. Four things are now established, and one earlier suspicion is ruled out:
- **It is one write, not a split one.** A raw listener that accumulates every chunk rather than
reading the first receives `40 octets across 1 chunks`, byte-identical to the 2026-09-06 capture,
with `command_length` reading 2,752,512. So the two leading zero octets are absent from the socket
write itself; nothing about our framing or the capture is involved.
- **The compiled `pack/2` is correct**, checked in the built tree rather than the repository:
`Len = size(BodyBin) + 16` written as `<<Len:32, CmdId:32, 0:32, SeqNum:32>>`, returned as the
iolist `[Header, BodyBin]`. For this bind that is 16 + 26 = 42.
- **The remaining suspect is `smpp_session.erl:158`**, which writes with `erlang:port_command/2`
rather than `gen_tcp:send/2` — an undocumented fast path in oserl code that predates OTP 27.
- **That suspect is untested.** Two attempts to swap it were both invalidated by rebar3 dep caching:
editing a fetched dependency's source does not rebuild its beam, and the `_checkouts/` route
re-verifies every dependency, which needs network and git in the build container. Whoever picks
this up should patch before the first compile, or force the dep to rebuild, and confirm the beam
actually changed before believing a result.
Enough for an upstream report — a reproducer needing no SMSC, the exact octets, and a named
suspect — but not enough for a patch, since the one-line candidate has never actually run.
Recorded as **blocked**; `smppload.test.ts`
keeps a live reproducer asserting what our server does when it receives it (refuses the stream as keeps a live reproducer asserting what our server does when it receives it (refuses the stream as
unframeable - see Scenarios) rather than removing the peer. `smpp-dumb-client` covers S9, and unframeable - see Scenarios) rather than removing the peer. `smpp-dumb-client` covers S9, and
substitutes for S6 and (partially) S8 - see below. substitutes for S6 and (partially) S8 - see below.
@@ -64,6 +64,7 @@ public final class Driver {
server.createContext("/bind", Driver::handleBind); server.createContext("/bind", Driver::handleBind);
server.createContext("/unbind", Driver::handleUnbind); server.createContext("/unbind", Driver::handleUnbind);
server.createContext("/submit", Driver::handleSubmit); server.createContext("/submit", Driver::handleSubmit);
server.createContext("/load", Driver::handleLoad);
server.createContext("/windowBurst", Driver::handleWindowBurst); server.createContext("/windowBurst", Driver::handleWindowBurst);
server.createContext("/sendWindowSize", Driver::handleSendWindowSize); server.createContext("/sendWindowSize", Driver::handleSendWindowSize);
server.setExecutor(null); server.setExecutor(null);
@@ -224,6 +225,62 @@ public final class Driver {
} }
} }
/**
* Pushes count messages and reports only the rate. Cloudhopper's submit blocks on the response,
* so the pool size is what puts requests in flight — the same shape as the other peers' load.
*/
private static void handleLoad(HttpExchange exchange) {
Map<String, String> p = queryParams(exchange);
SmppSession session = sessions.get(p.getOrDefault("session", "default"));
if (session == null) {
respond(exchange, 200, Json.write(Map.of("ok", false, "error", "no such session")));
return;
}
int count = Integer.parseInt(p.getOrDefault("count", "20000"));
int concurrency = Integer.parseInt(p.getOrDefault("concurrency", "50"));
long timeoutMs = Long.parseLong(p.getOrDefault("timeoutMs", "60000"));
String from = p.getOrDefault("from", "1000");
String to = p.getOrDefault("to", "2000");
AtomicInteger issued = new AtomicInteger();
AtomicInteger failed = new AtomicInteger();
ExecutorService pool = Executors.newFixedThreadPool(concurrency);
long started = System.nanoTime();
for (int worker = 0; worker < concurrency; worker++) {
pool.execute(() -> {
while (issued.getAndIncrement() < count) {
try {
session.submit(buildSubmit(from, to, "benchmark"), timeoutMs);
} catch (Exception e) {
failed.incrementAndGet();
}
}
});
}
pool.shutdown();
try {
if (!pool.awaitTermination(10, java.util.concurrent.TimeUnit.MINUTES)) pool.shutdownNow();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
double seconds = (System.nanoTime() - started) / 1e9;
Map<String, Object> result = new LinkedHashMap<>();
result.put("ok", true);
result.put("count", count);
result.put("concurrency", concurrency);
result.put("failed", failed.get());
result.put("seconds", Math.round(seconds * 1000d) / 1000d);
result.put("perSecond", Math.round(count / seconds));
respond(exchange, 200, Json.write(result));
}
/** Fires `count` submits at once, each tagged by index in its text, to probe window pressure. */ /** Fires `count` submits at once, each tagged by index in its text, to probe window pressure. */
private static void handleWindowBurst(HttpExchange exchange) { private static void handleWindowBurst(HttpExchange exchange) {
Map<String, String> p = queryParams(exchange); Map<String, String> p = queryParams(exchange);
@@ -38,6 +38,10 @@ import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/** /**
* An HTTP-driven jsmpp ESME: each request binds (if needed), performs one scenario action against * An HTTP-driven jsmpp ESME: each request binds (if needed), performs one scenario action against
@@ -64,6 +68,7 @@ public final class Driver {
server.createContext("/unbind", Driver::handleUnbind); server.createContext("/unbind", Driver::handleUnbind);
server.createContext("/enquireLink", Driver::handleEnquireLink); server.createContext("/enquireLink", Driver::handleEnquireLink);
server.createContext("/submit", Driver::handleSubmit); server.createContext("/submit", Driver::handleSubmit);
server.createContext("/load", Driver::handleLoad);
server.createContext("/querySm", exchange -> handleUnhandledCommand(exchange, "query")); server.createContext("/querySm", exchange -> handleUnhandledCommand(exchange, "query"));
server.createContext("/cancelSm", exchange -> handleUnhandledCommand(exchange, "cancel")); server.createContext("/cancelSm", exchange -> handleUnhandledCommand(exchange, "cancel"));
server.createContext("/replaceSm", exchange -> handleUnhandledCommand(exchange, "replace")); server.createContext("/replaceSm", exchange -> handleUnhandledCommand(exchange, "replace"));
@@ -267,6 +272,70 @@ public final class Driver {
} }
} }
/**
* Pushes count messages over an already-bound session and reports the rate. jsmpp's submit is
* blocking, so threads are what put requests in flight here — the window is the pool size.
*/
private static void handleLoad(HttpExchange exchange) {
Map<String, String> p = queryParams(exchange);
SMPPSession session = sessions.get(p.getOrDefault("session", "default"));
if (session == null) {
Map<String, Object> missing = new LinkedHashMap<>();
missing.put("ok", false);
missing.put("error", "no such session");
respondOk(exchange, missing);
return;
}
int count = Integer.parseInt(p.getOrDefault("count", "20000"));
int concurrency = Integer.parseInt(p.getOrDefault("concurrency", "50"));
String from = p.getOrDefault("from", "BENCH");
String to = p.getOrDefault("to", "46709771337");
String text = p.getOrDefault("text", "benchmark");
AtomicInteger issued = new AtomicInteger();
AtomicInteger failed = new AtomicInteger();
ExecutorService pool = Executors.newFixedThreadPool(concurrency);
long started = System.nanoTime();
for (int worker = 0; worker < concurrency; worker++) {
pool.execute(() -> {
while (issued.getAndIncrement() < count) {
try {
session.submitShortMessage("CMT",
TypeOfNumber.INTERNATIONAL, NumberingPlanIndicator.UNKNOWN, from,
TypeOfNumber.INTERNATIONAL, NumberingPlanIndicator.UNKNOWN, to,
new ESMClass(), (byte) 0, (byte) 1, null, null,
new RegisteredDelivery(SMSCDeliveryReceipt.DEFAULT), (byte) 0,
dataCoding("ascii"), (byte) 0, encode(text, "ascii"));
} catch (Exception e) {
failed.incrementAndGet();
}
}
});
}
pool.shutdown();
try {
if (!pool.awaitTermination(10, TimeUnit.MINUTES)) pool.shutdownNow();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
double seconds = (System.nanoTime() - started) / 1e9;
Map<String, Object> result = new LinkedHashMap<>();
result.put("ok", true);
result.put("count", count);
result.put("concurrency", concurrency);
result.put("failed", failed.get());
result.put("seconds", Math.round(seconds * 1000d) / 1000d);
result.put("perSecond", Math.round(count / seconds));
respondOk(exchange, result);
}
private static void submitPlain(SMPPSession session, String from, String to, String text, String encoding, private static void submitPlain(SMPPSession session, String from, String to, String text, String encoding,
java.util.List<Map<String, Object>> segments) throws Exception { java.util.List<Map<String, Object>> segments) throws Exception {
SubmitSmResult r = session.submitShortMessage("CMT", SubmitSmResult r = session.submitShortMessage("CMT",
+9
View File
@@ -225,6 +225,15 @@ and is also what the panel ranked hardest — two methods, one answer.
send — and not this one. If goal 2 forbids reporting on a message we cannot fully account for, send — and not this one. If goal 2 forbids reporting on a message we cannot fully account for,
that is the answer; it is stated in no file today either way. that is the answer; it is stated in no file today either way.
### Throughput — goal 6, and the default window is where we are slowest
- [ ] **Close the gap to jsmpp at `maxOutstanding: 10`.** Measured 2026-09-20 against the same sink,
100,000 messages each: this library 25,358/s, jsmpp 30,771/s, Cloudhopper 27,945/s — we are
last at the one window most callers will ever run, while leading Cloudhopper and trailing jsmpp
by only 5% at 50 and 200. So the cost is not the codec, which the higher windows exercise just
as hard; it is something per-request that the window hides once enough requests overlap.
`benchmarks/` reproduces all three. Goal 6.
### Locality — 5–6 today, and the gate is 7 ### Locality — 5–6 today, and the gate is 7
- [ ] **Give `IncomingRequests` a port instead of the `Session` it drives.** It holds its owner and - [ ] **Give `IncomingRequests` a port instead of the `Session` it drives.** It holds its owner and