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

Object architecture

The objects a program makes, the ones the library keeps for it, which thread owns each, and how a reply from a data node finds its object.

The map · ownership as nesting

The object map: global, connection, executor

The global is shared by every thread of the process: one per cluster, with its per-node state, its thread table, its dictionary cache and its threads.

Each user thread makes a connection from the global: its own block number, inbox and objects, used by one thread at a time, with no lock on its objects.

An executor is a thread of the library's with a connection of its own; application threads reach it through submitters.

Definitions and programs come in an Arc, records are cloned where kept: handles made once, used by any thread, which may outlive the connection that bound them.

iclaustron: crates/ic_apid/src/apid_global.rs (ApidShared, ApidGlobal::start_with_options); apid_conn.rs (ApidConnection); executor.rs (Runner, Submitter); dict_cache.rs (module notes: a TableDef behind an Arc); record.rs (Record); program.rs (Program)

Global · ApidGlobal

The global: one per cluster per process, shared by every thread

typeApidGlobal one owner, lent as &ApidGlobal

makeApidGlobal::connect

useglobal.wait_for_all_startedglobal.create_connectionExecutor::start

dropglobal.stop() or its drop, after its connections and executors

One per cluster: a program that talks to two clusters starts two.
One NodeShared per data node, fixed at start. Each node is dealt to one receive thread; never more receive threads than nodes.

iclaustron: crates/ic_apid/src/apid_global.rs (NodeShared, Sender, ApidShared, ApidGlobal::connect, start_with_options, create_connection, stop, Drop); thread_conn.rs (ThreadTable); auto_increment.rs (CounterSlot); executor.rs (Executor::start); doc/user-guide.md §2

Global · made first, dropped last

Lifetime: the global is made first and dropped last

let global = ApidGlobal::connect("localhost:1186", Some("my_app"),
                                 &GlobalOptions::default())?;
global.wait_for_all_started(30_000);
let mut conn = global.create_connection()?;
let table = conn.table_bind("ic_ex", "t1")?;
let rec = Record::default_for(&table, None)?;
let executor =
  Executor::start(&global, &ExecutorOptions::default())?;
let submitter = executor.submitter();
// ...

Nothing to free: each object goes when it is dropped. Locals drop in reverse order, so these go submitter first, global last.
Dropping the global, or stop(), stops every thread of the library and waits for each. A connection holds no borrow of it: keep the global while its connections are used.
Dropping a connection first waits for the answers to the coordinator records it asked for: well under a millisecond as a rule, 5 s at most. A transaction left open does not commit.
Dropping an executor stops it and joins its thread. An Arc<TableDef> or a Record may outlive the connection that bound it.

iclaustron: doc/user-guide.md §2; crates/ic_apid/src/lib.rs (crate docs: connect, bind, record); examples/rust/scan_executor.rs (run: the executor and its submitter); examples/rust/lib.rs (connect: keep the global); crates/ic_apid/src/apid_global.rs (stop, Drop); apid_conn.rs (Drop, settle_allocates); executor.rs (Executor::stop, Drop)

Connection · ApidConnection

A connection is one user thread's view of the cluster

typeApidConnection Send, one thread at a time

makeglobal.create_connection()

useits methods, and through them the queries, transactions and scans its ids name

dropits queries, transactions and scans go with it

It may move to another thread between calls, as an executor's moves into its thread; its calls take &mut self, so one thread uses it at a time.
At most 4 711 per process, executors' own included (7139).

iclaustron: crates/ic_apid/src/apid_conn.rs (ApidConnection, Drop, CoordRecord, Expectations); thread_conn.rs (ThreadConnection::block_number, ThreadTable::allocate); executor.rs (Executor::start: its connection moved into its thread); tools/ic_bench/src/main.rs (bench); doc/user-guide.md §2

Connection · ids, not references

Queries, transactions and scans are named by generational ids

A QueryId is a place in the connection's object map with a generation over it: a Copy word, never 0, that borrows nothing. The query stays in the connection.

A freed query's place is reused under the next generation, and a place whose generation would wrap is retired: conn.query(old) gives None, a call naming it is refused, and a late reply naming it is dropped.

iclaustron: crates/ic_util/src/ptr_array.rs (PtrId, wire_word, PtrArray::insert, get, remove, id_of_wire); crates/ic_apid/src/query.rs (QueryId, TransId); scan.rs (ScanId, ScanBatchId, batch_scan, IC_RECEIVER_TAG, fits_below_tag); transaction.rs (query, query_mut, free_query)

