Documentazione di CapDAG

Invocazione peer

Come una cartuccia chiama una capacità fornita da un’altra cartuccia mentre gestisce una richiesta.

In questa sezione

Come una cartuccia chiama la cap di un’altra cartuccia durante la gestione di una richiesta.

Concetto

Una cartuccia che gestisce una richiesta può invocare una capacità fornita da una cartuccia diversa. La cartuccia chiamante non sa, né le interessa, quale cartuccia fornisca la cap di destinazione; è il RelaySwitch a gestire l’instradamento in base alla corrispondenza dei cap URN.

Il caso d’uso più comune sono le cartucce di ML che chiamano modelcartridge per scaricare i file dei modelli prima di caricarli. La cartuccia dei modelli gestisce i download HTTP, la cache e la segnalazione dell’avanzamento. La cartuccia di ML si limita a chiamare la cap di download e riceve il percorso del file.

Le invocazioni peer avvengono durante l’esecuzione dell’handler, non prima né dopo. L’handler è a metà di una richiesta, ha bisogno di qualcosa da un’altra cartuccia, la chiama, attende la risposta e prosegue.

Sorgente: capdag-rs/src/bifaci/cartridge_runtime.rs (trait PeerInvoker).

Trait PeerInvoker

Gli handler accedono all’invocazione peer attraverso il trait PeerInvoker, ottenuto tramite req.peer():

#[async_trait]
pub trait PeerInvoker: Send + Sync {
    fn call(&self, cap_urn: &str) -> Result<PeerCall, RuntimeError>;

    async fn call_with_bytes(
        &self,
        cap_urn: &str,
        args: &[(&str, &[u8])],
    ) -> Result<PeerResponse, RuntimeError>;
}

call(cap_urn) apre una nuova invocazione peer. Restituisce un handle PeerCall per inviare gli argomenti. Il frame REQ viene inviato immediatamente con il cap URN indicato.

call_with_bytes(cap_urn, args) è un metodo di comodo che combina l’intera sequenza: apre la chiamata, crea uno stream per argomento, scrive i byte, chiude ogni stream, invia END e restituisce la PeerResponse. Ogni elemento di args è una coppia (media_urn, data).

// Di comodo: una sola chiamata, gestione automatica degli stream
let response = req.peer()
    .call_with_bytes("cap:in=...;out=...;download", &[
        ("media:model-spec;enc=utf-8", model_spec.as_bytes()),
    ])
    .await?;
let model_path = response.collect_bytes().await?;

PeerCall

PeerCall è l’handle per costruire passo per passo un’invocazione peer con più argomenti:

pub struct PeerCall {
    sender: Arc<dyn FrameSender>,
    request_id: MessageId,
    max_chunk: usize,
    response_rx: Option<mpsc::UnboundedReceiver<Frame>>,
}

arg(media_urn) crea un OutputStream per un argomento. Ogni argomento è uno stream indipendente con il proprio stream_id UUID. Il chiamante scrive i dati nello stream e lo chiude:

let call = req.peer().call("cap:in=...;out=...;something")?;
let arg1 = call.arg("media:enc=utf-8");
arg1.start(false, None)?;
arg1.write(prompt.as_bytes()).await?;
arg1.close().await?;

let arg2 = call.arg("media:binary");
arg2.start(false, None)?;
arg2.write(&image_bytes).await?;
arg2.close().await?;

let response = call.finish().await?;

finish() invia il frame END per la richiesta peer, avvia il demux della risposta e restituisce una PeerResponse. Dopo aver chiamato finish(), non è più possibile inviare argomenti. Chiamare finish() due volte restituisce un errore.

Sorgente: cartridge_runtime.rs (PeerCall).

PeerResponse

PeerResponse produce i dati del peer e i suoi frame LOG, intercalati in ordine di arrivo.

Una risposta ha uno stream per ciascun output su cui la cap del peer risponde. Lo stream che non nomina alcun output è l’output principale della cap, e i suoi elementi sono i Data della risposta: un chiamante legge la risposta della cap senza sapere se la cap dichiara altri output. Gli elementi di ogni altro stream arrivano come Output, etichettati con il media URN dell’output che il loro stream ha nominato. La risposta termina con l’END (o l’ERR) della richiesta; la fine di uno stream termina solo quello stream. Ogni stream riceve crediti per nome man mano che i suoi elementi vengono consumati, così un chiamante lento rallenta ciascuno degli output del peer.

recv() restituisce l’elemento successivo. Restituisce None quando la risposta termina.

collect_bytes() accumula tutti gli elementi Data in un vettore di byte e fallisce se incontra un elemento Log non gestito. Gli elementi di un altro output non fanno parte di ciò che viene raccolto. Usa collect_bytes_forwarding() o svuota recv() quando il peer può emettere diagnostica.

collect_value() richiede esattamente un elemento Data e fallisce se è presente un qualsiasi elemento Log. Non scarta mai silenziosamente la diagnostica del peer.

collect_bytes_forwarding(output, base, weight) accumula dati peer finiti, inoltrando nel contempo l’avanzamento nell’intervallo di avanzamento del chiamante e preservando l’attribuzione alla sorgente dei normali frame diagnostici.

PeerResponseItem

