iClaustron training for C developers · module 5 of 8 · 16 min

Key operations, asynchronous

Define the work, send it, carry on, and collect the results as they come: many transactions in flight on one thread.

The calls · from the header

Send, poll and flush: the three calls of asynchronous work

int32_t ic_apid_conn_send(IC_APID_CONNECTION *conn, bool force);

int32_t ic_apid_conn_poll(IC_APID_CONNECTION *conn,
                          uint32_t wait_ms,
                          uint32_t *signals);

int32_t ic_apid_conn_flush(IC_APID_CONNECTION *conn,
                           uint32_t wait_ms,
                           bool force,
                           uint32_t *signals);
sendEvery query defined, commit and rollback: one write per data node.
pollWaits up to wait_ms, completes what replies finish, runs callbacks.
flushSend, then poll. The poll always runs, and reports a lost link.
errorsCome back on the queries and transactions, not from poll.

iclaustron: include/iclaustron/ic_apid.h (ic_apid_conn_send, ic_apid_conn_poll, ic_apid_conn_flush); crates/ic_apid/src/transaction.rs (send_queries, flush); doc/user-guide.md §5

The example · examples/c/pipeline.c · 32 in flight by default

The pipeline example: a slot per transaction, passed as user_ref

struct pipeline {
  IC_TABLE_DEF *table;
  IC_RECORD *rec;
  uint32_t next;       /* the next id to write */
  /* ... end, done, failed, first_error */
};
/* One query in flight: the callback's user_ref. */
struct slot {
  struct pipeline *p;
  IC_QUERY_ID query;
  IC_TRANS_ID trans;
};
static struct slot slots[MAX_IN_FLIGHT];
/* ... */
for (i = 0; i < width; i++) {
  slots[i].p = &p;
  EX_CHECK(ic_apid_query_create(conn, p.table, p.rec, p.rec,
                                &slots[i].query));
  EX_CHECK(start_write(conn, &slots[i], p.next++));
}
while (p.done < rows)
  EX_CHECK(ic_apid_conn_flush(conn, 10, 0, NULL));

The slots are static, so each user_ref stays valid while its query is out. main starts a write per slot, then only flushes.

iclaustron: examples/c/pipeline.c (struct pipeline, struct slot, main); examples/README.md (pipeline); doc/user-guide.md §5, §8

Step 1 · define · start_write() in the example

Step 1, define: the request is packed for its node, not sent

key = ic_apid_query_key_row(conn, s->query, &key_len);
attr = ic_apid_query_attr_row(conn, s->query, &attr_len);
/* ... */
memcpy(key, &row, sizeof row);
memcpy(attr, &row, sizeof row);
hint = ic_apid_conn_hint_for_key(conn, p->table, p->rec, key, key_len);
ret = ic_apid_conn_start_transaction(conn, hint, p->table, &s->trans);
/* ... */
ic_write_key_args_init(&args, sizeof args);
args.kind = IC_WRITE_KIND_WRITE;
args.callback = on_written;
args.user_ref = s;
ret = ic_apid_conn_write_key(conn, s->query, s->trans, &args);
/* ... */
return ic_apid_conn_commit_transaction(conn, s->trans);
ThreadOnly this user thread: the request waits for the next send.
TransactionA coordinator record at the hinted node, kept by the connection.
QueryIdle → defined: its rows are the library's until it completes.
RequestKEY_OP_REQ packed into that node's buffer; the commit rides on it.

iclaustron: examples/c/pipeline.c (start_write); crates/ic_apid/src/transaction.rs (start_transaction, define, commit_transaction); crates/ic_apid/src/apid_conn.rs (coord_record)

Step 2 · send · the first half of ic_apid_conn_flush

Step 2, send: all the requests, one write per data node

/* Send what is defined and take the replies; the callbacks, run
 * inside, keep the pipeline full. Without force, a send may wait a
 * moment for company, never longer than the adaptive send allows. */
