Key operations, asynchronous

The calls · methods of ApidConnection

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

impl ApidConnection {
  pub fn send_queries(&mut self, force: bool) -> Result<(), IcError>

  pub fn poll(&mut self, wait_ms: u32) -> usize

  pub fn flush(&mut self, wait_ms: u32, force: bool)
    -> Result<usize, IcError>
}

iclaustron: crates/ic_apid/src/transaction.rs (ApidConnection::send_queries, flush); crates/ic_apid/src/apid_conn.rs (ApidConnection::poll); doc/rust/04-api-design.md (Sending, polling and completion); doc/user-guide.md §5

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

The pipeline example: statics for what main and callbacks share

/// The table, set once before the first write.
static TABLE: OnceLock<T1> = OnceLock::new();
/// The next id to write, and one past the last.
static NEXT: AtomicU32 = AtomicU32::new(0);
static END: AtomicU32 = AtomicU32::new(0);
/// Writes completed, and how many of them failed, with the first error.
static DONE: AtomicU32 = AtomicU32::new(0);
static FAILED: AtomicU32 = AtomicU32::new(0);
static FIRST_ERROR: AtomicI32 = AtomicI32::new(0);

let _ = TABLE.set(bound);
// ...
while i < width {
  let qid = conn.create_query(&t1.table, &t1.rec, &t1.rec)?;
  queries.push(qid);
  let id = NEXT.fetch_add(1, Ordering::Relaxed);
  start_write(&mut conn, qid, id)?;
  i += 1;
}
while DONE.load(Ordering::Relaxed) < rows {
  conn.flush(10, false)?;
}

A callback is a plain function, so what it shares with the main loop is in statics. run starts a write per query, then only flushes.

iclaustron: examples/rust/pipeline.rs (module doc, the statics, run); 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

match conn.query_mut(qid) {
  Some(query) => {
    t1.fill(&t1.rec, query.key_row_mut(), &values)?;
    t1.fill(&t1.rec, query.attr_row_mut(), &values)?;
  }
  // ...
}
// ... hint = conn.hint_for_key(&t1.table, &t1.rec, query.key_row())
let trans = conn.start_transaction(hint, Some(&t1.table))?;
let args = WriteKeyArgs {
  kind: WriteKind::Write,
  callback: Some(on_written),
  ..WriteKeyArgs::default()
};
conn.write_key(qid, trans, &args)?;
conn.commit_transaction(trans)

iclaustron: examples/rust/pipeline.rs (start_write); crates/ic_apid/src/query.rs (ApidQuery::key_row_mut, WriteKeyArgs); crates/ic_apid/src/transaction.rs (start_transaction, write_key, define, commit_transaction); crates/ic_apid/src/apid_conn.rs (coord_record)

Step 2 · send · the first half of 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 DONE.load(Ordering::Relaxed) < rows {
  conn.flush(10, false)?;
}

iclaustron: examples/rust/pipeline.rs (run); crates/ic_apid/src/transaction.rs (flush, 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 conn.flush

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

conn.flush(10, false)?;   // send, then poll up to 10 ms

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

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

pub type QueryCallback = fn(&mut ApidConnection, QueryId, usize);

pub struct ReadKeyArgs {   // and WriteKeyArgs
  // ... kind, abort_option
  pub user_ref: usize,   // given back with the completed query
  pub callback: Option<QueryCallback>,   // None: the executed list
  // ... binding
}

// ApidConnection, and ApidQuery from conn.query(id):
pub fn get_next_executed_query(&mut self) -> Option<QueryId>
pub fn user_ref(&self) -> usize

iclaustron: crates/ic_apid/src/query.rs (QueryCallback, ReadKeyArgs, WriteKeyArgs, ApidQuery::user_ref); crates/ic_apid/src/transaction.rs (get_next_executed_query, finish_query)

Callbacks in Rust · QueryCallback, in query.rs

A Rust callback is a plain function, not a closure

The pipeline example sets no user_ref: the query's id is its slot. Its statics are a OnceLock for the table and atomics for the counters: no unsafe.

iclaustron: crates/ic_apid/src/query.rs (QueryCallback, ReadKeyArgs, WriteKeyArgs); crates/ic_capi/src/lib.rs (trampoline); examples/rust/pipeline.rs (module doc, the statics, on_written); doc/rust/07-style-guide.md (Closures); doc/user-guide.md §8

Callbacks · the rules

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

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.rs

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

fn on_written(conn: &mut ApidConnection, qid: QueryId, _user_ref: usize) {
  let (error, trans) = match conn.query(qid) {
    Some(query) => (query.error(), query.transaction()),
    None => return,
  };
  // ... counted in DONE, and a failure in FAILED, FIRST_ERROR
  if let Some(trans) = trans {
    let _ = conn.close_transaction(trans);
  }
  loop {
    let id = NEXT.fetch_add(1, Ordering::Relaxed);
    // ... none left past END: return
    match start_write(conn, qid, id) {
      Ok(()) => return,
      // ...
    }
  }
}

iclaustron: examples/rust/pipeline.rs (on_written, start_write); crates/ic_apid/src/query.rs (ApidQuery::error, transaction); 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

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

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

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

// committed(): the transaction's error, else the query's, else 4012
match committed(&mut conn, trans, updater) {
  // ... Ok(()): committed
  Err(e) if e.is_temporary() && attempt < ATTEMPTS => {
    // ... a new transaction, after a pause
  }
  Err(e) => return Err(e),
}

iclaustron: examples/rust/retry.rs (run); examples/rust/lib.rs (committed, finish); crates/ic_port/src/err.rs (IcError::is_temporary); 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

poll knows of no hand-off from another thread. Work that arrives on other threads goes through an executor instead, handed over by Submitter::submit, with a callback for the result: module 06.

iclaustron: doc/rust/architecture.html §3 (the limitation; an executor with two inputs); crates/ic_apid/src/apid_conn.rs (ApidConnection: one thread uses it); crates/ic_apid/src/executor.rs (Submitter::submit, Done::Callback); 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

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 Rust developers · 9 modules · you are at 05 · this module for C developers

Next · 06 · Through an executor