iClaustron training for Rust developers · module 7 of 9 · 20 min

Scans in the five programming models

How a scan runs on the threads and objects of modules 02 and 03, and how to read one row by row, a batch at a time, asynchronously and through an executor's reader.

How a scan runs · the objects

A scan is made once and run many times, each run with its receivers

connectionone per user thread scanScanIda table and a recordits own rowconn.create_scan:once runIdle, Starting,Running, Closing,Doneconn.start_scan:as often as you like coordinator recordits own, from a pool per node,topped up ahead one receiver per fragment scanned at once R1 R2 R3 its batch its batch its batch

conn.create_scan makes a scan once in its connection: a table, a record, and the scan's own row, which each row read is unpacked into.

conn.start_scan starts a run. It takes a coordinator record from a pool per node; with none free it asks, and starts in a later poll.

A run has one receiver per fragment scanned at once, each bringing a batch of rows. ScanBatchId names a batch by an id of its own, which names nothing once the batch is handed back.

iclaustron: crates/ic_apid/src/scan.rs (module note; ApidConnection::create_scan, start_scan; ScanId, ScanState, ScanBatchId) · apid_conn.rs (take_scan_record, ask_scan_record, top_up_scan_pool)

How a scan runs · what each run asks for

ScanArgs: what a run asks for

FieldDefaultWhat it does
lockScanLock::CommittedShared, Exclusive: rows locked until their batch is handed back
parallelism0: every fragmentfragments scanned at once, one receiver each
batch_rows, batch_bytes0: from the configurationa receiver's batch; at most 992 rows locking, 4 000 committed
partitionNoneSome(id): scan this partition only
timeout_ms0: no limitclose the run with IC_ERROR_TIMEOUT when it passes
callback, user_refNone, 0each complete batch to the callback, not the ready list
read_ahead, continuousfalsecommitted reads: read ahead at the executor or in the data nodes
trans, ordered, descendingNone, falsetaking rows over, index order: module 08

Name what you want and take the rest with ..ScanArgs::default(); start_scan borrows the arguments, so one ScanArgs serves every run.

iclaustron: crates/ic_apid/src/scan.rs (ScanArgs, ScanLock, ScanCallback; ApidConnection::start_scan; batch_size) · crates/ic_signals/src/scans.rs (IC_SCAN_MAX_LOCKED_ROWS, IC_SCAN_MAX_COMMITTED_ROWS) · examples/rust/scan.rs (run)

How a scan runs · the signals

A run's signals: SCAN_REQ, rows, SCAN_CONF, SCAN_MORE_REQ

application threadthe application's calls user threadsame thread: connection receive threadroutes signals data nodescoordinator, fragments start_scan recordthe coordinator'sfrom the pool SCAN_REQ coordinatorreceivers R1 R2 R3 fragments1, 2, 3 at once ROW_DATA_REP receiversrows copied inR1, R2, R3 coordinatorSCAN_CONF SCAN_CONF batch completeits entry, andall its words delivereduntil handed back(locks held) rows read handed back R1 asked againwith the nextsend or poll SCAN_MORE_REQ fragment 1its next batch coordinatornull record: done SCAN_CONF run donerecord backto its pool Ok(false)

A run takes a coordinator record from a pool per node, then sends SCAN_REQ naming one receiver per fragment scanned at once.

The coordinator scans the fragments at once. Each row comes as a ROW_DATA_REP naming its receiver, which copies it into its batch.

SCAN_CONF gives each receiver's rows and words. A batch is complete once its entry and all its words have come, in either order.

The complete batch goes to the application: next_row, the ready list or a callback. A locking scan's rows stay locked meanwhile.

Handed back, its receiver is named in the scan's next SCAN_MORE_REQ, which goes with the next send or poll; that fragment goes on.

An entry naming the null record ends its receiver. With none left the run is done, and the record goes back to its pool.