while (p.done < rows)
  EX_CHECK(ic_apid_conn_flush(conn, 10, 0, NULL));

Each round of this loop sends, in one go, every write the callbacks of the round before defined.

TransactionsOnly those with work; start, execute and commit flags set in place.
QueriesDefined → sent, all of them in this one call.
Per nodeEvery transaction's requests for a node in one buffer: one write.
ThreadsThis thread writes the socket itself, unless another one is writing.

iclaustron: examples/c/pipeline.c (main); crates/ic_apid/src/transaction.rs (send_queries, send_transaction, send_query); crates/ic_apid/src/apid_global.rs (NodeShared::send_words_for_generation)

Step 3 · poll · the second half of ic_apid_conn_flush

Step 3, poll: the receive thread routes, the poll completes

EX_CHECK(ic_apid_conn_flush(conn, 10, 0, NULL));   /* send, then poll up to 10 ms */
data nodes KEY_OP_CONF:coordinator ROW_DATA_REP:reading node receive thread reads its socketsunpacks BUNDLE routes each signalby block number inbox a page per threadper round, under one short lockwakes it if asleep poll on the user threadKEY_OP_CONF: index ROW_DATA_REP: idrows to attr rows complete off its transactionthen its callback, or the executed list

Replies come from two places, in either order: KEY_OP_CONF from the coordinator, ROW_DATA_REP from the node that read the row.

The receive thread reads nothing but the block number, and posts one page per user thread per round, under one short lock.

The poll takes every posted page at once. A confirmation finds its transaction by one array index, a row its query by its id.

A completed query leaves its transaction's lists; then its callback runs, or it goes on the executed list.

iclaustron: crates/ic_apid/src/thread_conn.rs (ThreadConnection::post, exchange_with_requests); crates/ic_apid/src/apid_conn.rs (poll_with_requests); crates/ic_apid/src/transaction.rs (take_key_conf, take_row, finish_query); doc/rust/05-ndb-protocol.md §6.5

In flight · the threads over time

Many transactions in flight: the thread works while the nodes do

user threadits connection receive threadroutes only data nodestwo of them time → define T1 T2 T3 send: one writeper data node node 1: T1 T3 node 2: T2 define T4 T5 T6 node 1: T4 T6 node 2: T5 poll: waits, then completes T1–T3 their callbacks define T7–T9 route node 1: T8 node 2: T7 T9 route inbox poll: completes T4–T6 callbacks define T10–T12 …

Three single-write transactions are defined, then sent: one write per data node, by this thread itself, straight to the socket.

The thread does not wait for them. It defines three more and sends those too: six transactions in flight on one connection.

A receive thread only routes the replies to this thread's inbox. The poll completes T1–T3 and runs their callbacks, which define T7–T9.

The next send carries T7–T9, one write per node, while T4–T6's replies wait in the inbox for the next poll: two batches always out.

iclaustron: doc/rust/architecture.html §3 (pipelines on one thread); crates/ic_apid/src/transaction.rs (send_queries, finish_query); crates/ic_apid/src/thread_conn.rs (post); doc/user-guide.md §5

Completions · the executed list or a callback

Two ways to collect completions: the executed list or a callback

typedef struct {
  /* ... size, kind, abort_option */
  void (*callback)(IC_APID_CONNECTION *conn, IC_QUERY_ID query,
                   void *user_ref);  /* NULL: the executed list */
  void *user_ref;   /* given back with the completed query */
} IC_READ_KEY_ARGS;   /* and IC_WRITE_KEY_ARGS */

bool ic_apid_conn_get_next_executed_query(IC_APID_CONNECTION *conn,
                                          IC_QUERY_ID *query);
void *ic_apid_query_user_ref(IC_APID_CONNECTION *conn, IC_QUERY_ID query);
Executed listNo callback: it waits on the list, oldest first; taking it idles it.
CallbackCalled inside the poll, on the polling thread, with its user_ref.
user_refAny pointer of yours, given back with the query either way.
Per queryChosen at each definition: one connection may use both.

