/src/crosvm/base/src/tube.rs

Source
// Copyright 2021 The ChromiumOS Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use std::io;
use std::time::Duration;

use remain::sorted;
use serde::de::DeserializeOwned;
use serde::Deserialize;
use serde::Serialize;
use thiserror::Error as ThisError;

pub use crate::sys::tube::*;

impl Tube {
    /// Given a Tube end, creates two new ends, one each for sending and receiving.
    pub fn split_to_send_recv(self) -> Result<(SendTube, RecvTube)> {
        // Safe because receiving isn't allowd on this end.
        #[allow(deprecated)]
        let send_end = self.try_clone()?;

        Ok((SendTube(send_end), RecvTube(self)))
    }

    /// Creates a Send/Recv pair of Tubes.
    pub fn directional_pair() -> Result<(SendTube, RecvTube)> {
        let (t1, t2) = Self::pair()?;
        Ok((SendTube(t1), RecvTube(t2)))
    }

    pub fn try_clone_send_tube(&self) -> Result<SendTube> {
        // Safe because receiving is only allowed on original Tube.
        #[allow(deprecated)]
        let send_end = self.try_clone()?;
        Ok(SendTube(send_end))
    }
}

use crate::AsRawDescriptor;
use crate::ReadNotifier;

#[derive(Serialize, Deserialize)]
#[serde(transparent)]
/// A Tube end which can only send messages. Cloneable.
pub struct SendTube(pub(crate) Tube);

#[allow(dead_code)]
impl SendTube {
    /// TODO(b/145998747, b/184398671): this method should be removed.
    pub fn set_send_timeout(&self, _timeout: Option<Duration>) -> Result<()> {
        unimplemented!("To be removed/refactored upstream.");
    }

    pub fn send<T: Serialize>(&self, msg: &T) -> Result<()> {
        self.0.send(msg)
    }

    pub fn try_clone(&self) -> Result<Self> {
        Ok(SendTube(
            #[allow(deprecated)]
            self.0.try_clone()?,
        ))
    }

    /// Never call this function, it is for use by cros_async to provide
    /// directional wrapper types only. Using it in any other context may
    /// violate concurrency assumptions. (Type splitting across crates has put
    /// us in a situation where we can't use Rust privacy to enforce this.)
    #[deprecated]
    pub fn into_tube(self) -> Tube {
        self.0
    }
}

#[derive(Serialize, Deserialize)]
#[serde(transparent)]
/// A Tube end which can only recv messages.
pub struct RecvTube(pub(crate) Tube);

#[allow(dead_code)]
impl RecvTube {
    pub fn recv<T: DeserializeOwned>(&self) -> Result<T> {
        self.0.recv()
    }

    /// TODO(b/145998747, b/184398671): this method should be removed.
    pub fn set_recv_timeout(&self, _timeout: Option<Duration>) -> Result<()> {
        unimplemented!("To be removed/refactored upstream.");
    }

    /// Never call this function, it is for use by cros_async to provide
    /// directional wrapper types only. Using it in any other context may
    /// violate concurrency assumptions. (Type splitting across crates has put
    /// us in a situation where we can't use Rust privacy to enforce this.)
    #[deprecated]
    pub fn into_tube(self) -> Tube {
        self.0
    }
}

impl ReadNotifier for RecvTube {
    fn get_read_notifier(&self) -> &dyn AsRawDescriptor {
        self.0.get_read_notifier()
    }
}