iclaustron: doc/rust/02-architecture.md (Data flow for a scan) · doc/rust/05-ndb-protocol.md §7.1, §7.2 · crates/ic_apid/src/scan.rs (send_scan_req, take_scan_row, take_entry, check_batch, queue_more, coord_finished)

How a scan runs · flow control

The application sets the pace: a held batch holds its receiver

R1 asked batch asked batch asked batch asked batch R2 asked batch asked R3 asked batch asked batch asked batch asked batch held: R2 asks for nothing, a locking scan's rows stay locked asked: the batch is on its way with the application, until handed back

Each receiver has one batch out. It asks for the next only once the application hands that one back.

A batch held stalls only its own receiver: R2 asks for nothing meanwhile, and a shared or exclusive scan keeps those rows locked.

So memory stays within the receivers times the batch size. A batch held too long ends the scan with 296, IC_NDB_ERROR_SCAN_IDLE.

iclaustron: crates/ic_apid/src/scan.rs (module note; ScanLock::Shared, IC_NDB_ERROR_SCAN_IDLE, release_scan_batch, check_batch) · doc/rust/04-api-design.md (Scans: Batches, Why so)

Synchronous · the code

Synchronous scans: a run read row by row with next_row

let scan = conn.create_scan(&t1.table, &t1.rec)?;

// Row by row: committed reads, with no lock, of every fragment at
// once. next_row waits for batches and hands each back as it goes.
conn.start_scan(scan, &ScanArgs::default())?;
let mut rows: u64 = 0;
while conn.next_row(scan)? {
  rows += 1;
  if let Some(val) = get_i32(&t1.rec, conn.scan_row(scan), t1.val) {
    // ...
  }
}
  • The defaults: a committed read of every fragment. next_row hands each batch back once its rows are read, and polls while none is complete.
  • conn.scan_row(scan) is the scan's own row, lent by the connection and overwritten by each row; next_row_into fills yours, a copy less.
  • Ok(false) at the end; a failed run gives its Err and drops the rows not read.

iclaustron: examples/rust/scan.rs (run, row by row; shortened) · crates/ic_apid/src/scan.rs (ApidConnection::create_scan, start_scan, next_row, next_row_into, scan_row)

Synchronous · threads and objects

Synchronous scans on the threads: next_row polls as it waits

application threadthe application's calls user threadsame thread: connection receive threadroutes signals data nodescoordinator, fragments next_row → row R1's batchrows unpackedone per call next_row R1 used uphanded backR1 to ask again no batch yetpoll: sends whatis due, waits SCAN_MORE_REQ fragment 1its next batch repliesfor all its work to its inbox the pollruns each replyof the connection served toocallbacks run,B → ready list R2's batchthis scan'soldest, taken next_row → row

Each conn.next_row unpacks the next row of the batch it holds into the scan's row: no wait while the batch lasts.

When that batch is used up, next_row hands it back itself, so that its receiver, R1, asks for more.

With no other batch of this scan complete, next_row polls. The poll first sends what is due, R1's SCAN_MORE_REQ, then waits.

The receive thread routes every reply for this thread to its inbox, and the poll runs them all: this scan's and the connection's other work.

So other queries complete and their callbacks run, and other scans' batches go on the ready list, there to wait for their reader.

next_row takes this scan's oldest complete batch, R2's, and returns its first row.

iclaustron: crates/ic_apid/src/scan.rs (next_row_with, wait_scan_batch, wait_for_scan, take_ready_batch_of, deliver) · apid_conn.rs (poll_with_requests) · doc/rust/02-architecture.md (Reading a scan)

Batched synchronous · the code

Batched synchronous scans: a batch at a time with next_batch