iclaustron: include/iclaustron/ic_apid.h (IC_READ_KEY_ARGS, IC_WRITE_KEY_ARGS, ic_apid_conn_get_next_executed_query, ic_apid_query_user_ref); crates/ic_apid/src/transaction.rs (get_next_executed_query, finish_query)

Callbacks · the rules

Inside a callback: what it sees, what it may do, what it must not

It may

  • read the query's error and its rows
  • define queries, this one included
  • start, commit, roll back and close transactions
  • send

It runs

  • on the polling thread, inside the poll
  • with its query complete, the transaction settled as far as this query decides it
  • where a poll does nothing, and a flush only sends
  • and must not free the connection

A committed one-query transaction is done in its callback: close it there. On return the query is idle, unless defined anew.

iclaustron: crates/ic_apid/src/transaction.rs (finish_query, take_key_conf, flush); crates/ic_apid/src/apid_conn.rs (poll_with_requests: in_callback); crates/ic_apid/src/query.rs (QueryCallback); doc/user-guide.md §8

The example · on_written() in pipeline.c

The pipeline's callback closes its transaction and starts the next

static void on_written(IC_APID_CONNECTION *conn, IC_QUERY_ID query,
                       void *user_ref)
{
  struct slot *s = user_ref;
  struct pipeline *p = s->p;
  int32_t ret = ic_apid_query_error(conn, query);
  /* ... count it done, and a failure */
  ic_apid_conn_close_transaction(conn, s->trans);
  while (p->next < p->end) {
    ret = start_write(conn, s, p->next++);
    if (ret == 0)
      break;
    /* ... */
  }
}
user_refThe slot: its query, its transaction, and the shared state via s->p.
ErrorThe query is complete: its error is 0, or why the write failed.
CloseThe commit went with the write: the transaction is done, closed here.
NextSame query on a new transaction, sent by the next flush.

iclaustron: examples/c/pipeline.c (on_written, start_write); crates/ic_apid/src/transaction.rs (take_key_conf: the commit recorded before the callbacks; finish_query); doc/user-guide.md §8

force · adaptive send and the send pool

Without force, a send may wait a moment for company on its link

user thread Aic_apid_conn_send(conn, false) node 1 · send buffer one mutex, one writer at a time adaptive send: write now,or hold for company held at most 200 µs data node 1 per write: one receive, one execution round, one reply packet user thread Bsends a moment later B writes both, once send poolwrites at the deadline

Without force, A's requests are appended to the node's buffer, and adaptive send may hold them back for company: 200 µs at most today.

B's requests join them, and B writes both in one write: the data node pays one receive, one execution round, one reply packet.

Had none come, the send pool writes them at the deadline. A thread that sends alone teaches the algorithm not to hold at all.

With force the write happens now. A thread that goes on to poll loses nothing by leaving it off, as the pipeline example does.

iclaustron: crates/ic_apid/src/adaptive_send.rs (AdaptiveSend::decide, deadline, IC_DEFAULT_MAX_SEND_WAIT_NANOS); crates/ic_apid/src/apid_global.rs (send_words_for_generation, send_due); crates/ic_apid/src/send_pool.rs; transaction.rs (send_queries)

Cost · chapter 12's model of a batch

Why many in flight pays: a batch's fixed costs are shared

Cost per operation

d = d_op + d_batch / b

  • d_op ≈ 300 ns: define, complete, route
  • d_batch, several µs: a write and a read per node, a round, a wake-up
  • past b ≈ 200 the batch term is small

Little's law

X = depth · b / R

  • depth batches of b in flight, each R from send to done
  • deeper raises X until R grows: until a server is full

Depth adds throughput, not cost: measured client CPU per read stays near 545 ns from depth 1 to depth 4.

iclaustron: doc/rust/12-performance-model.md (The path of a batch; Operational laws: Little's law, Cost per operation); doc/rust/architecture.html §3 (the executor's loop: a write, a read and a round per node)

Measured · ic_bench runs this model