#[sorted]
#[derive(ThisError, Debug)]
pub enum Error {
    #[cfg(windows)]
    #[error("attempt to duplicate descriptor via broker failed")]
    BrokerDupDescriptor,
    #[error("failed to clone transport: {0}")]
    Clone(io::Error),
    #[error("tube was disconnected")]
    Disconnected,
    #[error("failed to duplicate descriptor: {0}")]
    DupDescriptor(io::Error),
    #[cfg(windows)]
    #[error("failed to flush named pipe: {0}")]
    Flush(io::Error),
    #[cfg(unix)]
    #[error("byte framing mode is not supported")]
    InvalidFramingMode,
    #[error("failed to serialize/deserialize json from packet: {0}")]
    Json(serde_json::Error),
    #[error("cancelled a queued async operation")]
    OperationCancelled,
    #[error("failed to crate tube pair: {0}")]
    Pair(io::Error),
    #[cfg(any(windows, feature = "proto_tube"))]
    #[error("encountered protobuf error: {0}")]
    Proto(protobuf::Error),
    #[error("failed to receive packet: {0}")]
    Recv(io::Error),
    #[error("attempted to receive too many file descriptors")]
    RecvTooManyFds,
    #[error("Received a message with a zero sized body. This should not happen.")]
    RecvUnexpectedEmptyBody,
    #[cfg(unix)]
    #[error("failed to construct ScmSocket: {0}")]
    ScmSocket(io::Error),
    #[error("failed to send packet: {0}")]
    Send(io::Error),
    #[error("failed to write packet to intermediate buffer: {0}")]
    SendIoBuf(io::Error),
    #[error("attempted to send too many file descriptors")]
    SendTooManyFds,
    #[error("failed to set recv timeout: {0}")]
    SetRecvTimeout(io::Error),
    #[error("failed to set send timeout: {0}")]
    SetSendTimeout(io::Error),
}

pub type Result<T> = std::result::Result<T, Error>;

Coverage Report

Created: 2025-12-31 06:16