let args = ScanArgs {
  batch_rows: 16,
  ..ScanArgs::default()
};
conn.start_scan(scan, &args)?;
while let Some(batch) = conn.next_batch(scan)? {
  batches += 1;
  while conn.next_scan_row(batch)? {
    again += 1;
  }
  conn.release_scan_batch(batch)?;
}
  • next_batch waits as next_row does, and gives a whole batch; scan_batch_rows says how many rows.
  • next_scan_row unpacks the next row into the scan's row, or next_scan_row_into yours.
  • The application holds the batch, a Copy id, as long as it likes, several at once if it wants; its receiver waits for release_scan_batch.

iclaustron: examples/rust/scan.rs (run, a batch at a time; shortened) · crates/ic_apid/src/scan.rs (next_batch, scan_batch_rows, next_scan_row, next_scan_row_into, release_scan_batch; ScanBatchId)

Batched synchronous · threads and objects

Several scans on one connection, each read in turn

application threadthe application's calls user threadsame thread: connection receive threadroutes signals data nodescoordinator, fragments start A, start B two runsa record each,receivers each SCAN_REQ ×2 coordinatorsA's and B's next_batch(A) the pollwaits for A,takes all replies rows, SCAN_CONF fragmentsof A and of B completeA's → next_batchB's → ready list A's rows read release A A asks againB still waits SCAN_MORE_REQ A A's fragmentgoes on next_batch(B) B's batchon the list:no wait, no poll release B B asks againwith the nextsend or poll SCAN_MORE_REQ B B's fragmentgoes on

Two scans started on one connection run at once, each run with a coordinator record of its own.

next_batch on scan A polls, and the poll completes whatever comes: batches of A and of B alike.

It takes only A's batch. B's goes on the ready list and waits there; B's receiver asks for nothing meanwhile.

A's batch read and released: A's receiver is asked again with the next send or poll.

next_batch on B finds a batch already complete, and returns it without a wait.

Read every scan in turn: a batch left waiting holds its receiver and a locking scan's rows, and held too long ends the scan with 296.

iclaustron: crates/ic_apid/src/scan.rs (next_batch, wait_scan_batch, take_ready_batch_of, release_scan_batch) · doc/rust/04-api-design.md (Consuming scans) · doc/rust/06-phases.md §8.4

Asynchronous · the code

Asynchronous scans: after each poll, batches from the ready list

conn.start_scan(scan, &ScanArgs::default())?;
conn.send_queries(false)?;
while !conn.scan_done(scan) {
  conn.poll(100);
  while let Some(batch) = conn.get_next_scan_batch() {
    while conn.next_scan_row(batch)? {
      let row = conn.scan_row(scan);
    }
    conn.release_scan_batch(batch)?;
  }
}
  • The request goes with the next send or poll; each poll completes batches.
  • get_next_scan_batch takes the batch completed longest ago, of any scan with no callback; conn.batch_scan(batch) names its scan.
  • Read it, release it: that receiver goes on. Queries and other scans complete in the same polls.

iclaustron: doc/rust/04-api-design.md (Scans, the Rust example; lock and comments left out) · crates/ic_apid/src/scan.rs (get_next_scan_batch, batch_scan, scan_done) · apid_conn.rs (poll) · transaction.rs (send_queries)

Asynchronous · a callback per batch

A scan callback: each batch handed over inside the poll

pub type ScanCallback = fn(&mut ApidConnection, ScanBatchId, usize);

/// A scan's callback that hands each batch back unread.
fn hand_back_at_once(conn: &mut ApidConnection, batch: ScanBatchId, _: usize) {
  conn.release_scan_batch(batch).expect("hand back");
}

let called = ScanArgs {
  callback: Some(hand_back_at_once),
  ..ScanArgs::default()
};
conn.start_scan(first, &called).expect("start with a callback");
  • Each complete batch of the run goes to the callback, on the thread that polls, inside the poll; never to the ready list.
  • It may read the batch and release it there, or keep it and release it later.
  • next_row and next_batch refuse a scan with a callback (IC_ERROR_NOT_SUPPORTED); inside a callback a poll does nothing.
  • A plain fn, so a closure that captures will not do: what it needs comes through user_ref, a usize.