Measured asynchronous reads: 2.6 million a second on one thread

SetupReads a secondClient CPU per readWhat limits it
Mac, 1 user thread, depth 42.6 M545 nsThe user thread: about 350 ns of user time per read
Mac, 4 user threads, batch 3005.8 M495 nsCores shared with the data nodes' threads
Mac, 8 user threads, batch 3007.0 – 8.9 M512 – 565 nsOne coordinator thread of a data node at 96%
Linux, 32 CPUs, 8 user threads, 2 receive threadsup to 7.1 M—The data node's receive thread

ic_bench, primary key reads of t9: two release data nodes on the same machine, batches of 200 or 300, two in flight per thread unless stated.

iclaustron: doc/rust/architecture.html §1 (What it does, measured); doc/rust/12-performance-model.md (Calibration); tools/ic_bench/src/main.rs (run_thread: depth batches in flight)

Errors · in the asynchronous model

Errors in flight: on the query and its transaction, retried whole

ret = ic_apid_transaction_error(conn, trans);
if (ret == 0)
  ret = ic_apid_query_error(conn, query);
/* ... */
if (!ic_error_is_temporary(ret))
  ex_fail("the update", ret);

A node failure reaches the callbacks too: a poll ends what it aborted with a temporary error, and what had committed comes back committed (module 02).

QueryIts error. By default a failed write rolls its transaction back.
TransactionIts error, and its state: committed or rolled back.
TemporaryEnds the transaction: start it again, after a pause that grows.
4012No reply, outcome unknown: find out before retrying a write.

iclaustron: examples/c/retry.c (wait_for, main); crates/ic_apid/src/transaction.rs (end_transaction, fail_lost_transactions); doc/user-guide.md §9, §10; doc/rust/05-ndb-protocol.md §6.5

The limit · one thread's work

The limit: new work comes only from this thread or its callbacks

its own loopdefines, sends, polls its callbacksrun inside its poll user thread · connection one thread at a timepoll waits for replies only HTTP handleranother thread gRPC workeranother thread no way in

poll knows of no hand-off from another thread. Work that arrives on other threads goes through an executor instead: module 06.

iclaustron: doc/rust/architecture.html §3 (the limitation; an executor with two inputs); include/iclaustron/ic_apid.h (Threads: a connection is used by one thread at a time); doc/user-guide.md §5

Compared · the three models on a connection

Three models on a connection: what each costs and when to use it

SynchronousBatched synchronousAsynchronous
The caller writesone call per query, which waitsqueries on transactions, commit, flush until each is donequeries with a callback or for the executed list; send; poll
In flight on a threadone queryone batch, drained before the nextseveral batches, refilled as they come back
Reads a second198 000 from 64 threads1.17 – 1.35 M, one thread, depth 12.55 – 2.61 M, one thread, depth 4
Client CPU per read18.3 µsabout 545 nsabout 545 ns
Use it fortools, tests, low-rate servicesa thread that needs every result before it goes onloaders, benchmarks, servers built for it

Measured on a Mac (M5 Pro), two release data nodes on the same machine, t9 reads by primary key, batches of 200; synchronous: one call at a time per thread.

iclaustron: doc/rust/12-performance-model.md (Calibration; Executors: 64 threads, a connection each); doc/rust/architecture.html §4 (the comparison); tools/ic_bench/src/main.rs (--depth 1, the plain form)

Recap · module 05

Asynchronous key operations: what to take away

  • Define is local; send writes it all, one write per data node; poll completes it.
  • Receive threads only route; the user thread decodes, completes and calls back.
  • Completions go to the executed list, or to a callback inside the poll.
  • A callback may define, commit, close and send: the pipeline refills itself.
  • Many in flight share each batch's costs: 2.6 M reads a second on one thread.
  • New work comes only from this thread; other threads' work needs an executor.

Next · 06 · Key operations through an executor

iClaustron training for C developers · 8 modules · you are at 05 · this module for Rust developers

Next · 06 · Through an executor