Malachi.Loadtest.Payload (malachi v0.22.7)

View Source

The record values the load generators send, as one of three modes chosen by --payload:

  • :constant (default) - every value is the byte x repeated. The series published so far were all measured with it, which is the only reason it stays the default: a run of one byte compresses to almost nothing, so no compression figure taken with it means anything.
  • :json - a seeded, event-like JSON document of exactly record_size bytes (see below).
  • :random - seeded uniform bytes, the incompressible control (ratio about 1.00).

scripts/lib/payload.js implements the same algorithm, the same document and the same pool, so the same mode, seed and size give byte-identical values in both generators. Golden vectors shared by both test suites (test/support/fixtures/loadtest/payload_vectors.json) pin that.

The generator

xoshiro128** (Blackman and Vigna, "Scrambled linear pseudorandom number generators", 2018), whose state is four 32-bit words. It needs only 32-bit shifts, rotations and multiplications by 5 and 9, which both languages can do exactly: here every step is masked to 32 bits, and the JavaScript side uses Math.imul and >>> 0. The 32-bit seed is expanded into the state with SplitMix64, the seeding the algorithm's authors recommend: two SplitMix64 outputs, each split into its low and high 32-bit words, give s0, s1, s2 and s3 in that order. A bounded draw is rem(next, n); its modulo bias is irrelevant here and identical in both languages, which is what matters.

The JSON document

ASCII only, keys in this order, no whitespace:

{"id":"<16 hex>","ts":<13 digits>,"type":"<1 of 8>","user":"u<0000..9999>",
 "region":"<1 of 6>","status":"<1 of 4>","amount":<100..99999>,"msg":"<words>"}

id is two draws in lowercase hex (unique in practice, like an event id). ts grows with the value's position in the pool from a fixed base, plus 0 to 6 ms, never from the clock, so a value depends on the seed alone. msg is words drawn from a 64-word vocabulary, separated by spaces, and filled until the document has exactly record_size bytes, cutting the last word. A size below min_json_size/0 cannot hold the fields and is refused rather than shortened.

The cardinalities are what decide the compression ratio, and the block size (the records per produce, --batch) decides how much of that redundancy one block can see. With zstd level 1 at 256-byte values the document lands near the reference band measured in #186 on JSON of about 200 bytes (1.17x at one record per block, 2.6x at 10, 3.8x at 100, 4.1x at 1000); the test suite pins it inside that band. Measured on Linux with seed 1: 1.32x, 2.66x, 3.55x and 3.82x at 1, 10, 100 and 1000 records per block. Random values give 1.00x from 10 records per block (0.96x for one, the frame's own overhead), and constant bytes 13x at one record and 135x at ten.

The pool

Generating values inside the measured window would make the generator the bottleneck, so the values are generated once, before it, into a pool every connection shares. A batch never repeats a value: the pool holds whole batches of distinct values. One connection only sends a value again after it has sent the whole pool, more than the window zstd uses at its low levels.

The pool holds min(ceil(8MiB / record_size), 65536) values, rounded up to whole batches, and at least pipeline batches for every connection, where pipeline is how many produces a connection keeps in flight (--pipeline, 1 for a closed loop). The cap keeps the pool to 65536 values below 128-byte values instead of millions of tiny ones, so there it is smaller than 8MiB (4MiB at 64 bytes). Connection i starts at batch i * stride, with stride = div(batches, connections), which is therefore at least pipeline, and walks forward on its own cursor with no counter shared between connections. So every connection's first burst of pipeline batches is its own. Large batches at many connections, and deep pipelines, get a larger pool: batch 4096 of 256-byte values at 64 connections holds 64MiB, and 2GiB pipelined 32 deep; Malachi.Loadtest says so on stderr above 512MiB.

That is all that is guaranteed between connections. The cursors move at each connection's own pace, so a connection that runs stride batches ahead of its neighbour reaches the batches the neighbour is sending, and from then on both send the batches they share close together, which a store compressing several batches at once (#202, #209) sees as repetition. A wider stride, from a larger pool relative to the connections, takes longer to close.

Summary

Types

A payload mode.

The xoshiro128** state: four 32-bit words.

Functions

Raises ArgumentError when record_size cannot hold a json value.

The seed json and random use when none is given.

The pool for mode, as a tuple of batches already encoded by Malachi.Wire.encode_produce_records/1, so a produce only joins one to its topic. Value pos of the pool is keyed k<rem(pos, keys)>. The tuple is meant to be shared rather than handed to every connection: Malachi.Loadtest keeps it in :persistent_term for the run, since a term sent or captured into a process is copied there, small batches included.

The smallest record_size a json value fits in: the document with every field at its widest and an empty msg.

The payload modes, in the order --payload lists them.

The next 32-bit output and the advanced state.

How many values the pool holds for record_size, batch, connections and the pipeline depth each connection keeps in flight (1 for a closed loop): see the moduledoc. Always a whole number of batches, and at least pipeline batches per connection.

The xoshiro128** state for a 32-bit seed, expanded with SplitMix64.

The batch connection index starts at, out of batches, when connections share the pool.

The first count values of mode for seed at record_size bytes, in pool order. constant ignores the seed. Raises ArgumentError for a json size below min_json_size/0.

Types

mode()

@type mode() :: :constant | :json | :random

A payload mode.

state()

@opaque state()

The xoshiro128** state: four 32-bit words.

Functions

check_json_size!(record_size)

@spec check_json_size!(pos_integer()) :: :ok

Raises ArgumentError when record_size cannot hold a json value.

default_seed()

@spec default_seed() :: pos_integer()

The seed json and random use when none is given.

encoded_batches(mode, seed, record_size, batch, keys, connections, pipeline)

@spec encoded_batches(
  mode(),
  non_neg_integer(),
  pos_integer(),
  pos_integer(),
  pos_integer(),
  pos_integer(),
  pos_integer()
) :: tuple()

The pool for mode, as a tuple of batches already encoded by Malachi.Wire.encode_produce_records/1, so a produce only joins one to its topic. Value pos of the pool is keyed k<rem(pos, keys)>. The tuple is meant to be shared rather than handed to every connection: Malachi.Loadtest keeps it in :persistent_term for the run, since a term sent or captured into a process is copied there, small batches included.

min_json_size()

@spec min_json_size() :: pos_integer()

The smallest record_size a json value fits in: the document with every field at its widest and an empty msg.

modes()

@spec modes() :: [mode()]

The payload modes, in the order --payload lists them.

next(arg)

@spec next(state()) :: {non_neg_integer(), state()}

The next 32-bit output and the advanced state.

pool_size(record_size, batch, connections, pipeline)

@spec pool_size(pos_integer(), pos_integer(), pos_integer(), pos_integer()) ::
  pos_integer()

How many values the pool holds for record_size, batch, connections and the pipeline depth each connection keeps in flight (1 for a closed loop): see the moduledoc. Always a whole number of batches, and at least pipeline batches per connection.

seed(seed)

@spec seed(non_neg_integer()) :: state()

The xoshiro128** state for a 32-bit seed, expanded with SplitMix64.

start_batch(index, batches, connections)

@spec start_batch(non_neg_integer(), pos_integer(), pos_integer()) ::
  non_neg_integer()

The batch connection index starts at, out of batches, when connections share the pool.

values(atom, seed, record_size, count)

@spec values(mode(), non_neg_integer(), pos_integer(), non_neg_integer()) :: [
  binary()
]

The first count values of mode for seed at record_size bytes, in pool order. constant ignores the seed. Raises ArgumentError for a json size below min_json_size/0.