Handles · table and index definitions

Table and index definitions are bound once and shared through a cache

typeArc<TableDef>Arc<IndexDef> any thread

makeconn.table_bindconn.index_bindsubmitter.table_bind

usetable.field_idtable.field gives &ColumnPropsindex.table()

dropthe last Arc frees it; conn.table_unbind lets the connection's go

The Arc is the count: a definition the cache lets go of lives on while a query, record or shape holds it.
An IndexDef holds the version of its table it was bound for.

The first bind fetches the table's description from a data node, once for every connection of the global, and keeps it in the cache by name.

Every later bind, from any thread, finds it there and shares the same definition: no round trip.

When a data node says the table changed, or a query fails with a schema error, the definition is let go of: bind again and retry.

iclaustron: crates/ic_apid/src/dict_cache.rs (module notes, DictCache, TableDef, IndexDef); apid_conn.rs (table_bind, index_bind, table_unbind, table_changed); rec_thread.rs (TABLE_CHANGED_REP); transaction.rs (schema_error); executor.rs (Submitter::table_bind); doc/user-guide.md §3

Handles · records

A record lays a row out over a byte slice: offset and null bit per field

// id at 0; val at 4, null bit 0; code at 8; name at 12, its length
// byte and then 32 bytes, null bit 1; the null bits in byte 45
let spec = [
  FieldSpec::not_null(id_of_id, 0),
  FieldSpec::nullable(id_of_val, 4, 45, 0),
  FieldSpec::not_null(id_of_code, 8),
  FieldSpec::nullable(id_of_name, 12, 45, 1),
];
let rec = Record::new(&table, &spec, 48)?;
let mut row = vec![0u8; rec.row_size() as usize];

typeRecordFieldSpec

makeRecord::newRecord::default_for

userec.fieldrec.position_ofrec.row_sizerec.set_nullrec.is_null

dropa Clone holding its table's Arc; a query or shape keeps its own clone

Record::default_for(&table, None) lays the row out for you; rec.field says where each field went.
Checked once, when made: every field there, nothing overlapping. Any thread may use it.

iclaustron: doc/user-guide.md §3 (the Rust record); crates/ic_apid/src/lib.rs (crate docs: the row); record.rs (FieldSpec, Record::new, default_for, field, position_of, module notes); query.rs (ApidQuery::new: its own clones)

Handles · values in a row

A row holds each value as the data nodes keep it: nothing is converted

Type	In the row
TINYINT, SMALLINT, MEDIUMINT, INT, BIGINT	1, 2, 3, 4, 8 bytes, little-endian, signed or not
FLOAT, DOUBLE	4, 8 bytes, IEEE, little-endian
CHAR(n), BINARY(n)	the field's bytes, padded: CHAR with the character set's spaces, BINARY with zeros
VARCHAR, VARBINARY	one length byte up to 255 bytes, else two, little-endian; then the bytes
DECIMAL	MySQL's packed binary decimal
DATE, TIME, DATETIME, TIMESTAMP, YEAR	as the data nodes keep them, fractional forms included
BIT(n)	n bits, in 32-bit words
NULL	a null bit, set for NULL; a NOT NULL field has none

A field takes the room of its longest value with its length bytes: VARCHAR(32) is 33 bytes in latin1, 129 in utf8mb4. Blob columns stay out of records.

iclaustron: doc/user-guide.md §3 (How values are kept; Blob columns); crates/ic_apid/src/record.rs (module notes)

Connection · queries

A query is made once, owns its key and attribute rows, and is reused

// once, for each slot of the pipeline
let qid = conn.create_query(&t1.table, &t1.rec, &t1.rec)?;
// each write: the query owns its rows; the values go into them
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)?;
  }
  None => return Err(IcError::new(err::IC_ERROR_NO_SUCH_FIELD)),
}
// ... start a transaction, define the write, commit ...
// at the end
conn.free_query(qid)?;

typeQueryId a Copy id; the query lives in its connection

makeconn.create_queryconn.create_unique_query

useconn.query_mut(q)key_row_mutattr_rowconn.read_keyconn.write_keyerror

freeconn.free_query when idle or completed

Two records: the key record holds every primary key field, or every indexed field; one record over every field serves as both, as t1.rec here.
conn.query_mut(q) lends the query until the borrow ends; its rows are two Vec<u8> of its own, sized by its records.

