Struct persist::indexed::Indexed

pub struct Indexed<L: Log, B: BlobRead> {
    log: L,
    blob: BlobCache<B>,
    listeners: HashMap<Id, Vec<Sender<ListenEvent>>>,
    metrics: Arc<Metrics>,
    state: AppliedState,
    in_flight_trace_compactions: HashMap<Id, CompactTraceReq>,
    pending: Option<Pending>,
}

A persistent, compacting, indexed data structure of (Key, Value, Time, Diff) updates.

Indexed contains a set of named persistent Arrangements.

Notes:

• Requests are split into two types: unbatched and batched. An unbatched command is run entirely by itself (the applied state has just been written to durable storage, then the command is run, then the resulting state is immediately written to durable storage). A batched command is applied to the machine state, but instead of immediately serializing the state to storage, we buffer the command response in Pending. Any other batched command can also be run and similarly buffered in Pending. Then, the next time we get an unbatched command (or step is called), all pending batched commands are made durable at once (and responses filled, listeners updated, etc). This is a performance optimization to amortize the cost of writing to durable storage across many of those requests. The most common requests (write, seal, allow_compaction) are all batched to exploit this. All unbatched commands are expected to be relatively infrequent (to avoid excessive barriers in our pipelining).
• When evaluating a request, the work of updating the state is given to AppliedState (which has no knowledge of storage, etc). Then, if this was successful, Indexed will serialize AppliedState and durably write it down.

Fields

log: L

blob: BlobCache<B>

listeners: HashMap<Id, Vec<Sender<ListenEvent>>>

metrics: Arc<Metrics>

state: AppliedState

in_flight_trace_compactions: HashMap<Id, CompactTraceReq>

pending: Option<Pending>

Implementations

impl<L: Log, B: BlobRead> Indexed<L, B>

pub fn new(
    log: L,
    blob: BlobCache<B>,
    metrics: Arc<Metrics>
) -> Result<Self, Error>

Returns a new Indexed, initializing each Unsealed and Trace with the existing data for them in the blob storage, if any.

impl<L: Log, B: Blob> Indexed<L, B>

fn try_set_meta(&mut self, prev_meta: BlobMeta) -> Result<(), Error>

Serializes and attempt to commit the current in-memory AppliedState to durable storage, and if not, reverts back to the given previous version (which is expected to match what’s in durable storage).

Precondition: pending has been emptied

fn apply_unbatched_cmd<T, WorkFn: FnOnce(&mut AppliedState, &mut BlobCache<B>) -> Result<T, Error>>(
    &mut self,
    work_fn: WorkFn
) -> Result<T, Error>

Applies an unbatched cmd to the machine state and snapshots the result to durable storage.

Precondition: pending has been emptied

fn apply_batched_cmd<WorkFn: FnOnce(&mut AppliedState, &mut Pending)>(
    &mut self,
    work_fn: WorkFn
)

pub fn close(&mut self) -> Result<(), Error>

Releases exclusive-writer locks and causes all future commands to error.

This method is idempotent.

pub fn apply(&mut self, cmd: Cmd) -> bool

Applies the given input to the state machine.

Returns false to indicate a graceful shutdown, true otherwise.

impl<L: Log, B: BlobRead> Indexed<L, B>

pub fn apply_read(&mut self, cmd: CmdRead) -> bool

Applies the given read-only input to the state machine.

Returns false to indicate a graceful shutdown, true otherwise.

impl<L: Log, B: BlobRead> Indexed<L, B>

fn validate(&self) -> Result<(), Error>

fn validate_matches_storage(
    blob: &BlobCache<B>,
    meta: &BlobMeta
) -> Result<(), Error>

Validates that the meta we might roll back to must be equal to the durably persisted meta.

fn validate_referenced_keys_exist(&self) -> Result<(), Error>

Validates that all of the referenced blob keys in all unsealeds and traces actually exist in blob’s key-val map.

fn validate_pending_empty(&self) -> Result<(), Error>

pub fn has_pending_responses(&self) -> bool

Return true if Pending has at least one pending response.

impl<L: Log, B: Blob> Indexed<L, B>

fn drain_pending(&mut self) -> Result<(), Error>

Commit any pending in-memory changes to persistent storage, respond to clients and notify any listeners.

impl<L: Log, B: Blob> Indexed<L, B>

fn compact_unsealed(&mut self) -> Result<(), Error>

Truncate all unsealeds, as much as possible.

Precondition: pending has been emptied

TODO: currently we do not attempt to compact unsealed batches and instead logically delete them from unsealed after all updates contained within a given unsealed batch have been moved over to trace. This policy works fine assuming data mostly arrives in order, or not very far in advance of the currently sealed time. We will need to revisit the unsealed compaction if that assumption stops being true.

fn compact_trace_maintenance_reqs(
    &mut self
) -> Result<Vec<MaintenanceReq>, Error>

fn handle_compact_trace_res(
    &mut self,
    id: Id,
    res: Result<CompactTraceRes, Error>
)

Handle the results of a prior request to compact a trace.

fn handle_maintenance_res(&mut self, res: MaintenanceRes)

Handle the results of a previously sent maintenance request.

fn step(&mut self) -> Result<Vec<MaintenanceReq>, Error>

pub fn step_or_log(&mut self) -> Vec<MaintenanceReq>

Drains writes from the log into the unsealed and does any necessary resulting compaction work.

In production, step_or_log should just be called in a loop (probably with some smarts about waiting to call it only after there have been some writes), but it’s exposed this way so we can write deterministic tests.

impl<L: Log, B: Blob> Indexed<L, B>

fn update_listeners(
    &self,
    updates: HashMap<Id, Vec<ColumnarRecords>>,
    seals: HashMap<Id, u64>
)

impl<L: Log, B: BlobRead> Indexed<L, B>

fn do_listen(
    &mut self,
    id: Id,
    sender: Sender<ListenEvent>
) -> Result<ArrangementSnapshot, Error>

Trait Implementations

impl<L: Debug + Log, B: Debug + BlobRead> Debug for Indexed<L, B>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.