iclaustron: crates/ic_apid/src/scan.rs (ScanCallback, ScanArgs, deliver, wait_scan_batch) · crates/ic_apid/tests/integration.rs (hand_back_at_once; scan_row_by_row_beside_another_scan; shortened) · apid_conn.rs (poll_with_requests)

Asynchronous · threads and objects

Asynchronous scans on the threads: one poll serves them all

application threadthe application's calls user threadsame thread: connection receive threadroutes signals data nodescoordinator, fragments start_scan ×2 A: callbackB: ready list+ key queries send_queries one writeper data node SCAN_REQ ×2, KEY_OP_REQ data nodesscan, read keys poll repliesrows, entries to its inbox the pollruns them all A's batchto A's callback A's callback B's batchto the ready listqueries complete get_next_scan_batch B's batch readand released:B asks again

One thread starts scan A with a callback and scan B without, and defines key queries besides.

One conn.send_queries sends everything defined, one write per data node, the scans' requests with it.

conn.poll waits once for whatever comes; the receive thread has routed it all to this thread's inbox.

Inside the poll, A's complete batch is handed to A's callback, on this thread.

B's batch goes on the ready list, and the queries complete, their callbacks run too.

After the poll, get_next_scan_batch takes B's batch; read and released, B's receiver is asked again with the next send or poll.

iclaustron: crates/ic_apid/src/scan.rs (deliver, get_next_scan_batch, release_scan_batch) · crates/ic_apid/src/transaction.rs (send_queries) · doc/rust/06-phases.md §8.4 (Asynchronous)

Synchronous through an executor · the code

Scans through an executor: a shape, a reader, a submitter

// made once, shared
let shape = ScanShape::new(&table, &rec)?;
// Each thread its own submitter, the shape shared.
let mine = submitter.clone();
let shared = Arc::clone(&shape);
// ... both moved into a thread with no connection of its own:
let mut reader = ScanReader::new(&shared);
mine.scan(&mut reader, &ScanArgs::default())?;
while reader.next_row()? {
  let row = reader.row();
}
  • submitter.scan returns once the run is queued at the executor; reader.next_row waits for its rows.
  • Refused (IC_ERROR_NOT_SUPPORTED): a transaction or a callback in the arguments, and a call on the executor's own thread.
  • A reader runs again and again; is_done and error say how its run ended.