iclaustron: examples/rust/pipeline.rs (run, start_write; values elided); crates/ic_apid/src/transaction.rs (create_query, query_mut, free_query); query.rs (QueryId, ApidQuery, key_row_mut, check_records); doc/user-guide.md §3, §4

Connection · a query's life

A query goes Idle, Defined, Sent, Completed, and back to Idle

A query starts idle: the program fills its key row, and for a write its attribute row, through conn.query_mut(q).

conn.read_key or write_key defines it on a transaction: KEY_OP_REQ is packed at once into its node's staged buffer.

conn.send_queries, or a flush, writes every node's staged requests, one write per data node: the query is sent.

In a poll, KEY_OP_CONF and the row's ROW_DATA_REP, or a KEY_OP_REF, complete it: its row is in the attribute row, or its error is set.

Taken from the executed list, or once its callback returns, it is idle and can be defined again. From define to completion its rows are the library's.

iclaustron: crates/ic_apid/src/query.rs (QueryState, Execution, key_row_mut); transaction.rs (query_mut, read_key, send_queries, take_row, get_next_executed_query); doc/user-guide.md §4, §8

Connection · transactions

A transaction is local until its first send; the hint picks its coordinator

typeTransIdTransactionHint

makeconn.start_transaction from a hint and its table

useconn.hint_for_keyconn.hint_for_unique_keyconn.commit_transactionconn.rollback_transactionconn.transaction(t)

freeconn.close_transaction once done

conn.hint_for_key hashes the key as the data nodes do, character columns in their collation, and names the partition that holds the row.

Starting takes a coordinator record at a started node holding that partition: a free one of this connection's, or a new one, a round trip.

The transaction holds its 64-bit transaction id, its coordinator record, and its defined and sent queries. The data node knows nothing yet.

Its first KEY_OP_REQ carries the start flag: only then does the coordinator hear of it. A hint only chooses a node; any row may be touched.

iclaustron: crates/ic_apid/src/transaction.rs (Transaction, TransactionHint, start_transaction, choose_start_node, hint_for_key, hint_for_unique_key); apid_conn.rs (coord_record, next_transaction_id); doc/user-guide.md §4

Connection · ending a transaction

A transaction is done when it has ended and every query has completed

stateconn.transaction(t) gives it: commit_state(), one of CommitState::*

errorerror()error_data()

gcigci() the global checkpoint of a commit

closeconn.close_transaction its query lists kept for the next

A started transaction takes queries. Commit or rollback is asked for, and goes with the next send, on its last query or alone.

The coordinator commits it and says so, in KEY_OP_CONF or COMMIT_CONF, with the global checkpoint the commit went into.

A rollback asked for, a failed write that aborts on error, or the coordinator's TRANS_ABORTED_REP ends it rolled back; its error says why.

Done is ended with every query completed: its coordinator record goes back to the free list. Close it then.

iclaustron: crates/ic_apid/src/transaction.rs (CommitState, Transaction::is_done, commit_state, error, gci, take_key_conf, end_transaction, release_if_done, transaction, close_transaction); examples/rust/lib.rs (finish); doc/user-guide.md §4

Connection · scans

A scan holds its row and, while it runs, a receiver per fragment

typeScanIdScanBatchId

makeconn.create_scanconn.create_index_scan

useconn.start_scanconn.next_rowconn.next_batchconn.release_scan_batchconn.scan_row

freeconn.free_scan not while it runs

A scan is made once over a table and a record, and holds the row read last. Each run takes a coordinator record of its own.

While a run lasts it has one receiver per fragment scanned at once. A receiver's id has bit 23 set, which no query's id has.

A receiver brings a batch at a time, to the ready list or a callback; handed back, it asks for more. Module 07 teaches the use.

iclaustron: crates/ic_apid/src/scan.rs (Scan, Receiver, ScanState, ScanId, ScanBatchId, IC_RECEIVER_TAG, create_scan, free_scan, deliver); apid_conn.rs (scan_pools, receivers, batches, scan_of_record); doc/rust/04-api-design.md (Scans); doc/rust/02-architecture.md (Data flow for a scan)

Executor · executors and submitters

An executor runs other threads' requests on a connection of its own

typeExecutorSubmitter

makeExecutor::startexecutor.submitter() one per thread, or a clone

usesubmitter.table_bindsubmitter.executesubmitter.scanexecutor.stats

dropexecutor.stop() or its drop, before the global; a submitter left is refused

