liminal_server/server/connection/link_route/

executor.rs

//! Far requests use full supervised connections and the existing grant fences.

use crate::link::LinkHost;
use crate::link::calls::{CallError, CallExecution, CallExecutor, CallReply, CallRequest};
use crate::server::connection::{
    ConnectionHandle, ConnectionSupervisor, LoopbackClientEnd, LoopbackDuplex,
};
use liminal::protocol::{Frame, FrameHeader, codec};
use std::io::{Read, Write};
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use zeroize::Zeroizing;

struct Executor {
    supervisor: ConnectionSupervisor,
    ring_bytes: NonZeroUsize,
    answer_bytes: NonZeroUsize,
    retained: Arc<Retirement>,
}
#[derive(Default)]
struct Retirement {
    count: AtomicUsize,
    signal: Mutex<()>,
    changed: Condvar,
}
impl std::fmt::Debug for Executor {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("LinkCallExecutor")
            .field("ring_bytes", &self.ring_bytes)
            .field("answer_bytes", &self.answer_bytes)
            .finish_non_exhaustive()
    }
}

impl ConnectionSupervisor {
    /// Installs far calls after this supervisor's normal Lys admission is ready.
    /// `ring_bytes` bounds each inner transport; `answer_bytes` bounds a far reply
    /// before allocation, independently of the outer Noise frame ceiling.
    ///
    /// # Errors
    /// A second executor was installed on the same host.
    pub fn install_link_call_executor(
        &self,
        host: &LinkHost,
        ring_bytes: NonZeroUsize,
        answer_bytes: NonZeroUsize,
    ) -> Result<Arc<dyn CallExecutor>, CallError> {
        let executor: Arc<dyn CallExecutor> = Arc::new(Executor {
            supervisor: self.clone(),
            ring_bytes,
            answer_bytes,
            retained: Arc::new(Retirement::default()),
        });
        host.install_call_executor(&executor)?;
        Ok(executor)
    }
}

impl CallExecutor for Executor {
    fn retained(&self) -> usize {
        self.retained.count.load(Ordering::Acquire)
    }
    fn observe_empty(&self) -> Result<(), CallError> {
        let signal = self
            .retained
            .signal
            .lock()
            .map_err(|_| CallError::Unavailable)?;
        let signal = self
            .retained
            .changed
            .wait_while(signal, |()| {
                self.retained.count.load(Ordering::Acquire) != 0
            })
            .map_err(|_| CallError::Unavailable)?;
        drop(signal);
        Ok(())
    }
    fn start(
        &self,
        request: CallRequest,
        completed: Box<dyn FnOnce(Result<CallReply, CallError>) + Send>,
    ) -> Result<Box<dyn CallExecution>, CallError> {
        let correlation = request.correlation()?;
        let signal = self
            .retained
            .signal
            .lock()
            .map_err(|_| CallError::Unavailable)?;
        let (mut client, server) = LoopbackDuplex::bounded(self.ring_bytes.get());
        let handle = self
            .supervisor
            .spawn_loopback_connection(server)
            .map_err(|_| CallError::Capacity)?;
        let id = request.id();
        let bound = self.answer_bytes.get();
        let dispatcher = tracing::dispatcher::get_default(Clone::clone);
        let worker = std::thread::Builder::new()
            .name("liminal-linked-call".into())
            .spawn(move || {
                tracing::dispatcher::with_default(&dispatcher, || {
                    let answer = exchange(&mut client, &request, correlation, bound)
                        .and_then(|frame| CallReply::new(id, frame));
                    drop(request);
                    drop(client);
                    completed(answer);
                });
            })
            .map_err(|_| CallError::Unavailable)?;
        self.retained.count.fetch_add(1, Ordering::AcqRel);
        drop(signal);
        Ok(Box::new(Execution {
            handle,
            worker: Some(worker),
            retained: Arc::clone(&self.retained),
        }))
    }
}

struct Execution {
    handle: ConnectionHandle,
    worker: Option<JoinHandle<()>>,
    retained: Arc<Retirement>,
}
impl std::fmt::Debug for Execution {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("LinkCallExecution")
            .field("connection", &self.handle.pid())
            .finish_non_exhaustive()
    }
}
impl CallExecution for Execution {
    fn finish(&mut self) -> Result<(), CallError> {
        // A live loopback connection is asked to exit. The scheduler refuses
        // the ask only for a process that already holds its exit tombstone
        // (still in the table while its exit propagates), so a refusal means
        // the connection is exiting on its own and the worker is joined as it
        // is after an accepted ask. Reading the table again decided nothing:
        // the tombstone keeps the pid in it either way.
        if self.handle.is_live() && self.handle.request_crash().is_err() {
            tracing::debug!(
                connection = self.handle.pid(),
                "linked call connection is already exiting"
            );
        }
        if let Some(worker) = self.worker.take() {
            worker.join().map_err(|_| CallError::Unavailable)?;
        }
        Ok(())
    }
}
impl Drop for Execution {
    fn drop(&mut self) {
        if let Err(error) = self.finish() {
            tracing::error!(%error, "linked call execution retirement failed");
        }
        if let Ok(signal) = self.retained.signal.lock() {
            self.retained.count.fetch_sub(1, Ordering::AcqRel);
            self.retained.changed.notify_all();
            drop(signal);
        } else {
            tracing::error!("link_call_retirement_signal_unavailable");
        }
    }
}

fn exchange(
    client: &mut LoopbackClientEnd,
    request: &CallRequest,
    correlation: (u32, u64),
    bound: usize,
) -> Result<Frame, CallError> {
    client
        .write_all(request.wire_bytes())
        .map_err(|_| CallError::Unavailable)?;
    match read(client, bound)? {
        Frame::ConnectAck { .. } => {}
        Frame::ConnectError {
            reason_code,
            message,
            ..
        } => return Ok(refusal(correlation, reason_code, message)),
        _ => return Err(CallError::Invalid),
    }
    let frame = read(client, bound)?;
    match &frame {
        Frame::ConversationMessage {
            stream_id,
            conversation_id,
            ..
        }
        | Frame::ConversationError {
            stream_id,
            conversation_id,
            ..
        } if (*stream_id, *conversation_id) == correlation => Ok(frame),
        _ => Err(CallError::Correlation),
    }
}

fn read(client: &mut impl Read, bound: usize) -> Result<Frame, CallError> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    client
        .read_exact(&mut header)
        .map_err(|_| CallError::PeerGone)?;
    let length = usize::try_from(u32::from_be_bytes(
        header[6..10].try_into().map_err(|_| CallError::Invalid)?,
    ))
    .map_err(|_| CallError::Bound)?;
    let total = header
        .len()
        .checked_add(length)
        .filter(|length| *length <= bound)
        .ok_or(CallError::Bound)?;
    let mut bytes = Zeroizing::new(vec![0; total]);
    bytes[..header.len()].copy_from_slice(&header);
    client
        .read_exact(&mut bytes[header.len()..])
        .map_err(|_| CallError::PeerGone)?;
    let (frame, used) = codec::decode(&bytes).map_err(|_| CallError::Invalid)?;
    if used != total {
        return Err(CallError::Invalid);
    }
    Ok(frame)
}

const fn refusal(correlation: (u32, u64), reason_code: u16, message: Option<String>) -> Frame {
    Frame::ConversationError {
        flags: 0,
        stream_id: correlation.0,
        conversation_id: correlation.1,
        reason_code,
        message,
    }
}