iclaustron: examples/rust/scan_executor.rs (run, read_quarter; ranges and order left out) · doc/rust/04-api-design.md (Consuming scans: ScanShape::new, the reader's loop) · crates/ic_apid/src/executor.rs (Submitter::scan) · scan_reader.rs (ScanReader)

Synchronous through an executor · threads and objects

The executor's feed: batches move to the reader without a copy

application threadreader, no connection executor threadits connection receive threadroutes signals data nodescoordinator, fragments submitter.scan inboxrun queued;the call returns next_row: sleeps scan of the shapekept on itsconnection SCAN_REQ coordinatorfragments at once fragmentsrows, SCAN_CONF to its inbox batch completeat the executor to the feedbuffers swapped,no copy wakes and reads batch handed back nudgeon its own list,one per scan nudge servedrelease_scan_batchon its connection SCAN_MORE_REQ fragmentits next batch

submitter.scan puts the run in the executor's inbox and returns. Scans wait in a queue of their own, 64 running at most by default.

The executor runs it as a scan of the shape on its own connection, kept from run to run. reader.next_row sleeps meanwhile.

Rows and SCAN_CONF come to the executor's inbox, and its poll completes the batch.

The batch moves to the reader's feed without a copy: the receiver's buffers swap places with an empty pair. The reader wakes and reads.

Moving on, the reader hands the batch back on the feed and nudges the executor: a word on a list of its own, one waiting per scan at most.

Served the nudge, the executor hands the batch back to its receiver, which asks for more. A slow reader stalls only its own scan.

iclaustron: crates/ic_apid/src/executor.rs (run, admit_scan, take_scan_batches, serve_nudge) · scan_reader.rs (module note; hand_back) · thread_conn.rs (Nudges) · scan.rs (swap_scan_batch)

Batched synchronous through an executor · threads and objects

Batched through an executor: several readers on one thread

application threadreaders A and B executor threadits connection receive threadroutes signals data nodescoordinator, fragments submitter.scan ×2 inboxboth runs queued,both calls return two scansof their shapesrun at once SCAN_REQ ×2 coordinatorsA's and B's fragmentsrows of A and B to its inbox batches movedA's to A's feedB's to B's feed reads A A's nudgesA's receiversgo on SCAN_MORE_REQ A A's fragmentsgo on B's batches wait B's receiverswait: one batcheach, with B reads B B's nudgesB's receiversgo on

One thread starts two readers. Each submitter.scan returns once its run is queued, so both run at once.

The executor runs both on its connection, each as a scan of its shape.

Each complete batch moves to its own reader's feed.

The thread reads A. Each batch it finishes goes back with a nudge, and A's receivers go on.

B's batches wait in B's feed, one per receiver, and B's receivers wait with them until B is read.

Then B is read the same way. In the scan_ tests, while one reader held a batch, five others on the same executor each read the whole table: 7 ms together.

iclaustron: crates/ic_apid/src/executor.rs (Submitter::scan, take_scan_batches) · scan_reader.rs (begin_run, next_row_with) · doc/rust/06-phases.md §8.4 (the scan_ integration group, 7 ms)

Batched synchronous through an executor · reading ahead

read_ahead: the next batch comes while the reader reads

default reads batch k waits reads k+1 waits reads k+2 waits k+1 comes k+2 comes k+3 comes read_ahead reads batch k reads k+1 reads k+2 reads k+3 reads k+4 k+1 comes k+2 comes k+3 comes k+4 comes k+5 comes

By default a receiver asks for its next batch only when the reader hands the last one back, so the reader waits for that round trip.

With read_ahead the receiver goes on as soon as its batch has moved to the reader: two batches out at most, twice the memory.

Committed reads through an executor only: refused with a lock, with continuous, and on a connection, where you release batches yourself.

iclaustron: crates/ic_apid/src/executor.rs (take_scan_batches, serve_nudge, AheadReceiver, IC_READ_AHEAD_BATCHES, Submitter::scan) · scan.rs (ScanArgs::read_ahead, continuous; start_scan) · doc/user-guide.md §5 (Scans)

All five models · ownership and threads

Ownership in scans: Copy ids on a connection, a reader that owns

  • ScanId and ScanBatchId are Copy ids that borrow nothing: the scan and its batches stay in the connection, which you go on using while you hold them.
  • conn.scan_row(scan) and reader.row() lend the row until the next call that takes &mut, such as next_row; next_row_into unpacks into a buffer of yours instead.
  • ScanShape::new gives an Arc<ScanShape> to clone into each thread; a Submitter is cloned, one per thread, and moved in.
  • A ScanReader owns its row and batches and holds the shape by an Arc: no lifetime, and Send, so it may move to another thread between calls; reading takes &mut self.
  • Not built: a Stream over the reader for async code, kept for later with the Rust futures over executors.

iclaustron: crates/ic_apid/src/scan.rs (ScanId, ScanBatchId; ApidConnection::scan_row, next_row, next_row_into) · scan_reader.rs (ScanShape::new; ScanReader, its fields, row, next_row) · executor.rs (Submitter, impl Clone) · examples/rust/scan_executor.rs (run) · doc/rust/04-api-design.md (Consuming scans; Kept for later)

All five models · closing early

Closing a run early: the coordinator stops at once

application threadthe application's calls user threadsame thread: connection receive threadroutes signals data nodescoordinator, fragments close_scan closingbatches heldlet go of stop at onceno wait for thebatches asked for SCAN_MORE_REQ stop coordinatorstops the scan late rowsalready on the way ROW_DATA_REP droppedreceiver id stale,counted as late coordinatorend of data SCAN_CONF run donerecord backto its pool can run again

conn.close_scan ends a run before it reads to the end; the batches the application holds are let go of.

The stop goes at once in a SCAN_MORE_REQ, without waiting for the batches asked for; pending takeovers go first (module 08).

Rows still on their way name a receiver of no live run: dropped, and counted as late.

The bare end of data confirms the close, in a later poll: the record goes back to its pool, and the scan can run again.

Through an executor, reader.close() does the same and waits for the executor to finish; dropping the reader does not wait.

iclaustron: crates/ic_apid/src/scan.rs (ApidConnection::close_scan, hold_batches, late_scan_signal, take_scan_conf) · scan_reader.rs (ScanReader::close; impl Drop for ScanReader)

All five models · when a run fails

Scan errors are temporary: start the run again

CodeNameWhat happened
296IC_NDB_ERROR_SCAN_IDLEthe run was left too long without asking for more: a batch not handed back
274IC_NDB_ERROR_SCAN_LOCK_TIMEOUTthe scan waited too long for a lock, probably in a deadlock
266no constant in ic_apida lock wait timed out, probably in a deadlock
4028IC_NDB_ERROR_SCAN_NODE_FAILUREthe coordinator's node failed, and the run's record with it
4012IC_NDB_ERROR_NO_REPLYthe coordinator, its link up, sent nothing for six minutes

All temporary (e.is_temporary()); the constants are in ic_apid::scan, 4012's in ic_apid::transaction. Nothing resumes a run: start it again, and its rows come again from the start.

iclaustron: crates/ic_apid/src/scan.rs (IC_NDB_ERROR_SCAN_IDLE, _SCAN_LOCK_TIMEOUT, _SCAN_NODE_FAILURE; fail_lost_scans, expire_scans, IC_SCAN_ANSWER_WAIT_MS) · transaction.rs (IC_NDB_ERROR_NO_REPLY) · crates/ic_port/src/err.rs (IcError::is_temporary) · ndb_err.rs (classes) · doc/user-guide.md §5, §9

Summary · scans in the five programming models

Scans in the five models, side by side

ModelCallsWho pollsWho holds a batchMemoryUse it for
Synchronousnext_rownext_rowthe librarya batch per receiverone scan at a time, simplest
Batched synchronousnext_batch, next_scan_row, release_scan_batchnext_batchthe applicationa batch per receiverwhole batches; several scans in turn
Asynchronouspoll, then get_next_scan_batch or a callbackthe applicationthe applicationa batch per receiverscans and queries in flight
Synchronous through an executorsubmitter.scan, reader.next_rowthe executorthe readera batch per receiverthreads with no connection
Batched synchronous through an executorseveral readers, read_aheadthe executorthe readerstwo per receiver with read_aheadseveral scans per thread

Calls are methods of the connection, the submitter and the reader. Every model runs the same signals.

iclaustron: doc/rust/04-api-design.md (Scans; Consuming scans) · crates/ic_apid/src/scan.rs, scan_reader.rs, executor.rs

Recap · module 07

Scans in five models: what to take away

  • A scan is made once and run many times; a run takes a pooled coordinator record and one receiver per fragment.
  • A receiver asks for its next batch only once the last is handed back: the application sets the pace, and locks last until then.
  • On a connection: next_row row by row, next_batch a batch at a time, or batches after a poll, from the ready list or a callback.
  • Through an executor: a shape, a reader, submitter.scan; batches move without a copy, and a slow reader stalls only its own scan.
  • read_ahead lets a committed read fetch ahead at twice the memory.
  • Scan errors are temporary: start the run again.

Next · 08 · Advanced scans

iClaustron training for Rust developers · 9 modules · you are at 07 · this module for C developers

Next · 08 · Advanced scans