Any number of application threads, each with its own submitter, put requests into the executor's inbox, beside the signal pages of the receive threads.

The executor runs each request as a transaction on its connection, with queries kept per shape, and leaves the result in the submitter's slot.

iclaustron: crates/ic_apid/src/executor.rs (Executor::start, submitter, stop, Drop, Submitter, Clone for Submitter, Slot, Runner, Pool, ScanPool, IC_EXECUTOR_QUEUE); thread_conn.rs (Inbox); doc/user-guide.md §5 (Through an executor)

Executor · shapes, requests, readers

Shapes describe an operation once; a request carries one transaction

let shape = OpShape::new(&table, &rec, &rec)?; // made once
// ...
let mut req =
  Request::with_op(KeyOp::read(&shape, ReadKind::Committed));
req.ops_mut()[0].key_row_mut().copy_from_slice(&key);
sub.execute(&mut req)?;
match req.ops()[0].error() {
  Some(e) if e.code as u32 == key_op::IC_NDB_ERROR_NO_SUCH_ROW => {
    // no such row
  }
  Some(e) => return Err(e),
  None => row.copy_from_slice(req.ops()[0].attr_row()),
}

op shapeArc<OpShape>OpShape::newOpShape::new_uniquewith_program

requestRequestRequest::with_oppushKeyOp::readKeyOp::write

scan shapeArc<ScanShape>ScanShape::newScanShape::new_indexwith_program

readerScanReaderScanReader::newnext_rowrowclose

A shape is made once and shared by every thread; a request owns its rows and is run again with new keys. Module 06 runs them, module 07 the readers.
The executor keeps queries for a shape only while an Arc of it lives. execute moves the request to the executor's thread and back.

iclaustron: doc/rust/04-api-design.md (Executors: the shape); doc/user-guide.md §5 (Through an executor, the Rust request); crates/ic_apid/src/executor.rs (OpShape, KeyOp, Request, Submitter::execute, Pool); scan_reader.rs (ScanShape, ScanReader)

Handles · programs, run in the data node

A program is made once over a table and run by any number of operations

fn val_still(t1: &T1) -> Result<Arc<Program>, IcError> {
  let mut b = ProgramBuilder::new(&t1.table);
  let val = b.column(t1.val);
  let read = b.input(0);
  let same = b.compare(Cmp::Eq, val, read);
  b.finish(Some(same))
}
let still = val_still(&t1)?;
let guarded = conn.create_query(&t1.table, &id_rec, &name_rec)?;
for (expected, name) in [(0, "checked"), (5, "too late")] {
  let query = match conn.query_mut(guarded) { /* ... */ };
  query.set_program(Some(&still))?;
  query.set_input(0, expected)?;   // ... define, commit, wait
}

typeArc<Program>ProgramBuilder

makeProgramBuilder::newb.finish(Some(cond))b.finish(None)

usequery.set_programconn.set_scan_programOpShape::with_programScanShape::with_program

dropthe builder once finished; the program lives while what runs it holds its Arc

Only the inputs change from one operation to the next: up to 16 integers each. Module 08 uses programs.

iclaustron: examples/rust/conditional_update.rs (val_still; run, part 1: the match arms, the key and the name elided); crates/ic_apid/src/program.rs (ProgramBuilder::finish, Program, IC_PROGRAM_INPUTS); query.rs (set_program, set_input); scan.rs (set_scan_program); doc/user-guide.md §6

Replies · finding the object

A reply finds its object: the block number, then one index

Each request names our block, 0x8000 plus the thread id, as the address for its reply; the reply's first word is a number of ours.

A receive thread reads the block number only, and posts the signal to that thread's inbox, a page per round. The user thread decodes it.

In its poll, a single request's reply, such as COORD_ALLOC_CONF, matches an expected reply: its request number, signal and node.

KEY_OP_CONF, COMMIT_CONF and ROLLBACK_CONF carry our pointer for the coordinator record: its place indexes active[], which holds the transaction.

ROW_DATA_REP and KEY_OP_REF name the query by its id; a scan's rows name a receiver, bit 23 set. Each is one index, no search.

iclaustron: crates/ic_apid/src/thread_conn.rs (Router, ThreadConnection::block_number); apid_conn.rs (receive_view, Expectations::offer, coord_index_of); transaction.rs (route_reply, trans_named, take_row); scan.rs (scan_named, receiver_named); doc/rust/architecture.html §2

Replies · the bits

Two numbers in a reply: our record pointer and the transaction id