Line	Count	Source
1		// Copyright 2021 The ChromiumOS Authors
2		// Use of this source code is governed by a BSD-style license that can be
3		// found in the LICENSE file.
4
5		use std::io;
6		use std::time::Duration;
7
8		use remain::sorted;
9		use serde::de::DeserializeOwned;
10		use serde::Deserialize;
11		use serde::Serialize;
12		use thiserror::Error as ThisError;
13
14		pub use crate::sys::tube::*;
15
16		impl Tube {
17		/// Given a Tube end, creates two new ends, one each for sending and receiving.
18	0	pub fn split_to_send_recv(self) -> Result<(SendTube, RecvTube)> {
19		// Safe because receiving isn't allowd on this end.
20		#[allow(deprecated)]
21	0	let send_end = self.try_clone()?;
22
23	0	Ok((SendTube(send_end), RecvTube(self)))
24	0	}
25
26		/// Creates a Send/Recv pair of Tubes.
27	0	pub fn directional_pair() -> Result<(SendTube, RecvTube)> {
28	0	let (t1, t2) = Self::pair()?;
29	0	Ok((SendTube(t1), RecvTube(t2)))
30	0	}
31
32	0	pub fn try_clone_send_tube(&self) -> Result<SendTube> {
33		// Safe because receiving is only allowed on original Tube.
34		#[allow(deprecated)]
35	0	let send_end = self.try_clone()?;
36	0	Ok(SendTube(send_end))
37	0	}
38		}
39
40		use crate::AsRawDescriptor;
41		use crate::ReadNotifier;
42
43		#[derive(Serialize, Deserialize)]
44		#[serde(transparent)]
45		/// A Tube end which can only send messages. Cloneable.
46		pub struct SendTube(pub(crate) Tube);
47
48		#[allow(dead_code)]
49		impl SendTube {
50		/// TODO(b/145998747, b/184398671): this method should be removed.
51	0	pub fn set_send_timeout(&self, _timeout: Option<Duration>) -> Result<()> {
52	0	unimplemented!("To be removed/refactored upstream.");
53		}
54
55	0	pub fn send<T: Serialize>(&self, msg: &T) -> Result<()> {
56	0	self.0.send(msg)
57	0	} Unexecuted instantiation: <base::tube::SendTube>::send::<_> Unexecuted instantiation: <base::tube::SendTube>::send::<base::VmEventType> Unexecuted instantiation: <base::tube::SendTube>::send::<bool> Unexecuted instantiation: <base::tube::SendTube>::send::<()>
58
59	0	pub fn try_clone(&self) -> Result<Self> {
60		Ok(SendTube(
61		#[allow(deprecated)]
62	0	self.0.try_clone()?,
63		))
64	0	}
65
66		/// Never call this function, it is for use by cros_async to provide
67		/// directional wrapper types only. Using it in any other context may
68		/// violate concurrency assumptions. (Type splitting across crates has put
69		/// us in a situation where we can't use Rust privacy to enforce this.)
70		#[deprecated]
71	0	pub fn into_tube(self) -> Tube {
72	0	self.0
73	0	}
74		}
75
76		#[derive(Serialize, Deserialize)]
77		#[serde(transparent)]
78		/// A Tube end which can only recv messages.
79		pub struct RecvTube(pub(crate) Tube);
80
81		#[allow(dead_code)]
82		impl RecvTube {
83	0	pub fn recv<T: DeserializeOwned>(&self) -> Result<T> {
84	0	self.0.recv()
85	0	} Unexecuted instantiation: <base::tube::RecvTube>::recv::<_> Unexecuted instantiation: <base::tube::RecvTube>::recv::<()>
86
87		/// TODO(b/145998747, b/184398671): this method should be removed.
88	0	pub fn set_recv_timeout(&self, _timeout: Option<Duration>) -> Result<()> {
89	0	unimplemented!("To be removed/refactored upstream.");
90		}
91
92		/// Never call this function, it is for use by cros_async to provide
93		/// directional wrapper types only. Using it in any other context may
94		/// violate concurrency assumptions. (Type splitting across crates has put
95		/// us in a situation where we can't use Rust privacy to enforce this.)
96		#[deprecated]
97	0	pub fn into_tube(self) -> Tube {
98	0	self.0
99	0	}
100		}
101
102		impl ReadNotifier for RecvTube {
103	0	fn get_read_notifier(&self) -> &dyn AsRawDescriptor {
104	0	self.0.get_read_notifier()
105	0	}
106		}
107
108		#[sorted]
109		#[derive(ThisError, Debug)]
110		pub enum Error {
111		#[cfg(windows)]
112		#[error("attempt to duplicate descriptor via broker failed")]
113		BrokerDupDescriptor,
114		#[error("failed to clone transport: {0}")]
115		Clone(io::Error),
116		#[error("tube was disconnected")]
117		Disconnected,
118		#[error("failed to duplicate descriptor: {0}")]
119		DupDescriptor(io::Error),
120		#[cfg(windows)]
121		#[error("failed to flush named pipe: {0}")]
122		Flush(io::Error),
123		#[cfg(unix)]
124		#[error("byte framing mode is not supported")]
125		InvalidFramingMode,
126		#[error("failed to serialize/deserialize json from packet: {0}")]
127		Json(serde_json::Error),
128		#[error("cancelled a queued async operation")]
129		OperationCancelled,
130		#[error("failed to crate tube pair: {0}")]
131		Pair(io::Error),
132		#[cfg(any(windows, feature = "proto_tube"))]
133		#[error("encountered protobuf error: {0}")]
134		Proto(protobuf::Error),
135		#[error("failed to receive packet: {0}")]
136		Recv(io::Error),
137		#[error("attempted to receive too many file descriptors")]
138		RecvTooManyFds,
139		#[error("Received a message with a zero sized body. This should not happen.")]
140		RecvUnexpectedEmptyBody,
141		#[cfg(unix)]
142		#[error("failed to construct ScmSocket: {0}")]
143		ScmSocket(io::Error),
144		#[error("failed to send packet: {0}")]
145		Send(io::Error),
146		#[error("failed to write packet to intermediate buffer: {0}")]
147		SendIoBuf(io::Error),
148		#[error("attempted to send too many file descriptors")]
149		SendTooManyFds,
150		#[error("failed to set recv timeout: {0}")]
151		SetRecvTimeout(io::Error),
152		#[error("failed to set send timeout: {0}")]
153		SetSendTimeout(io::Error),
154		}
155
156		pub type Result<T> = std::result::Result<T, Error>;