pub enum PeerResponseItem {
    Data(Result<ciborium::Value, StreamError>, Option<StreamMeta>),
    Output {
        output: String,
        item: Result<ciborium::Value, StreamError>,
        meta: Option<StreamMeta>,
    },
    Log(Frame),
}
  • Data: un chunk CBOR decodificato dell’output principale del peer, con i metadati del chunk. Byte, testo o altri tipi CBOR.
  • Output: un chunk CBOR decodificato di un altro output della cap del peer. output è il media URN di quell’output, come lo ha nominato il suo stream.
  • Log: un frame LOG grezzo del peer. Porta aggiornamenti di avanzamento, messaggi di stato e informazioni diagnostiche. I frame LOG vengono consegnati in tempo reale man mano che arrivano e non vengono trattenuti in buffer finché non iniziano i dati.

Sorgente: cartridge_runtime.rs (PeerResponse).

Inoltro dell’avanzamento del peer

Quando un handler invoca un peer (per esempio per il download di un modello), di solito vuole inoltrare gli aggiornamenti di avanzamento del peer nel proprio stream di avanzamento, mappati nell’intervallo appropriato. Lo schema standard:

let response = call.finish().await?;
while let Some(item) = response.recv().await {
    match item {
        PeerResponseItem::Data(Ok(value), _meta) => {
            data.extend(extract_bytes(&value));
        }
        // Un altro output della cap del peer: non ciò che questo chiamante ha chiesto.
        PeerResponseItem::Output { .. } => {}
        PeerResponseItem::Log(frame) => {
            if let Some(peer_progress) = frame.log_progress() {
                // Mappa il [0.0, 1.0] del peer nel nostro intervallo [0.0, 0.25]
                let mapped = peer_progress * 0.25;
                req.output().progress(mapped, frame.log_message().unwrap_or(""));
            }
        }
        PeerResponseItem::Data(Err(e), _) => return Err(e.into()),
    }
}
// Prosegui con il nostro lavoro da avanzamento 0.25 in su
req.output().progress(0.25, "Model downloaded, loading...");

La funzione map_progress in capdag-rs/src/orchestrator/types.rs fornisce una versione generale: map_progress(value, base, weight) mappa un valore in [0.0, 1.0] in [base, base + weight].

Gli handler di solito riservano la prima porzione del proprio intervallo di avanzamento alle chiamate peer (per esempio [0.0, 0.25] per il download del modello) e usano il resto per il proprio lavoro (per esempio [0.25, 1.0] per l’inferenza).

Flusso dei frame durante una chiamata peer

A livello di protocollo, l’invocazione peer coinvolge più componenti:

sequenceDiagram
    participant PA as Cartuccia A
    participant PHA as CartridgeHost A
    participant RSA as RelaySlave A
    participant SW as RelaySwitch
    participant PB as Cartuccia B

    Note over PA,PB: Fase di richiesta
    PA->>PHA: REQ(cap), senza XID
    Note over PHA: registra il RID in outgoing_rids
    PHA->>RSA: REQ(cap)
    RSA->>SW: REQ(cap)
    Note over SW: assegna l'XID, registra il peer
    SW->>PB: REQ(cap, xid)

    PA->>PHA: STREAM_START
    PA->>PHA: CHUNK(arg)
    PA->>PHA: STREAM_END
    PA->>PHA: END
    Note over PHA,PB: (inoltrati attraverso il relay)

    Note over PA,PB: Fase di risposta
    PB-->>SW: LOG(progress)
    SW-->>RSA: LOG(progress)
    RSA-->>PHA: LOG(progress)
    PHA-->>PA: LOG(progress)

    PB-->>SW: CHUNK(result)
    SW-->>RSA: CHUNK
    RSA-->>PHA: CHUNK
    PHA-->>PA: CHUNK

    PB-->>SW: END
    SW-->>RSA: END
    RSA-->>PHA: END
    PHA-->>PA: END

Punti chiave:

  1. La cartuccia A invia la REQ senza XID, perché le cartucce non conoscono gli ID di instradamento.
  2. Il CartridgeHostRuntime A registra il RID della richiesta in outgoing_rids per sapere dove instradare la risposta.
  3. Il RelaySwitch assegna un XID e registra la richiesta peer in peer_requests.
  4. I frame LOG della cartuccia B tornano indietro lungo lo stesso percorso. Il CartridgeHostRuntime A li instrada al canale PeerResponse della cartuccia A facendo corrispondere il RID in outgoing_rids.
  5. La risposta (CHUNK, END) segue lo stesso percorso di instradamento fino alla cartuccia A.

Sorgente: host_runtime.rs (outgoing_rids), relay_switch.rs (peer_requests), cartridge_runtime.rs.

NoPeerInvoker

NoPeerInvoker è uno stub che rifiuta tutte le chiamate peer con un errore. Viene usato in modalità CLI (dove non c’è alcun host attraverso cui instradare) e nei test che non hanno bisogno dell’invocazione peer.

pub struct NoPeerInvoker;

impl PeerInvoker for NoPeerInvoker {
    fn call(&self, _cap_urn: &str) -> Result<PeerCall, RuntimeError> {
        Err(RuntimeError::PeerRequest(
            "Peer invocation not supported in this context".to_string(),
        ))
    }
}

Sorgente: cartridge_runtime.rs (NoPeerInvoker).

Equivalente Swift

L’implementazione Swift in capdag-objc/Sources/Bifaci/CartridgeRuntime.swift fornisce:

  • PeerInvoker come protocollo Swift (equivalente al trait Rust).
  • PeerCall e PeerResponse con la stessa semantica ma API idiomatiche per Swift.
  • invoke(capUrn:args:) come metodo di comodo (equivalente a call_with_bytes).
  • Gli elementi della risposta consegnati come casi di enum Swift anziché come varianti di enum Rust.

Il flusso dei frame e il comportamento di instradamento sono identici; la semantica a livello di protocollo non differisce tra le implementazioni.