A late reply, left from an earlier use of the same place, carries another transaction id: it matches nothing and is dropped, and counted.

iclaustron: crates/ic_apid/src/apid_conn.rs (IC_COORD_PTR_TAG, coord_ptr_of, coord_index_of, transaction_id, trans_count_after, next_transaction_id, note_stray); thread_conn.rs (ThreadTable::trans_counter); transaction.rs (trans_named); doc/rust/architecture.html §2

Memory · once warm

Memory: once warm, nothing is allocated per transaction or per signal

Queries own their rows

Two Vec<u8> made with the query, sized by its records, reached through conn.query_mut(q). The library never points into the program's memory, and needs no unsafe code for the rows.

Kept, not freed

A closed transaction's query lists, up to 8 192 spare; coordinator records on free lists; send buffers swapped with a spare; signal pages at their largest size.

Copies at the edges

The synchronous calls copy from and into the caller's slices. A request moves to the executor and back; the executor copies its rows into its query and out. A batch moves to a reader without a copy.

On the define, send and poll path, a thread with hundreds of transactions in flight starts each in constant time and allocates nothing for it.

iclaustron: doc/rust/architecture.html §2 (Memory rules); crates/ic_apid/src/query.rs (module notes); transaction.rs (IC_SPARE_QUERY_LISTS, close_transaction); apid_conn.rs (coord_record); apid_global.rs (Sender, take_waiting); signal_page.rs; key_op.rs (read_committed, write_key); executor.rs (Submitter::execute, admit, completion copies); scan_reader.rs (module notes)

Ownership · Arc, owned, Copy

Who holds what, across threads

Type	Holds	Across threads
ApidGlobal	its threads, and their shared state in an `Arc`	one owner, lent as `&ApidGlobal`
ApidConnection	that `Arc`, bound tables, queries, transactions, scans	`Send`: moved to a thread, used by one
Arc<TableDef>, Arc<IndexDef>	a description the cache shares; an index, its table	shared by any thread
Record	its fields and its table's `Arc`; cloned where kept	shared by any thread
QueryId, TransId, ScanId	place and generation; the object is the connection's	`Copy`; names only in its connection
Arc<OpShape>, Arc<ScanShape>	table, records; the executor keeps queries or scans while one lives	shared by any thread
Arc<Program>	its code, for one table	shared by any thread
Submitter	the executor's inbox, and a slot of its own	`Send`; one per thread, or a clone
Request, ScanReader	their rows, and their shapes' `Arc`s	a request goes to the executor and back

An Arc is the count: what the cache lets go of lives on while anyone holds it. Module 09 shows how the crates build these.

iclaustron: crates/ic_apid/src/apid_global.rs (ApidGlobal); apid_conn.rs (ApidConnection); dict_cache.rs (module notes); record.rs (Record); query.rs (QueryId); program.rs (Program); executor.rs (Executor::start, OpShape, Pool, Submitter, Request); scan_reader.rs (ScanShape, ScanReader); examples/rust/scan_executor.rs (run); tools/ic_bench/src/main.rs (bench)

Summary · objects by model

Objects in the five programming models

Model	The program makes	The library uses for it	Owned by
Synchronous	an `ApidConnection`, `TableDef`s, `Record`s; a `ScanId`	per key call a coordinator record and an expected reply; a scan's receivers	the calling thread
Batched synchronous	`QueryId`s, `TransId`s; `ScanId`s and their `ScanBatchId`s	coordinator records, requests staged per node, receivers	the calling thread
Asynchronous	queries with a callback and user_ref, transactions; scans	the executed list and the ready list; callbacks run inside the poll	the thread that polls
Synchronous through an executor	an `Executor`, a `Submitter` per thread, `OpShape`s, a `Request`; `ScanShape`s, `ScanReader`s	the executor's connection: a kept query per operation, a transaction per request, a kept scan	the executor's thread; each caller its `Submitter`, `Request`, `ScanReader`
Batched synchronous through an executor	a `Request` of many `KeyOp`s; several `ScanReader`s	the same, with one send per data node for every caller's requests	the same

Modules 04 to 08 teach each model; this table says which objects each one stands on.

iclaustron: crates/ic_apid/src/key_op.rs (read_committed, write_key, run_op, read_unique_committed); transaction.rs; scan.rs; executor.rs (Runner, Flight, Pool, ScanPool, Request, KeyOp); scan_reader.rs (Feed, ScanReader); doc/user-guide.md §5

Recap · module 03