base/src/windows/tube.rs - platform/external/crosvm - Gitiles

 // Copyright 2021 The Chromium OS Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 use std::{
     io::{
         Cursor, Read, Write, {self},
     },
     time::Duration,
 };

 use crate::descriptor::{AsRawDescriptor, FromRawDescriptor, SafeDescriptor};
 use crate::{
     platform::{deserialize_with_descriptors, RawDescriptor, SerializeDescriptors},
     tube::{Error, RecvTube, Result, SendTube},
     BlockingMode, CloseNotifier, FramingMode, PollToken, ReadNotifier, StreamChannel,
 };
 use data_model::DataInit;
 use lazy_static::lazy_static;
 use serde::{de::DeserializeOwned, Deserialize, Serialize, Serializer};
 use std::{
     mem,
     os::windows::io::{AsRawHandle, RawHandle},
 };
 use winapi::shared::winerror::ERROR_MORE_DATA;

 /// Bidirectional tube that support both send and recv.
 ///
 /// NOTE: serializing this type across processes is slightly involved. Suppose there is a Tube pair
 /// (A, B). We wish to send B to another process, and communicate with it using A from the current
 /// process:
 ///     1. B's target_pid must be set to the current PID *before* serialization. There is a
 ///        serialization hook that sets it to the current PID automatically if target_pid is unset.
 ///     2. A's target_pid must be set to the PID of the process where B was sent.
 ///
 /// If instead you are sending both A and B to separate processes, then:
 ///     1. A's target_pid must be set to B's pid, manually.
 ///     2. B's target_pid must be set to A's pid, manually.
 ///
 /// Automating all of this and getting a completely clean interface is tricky. We would need
 /// intercept the serialization of Tubes in any part of Serde messages, and use Weak refs to sync
 /// state about PIDs between the ends. There are alternatives like reusing the underlying
 /// StreamChannel to share PIDs, or having a separate pipe just for this purpose; however, we've yet
 /// to find a compelling solution that isn't a mess to implement. Suggestions are welcome.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Tube {
     socket: StreamChannel,

     // Default target_pid to current PID on serialization (see `Tube` comment header for details).
     #[serde(serialize_with = "set_tube_pid_on_serialize")]
     target_pid: Option<u32>,
 }

 /// For a Tube which has not had its target_pid set, when it is serialized, we should automatically
 /// default it to the current process, because the other end will be in the current process.
 fn set_tube_pid_on_serialize<S>(
     existing_pid_value: &Option<u32>,
     serializer: S,
 ) -> std::result::Result<S::Ok, S::Error>
 where
     S: Serializer,
 {
     match existing_pid_value {
         Some(pid) => serializer.serialize_u32(*pid),
         None => serializer.serialize_u32(ALIAS_PID.lock().unwrap_or(std::process::id())),
     }
 }

 #[derive(Copy, Clone, Debug)]
 #[repr(C)]
 struct MsgHeader {
     msg_json_size: usize,
     descriptor_json_size: usize,
 }

 // Safe because it only has data and has no implicit padding.
 unsafe impl DataInit for MsgHeader {}

 lazy_static! {
     static ref DH_TUBE: sync::Mutex<Option<DuplicateHandleTube>> = sync::Mutex::new(None);
     static ref ALIAS_PID: sync::Mutex<Option<u32>> = sync::Mutex::new(None);
 }

 /// Set a tube to delegate duplicate handle calls.
 pub fn set_duplicate_handle_tube(dh_tube: DuplicateHandleTube) {
     DH_TUBE.lock().replace(dh_tube);
 }

 /// Set alias pid for use with a DuplicateHandleTube.
 pub fn set_alias_pid(alias_pid: u32) {
     ALIAS_PID.lock().replace(alias_pid);
 }

 impl Tube {
     /// Create a pair of connected tubes. Request is sent in one direction while response is
     /// received in the other direction.
     /// The result is in the form (server, client).
     pub fn pair() -> Result<(Tube, Tube)> {
         let (socket1, socket2) = StreamChannel::pair(BlockingMode::Blocking, FramingMode::Message)
             .map_err(|e| Error::Pair(io::Error::from_raw_os_error(e.errno())))?;

         Ok((Tube::new(socket1), Tube::new(socket2)))
     }

     /// Create a pair of connected tubes with the specified buffer size.
     /// Request is sent in one direction while response is received in the other direction.
     /// The result is in the form (server, client).
     pub fn pair_with_buffer_size(buffer_size: usize) -> Result<(Tube, Tube)> {
         let (socket1, socket2) = StreamChannel::pair_with_buffer_size(
             BlockingMode::Blocking,
             FramingMode::Message,
             buffer_size,
         )
         .map_err(|e| Error::Pair(io::Error::from_raw_os_error(e.errno())))?;
         let tube1 = Tube::new(socket1);
         let tube2 = Tube::new(socket2);
         Ok((tube1, tube2))
     }

     // Create a new `Tube`.
     pub fn new(socket: StreamChannel) -> Tube {
         Tube {
             socket,
             target_pid: None,
         }
     }

     pub(super) fn try_clone(&self) -> Result<Self> {
         Ok(Tube {
             socket: self.socket.try_clone().map_err(Error::Clone)?,
             target_pid: self.target_pid,
         })
     }

     pub fn send<T: Serialize>(&self, msg: &T) -> Result<()> {
         serialize_and_send(|buf| self.socket.write_immutable(buf), msg, self.target_pid)
     }

     pub fn recv<T: DeserializeOwned>(&self) -> Result<T> {
         deserialize_and_recv(|buf| (&self.socket).read(buf))
     }

     /// NOTE: On Windows this will only succeed if called on a server pipe. See #pair
     /// documentation to ensure you have a server pipe before calling.
     #[cfg(windows)]
     pub fn flush_blocking(&mut self) -> Result<()> {
         self.socket.flush_blocking().map_err(Error::Flush)
     }

     /// For Tubes that span processes, this method must be used to set the PID of the other end
     /// of the Tube, otherwise sending handles to the other end won't work.
     pub fn set_target_pid(&mut self, target_pid: u32) {
         self.target_pid = Some(target_pid);
     }

     /// Returns the PID of the process at the other end of the Tube, if any is set.
     pub fn target_pid(&self) -> Option<u32> {
         self.target_pid
     }

     /// 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.");
     }

     /// 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.");
     }
 }

 pub fn serialize_and_send<T: Serialize, F: Fn(&[u8]) -> io::Result<usize>>(
     write_fn: F,
     msg: &T,
     target_pid: Option<u32>,
 ) -> Result<()> {
     let msg_serialize = SerializeDescriptors::new(&msg);
     let msg_json = serde_json::to_vec(&msg_serialize).map_err(Error::Json)?;
     let msg_descriptors = msg_serialize.into_descriptors();

     let mut duped_descriptors = Vec::with_capacity(msg_descriptors.len());
     for desc in msg_descriptors {
         // Safe because these handles are guaranteed to be valid. Details:
         // 1. They come from sys_util::descriptor_reflection::with_as_descriptor.
         // 2. with_as_descriptor is intended to be applied to owned descriptor types (e.g. File,
         //    SafeDescriptor).
         // 3. The owning object is borrowed by msg until sending is complete.
         duped_descriptors.push(duplicate_handle(desc, target_pid)? as usize)
     }

     let descriptor_json = if duped_descriptors.is_empty() {
         None
     } else {
         Some(serde_json::to_vec(&duped_descriptors).map_err(Error::Json)?)
     };

     let header = MsgHeader {
         msg_json_size: msg_json.len(),
         descriptor_json_size: descriptor_json.as_ref().map_or(0, |json| json.len()),
     };

     let mut data_packet = Cursor::new(Vec::with_capacity(
         header.as_slice().len() + header.msg_json_size + header.descriptor_json_size,
     ));
     data_packet
         .write(header.as_slice())
         .map_err(Error::SendIoBuf)?;
     data_packet
         .write(msg_json.as_slice())
         .map_err(Error::SendIoBuf)?;
     if let Some(descriptor_json) = descriptor_json {
         data_packet
             .write(descriptor_json.as_slice())
             .map_err(Error::SendIoBuf)?;
     }

     // Multiple writers (producers) are safe because each write is atomic.
     let data_bytes = data_packet.into_inner();

     write_fn(&data_bytes).map_err(Error::SendIo)?;
     Ok(())
 }

 fn duplicate_handle(desc: RawHandle, target_pid: Option<u32>) -> Result<RawHandle> {
     match target_pid {
         Some(pid) => match &*DH_TUBE.lock() {
             Some(tube) => tube.request_duplicate_handle(pid, desc),
             None => {
                 win_util::duplicate_handle_with_target_pid(desc, pid).map_err(Error::DupDescriptor)
             }
         },
         None => win_util::duplicate_handle(desc).map_err(Error::DupDescriptor),
     }
 }

 /// Reads a part of a Tube packet asserting that it was correctly read. This means:
 /// * Treats partial "message" (transport framing) reads are Ok, as long as we filled our buffer.
 ///   We use this to ignore errors when reading the message header, which has the lengths we need
 ///   to allocate our buffers for the remainder of the message.
 /// * We filled the supplied buffer.
 fn perform_read<F: Fn(&mut [u8]) -> io::Result<usize>>(
     read_fn: &F,
     buf: &mut [u8],
 ) -> io::Result<usize> {
     let res = match read_fn(buf) {
         Ok(s) => Ok(s),
         Err(e)
             if e.raw_os_error()
                 .map_or(false, |errno| errno == ERROR_MORE_DATA as i32) =>
         {
             Ok(buf.len())
         }
         Err(e) => Err(e),
     };

     let bytes_read = res?;
     if bytes_read != buf.len() {
         Err(io::Error::new(
             io::ErrorKind::UnexpectedEof,
             "failed to fill whole buffer",
         ))
     } else {
         Ok(bytes_read)
     }
 }

 /// Deserializes a Tube packet by calling the supplied read function. This function MUST
 /// assert that the buffer was filled.
 pub fn deserialize_and_recv<T: DeserializeOwned, F: Fn(&mut [u8]) -> io::Result<usize>>(
     read_fn: F,
 ) -> Result<T> {
     let mut header_bytes = vec![0u8; mem::size_of::<MsgHeader>()];
     perform_read(&read_fn, header_bytes.as_mut_slice()).map_err(Error::Recv)?;

     // Safe because the header is always written by the send function, and only that function
     // writes to this channel.
     let header =
         MsgHeader::from_slice(header_bytes.as_slice()).expect("Tube header failed to deserialize.");

     let mut msg_json = vec![0u8; header.msg_json_size];
     perform_read(&read_fn, msg_json.as_mut_slice()).map_err(Error::Recv)?;

     if msg_json.is_empty() {
         // This means we got a message header, but there is no json body (due to a zero size in
         // the header). This should never happen because it means the receiver is getting no
         // data whatsoever from the sender.
         return Err(Error::RecvUnexpectedEmptyBody);
     }

     let msg_descriptors: Vec<RawDescriptor> = if header.descriptor_json_size > 0 {
         let mut msg_descriptors_json = vec![0u8; header.descriptor_json_size];
         perform_read(&read_fn, msg_descriptors_json.as_mut_slice()).map_err(Error::Recv)?;
         let descriptor_usizes: Vec<usize> =
             serde_json::from_slice(msg_descriptors_json.as_slice()).map_err(Error::Json)?;

         // Safe because the usizes are RawDescriptors that were converted to usize in the send
         // method.
         descriptor_usizes
             .iter()
             .map(|item| *item as RawDescriptor)
             .collect()
     } else {
         Vec::new()
     };

     let mut msg_descriptors_safe = msg_descriptors
         .into_iter()
         .map(|v| {
             Some(unsafe {
                 // Safe because the socket returns new fds that are owned locally by this scope.
                 SafeDescriptor::from_raw_descriptor(v)
             })
         })
         .collect();

     deserialize_with_descriptors(
         || serde_json::from_slice(&msg_json),
         &mut msg_descriptors_safe,
     )
     .map_err(Error::Json)
 }

 #[derive(PollToken, Eq, PartialEq, Copy, Clone)]
 enum Token {
     SocketReady,
 }

 impl AsRawDescriptor for Tube {
     fn as_raw_descriptor(&self) -> RawDescriptor {
         self.socket.as_raw_descriptor()
     }
 }

 impl AsRawHandle for Tube {
     fn as_raw_handle(&self) -> RawHandle {
         self.as_raw_descriptor()
     }
 }

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

 impl CloseNotifier for Tube {
     fn get_close_notifier(&self) -> &dyn AsRawDescriptor {
         self.socket.get_close_notifier()
     }
 }

 impl AsRawHandle for SendTube {
     fn as_raw_handle(&self) -> RawHandle {
         self.0.as_raw_descriptor()
     }
 }

 impl AsRawHandle for RecvTube {
     fn as_raw_handle(&self) -> RawHandle {
         self.0.as_raw_descriptor()
     }
 }

 /// A request to duplicate a handle to a target process.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct DuplicateHandleRequest {
     pub target_alias_pid: u32,
     pub handle: usize,
 }

 /// Contains a duplicated handle or None if an error occurred.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct DuplicateHandleResponse {
     pub handle: Option<usize>,
 }

 /// Wrapper for tube which is used to delegate DuplicateHandle function calls to
 /// the broker process.
 #[derive(Serialize, Deserialize, Debug)]
 pub struct DuplicateHandleTube(Tube);

 impl DuplicateHandleTube {
     pub fn new(tube: Tube) -> Self {
         Self(tube)
     }

     pub fn request_duplicate_handle(
         &self,
         target_alias_pid: u32,
         handle: RawHandle,
     ) -> Result<RawHandle> {
         let req = DuplicateHandleRequest {
             target_alias_pid,
             handle: handle as usize,
         };
         self.0.send(&req)?;
         let res: DuplicateHandleResponse = self.0.recv()?;
         res.handle
             .map(|h| h as RawHandle)
             .ok_or(Error::BrokerDupDescriptor)
     }
 }

 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::{EventContext, EventTrigger, PollToken, ReadNotifier};
     use std::time;

     const EVENT_WAIT_TIME: time::Duration = time::Duration::from_secs(10);

     #[derive(PollToken, Debug, Eq, PartialEq, Copy, Clone)]
     enum Token {
         ReceivedData,
     }

     #[test]
     fn test_serialize_tube() {
         let (tube_1, tube_2) = Tube::pair().unwrap();
         let event_ctx: EventContext<Token> = EventContext::build_with(&[EventTrigger::from(
             tube_2.get_read_notifier(),
             Token::ReceivedData,
         )])
         .unwrap();

         // Serialize the Tube
         let msg_serialize = SerializeDescriptors::new(&tube_1);
         let serialized = serde_json::to_vec(&msg_serialize).unwrap();
         let msg_descriptors = msg_serialize.into_descriptors();

         // Deserialize the Tube
         let mut msg_descriptors_safe = msg_descriptors
             .into_iter()
             .map(|v| Some(unsafe { SafeDescriptor::from_raw_descriptor(v) }))
             .collect();
         let tube_deserialized: Tube = deserialize_with_descriptors(
             || serde_json::from_slice(&serialized),
             &mut msg_descriptors_safe,
         )
         .unwrap();

         // Send a message through deserialized Tube
         tube_deserialized.send(&"hi".to_string()).unwrap();

         assert_eq!(event_ctx.wait_timeout(EVENT_WAIT_TIME).unwrap().len(), 1);
         assert_eq!(tube_2.recv::<String>().unwrap(), "hi");
     }
 }
	// Copyright 2021 The Chromium OS Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	use std::{
	io::{
	Cursor, Read, Write, {self},
	},
	time::Duration,
	};

	use crate::descriptor::{AsRawDescriptor, FromRawDescriptor, SafeDescriptor};
	use crate::{
	platform::{deserialize_with_descriptors, RawDescriptor, SerializeDescriptors},
	tube::{Error, RecvTube, Result, SendTube},
	BlockingMode, CloseNotifier, FramingMode, PollToken, ReadNotifier, StreamChannel,
	};
	use data_model::DataInit;
	use lazy_static::lazy_static;
	use serde::{de::DeserializeOwned, Deserialize, Serialize, Serializer};
	use std::{
	mem,
	os::windows::io::{AsRawHandle, RawHandle},
	};
	use winapi::shared::winerror::ERROR_MORE_DATA;

	/// Bidirectional tube that support both send and recv.
	///
	/// NOTE: serializing this type across processes is slightly involved. Suppose there is a Tube pair
	/// (A, B). We wish to send B to another process, and communicate with it using A from the current
	/// process:
	/// 1. B's target_pid must be set to the current PID before serialization. There is a
	/// serialization hook that sets it to the current PID automatically if target_pid is unset.
	/// 2. A's target_pid must be set to the PID of the process where B was sent.
	///
	/// If instead you are sending both A and B to separate processes, then:
	/// 1. A's target_pid must be set to B's pid, manually.
	/// 2. B's target_pid must be set to A's pid, manually.
	///
	/// Automating all of this and getting a completely clean interface is tricky. We would need
	/// intercept the serialization of Tubes in any part of Serde messages, and use Weak refs to sync
	/// state about PIDs between the ends. There are alternatives like reusing the underlying
	/// StreamChannel to share PIDs, or having a separate pipe just for this purpose; however, we've yet
	/// to find a compelling solution that isn't a mess to implement. Suggestions are welcome.
	#[derive(Serialize, Deserialize, Debug)]
	pub struct Tube {
	socket: StreamChannel,

	// Default target_pid to current PID on serialization (see `Tube` comment header for details).
	#[serde(serialize_with = "set_tube_pid_on_serialize")]
	target_pid: Option<u32>,
	}

	/// For a Tube which has not had its target_pid set, when it is serialized, we should automatically
	/// default it to the current process, because the other end will be in the current process.
	fn set_tube_pid_on_serialize<S>(
	existing_pid_value: &Option<u32>,
	serializer: S,
	) -> std::result::Result<S::Ok, S::Error>
	where
	S: Serializer,
	{
	match existing_pid_value {
	Some(pid) => serializer.serialize_u32(*pid),
	None => serializer.serialize_u32(ALIAS_PID.lock().unwrap_or(std::process::id())),
	}
	}

	#[derive(Copy, Clone, Debug)]
	#[repr(C)]
	struct MsgHeader {
	msg_json_size: usize,
	descriptor_json_size: usize,
	}

	// Safe because it only has data and has no implicit padding.
	unsafe impl DataInit for MsgHeader {}

	lazy_static! {
	static ref DH_TUBE: sync::Mutex<Option<DuplicateHandleTube>> = sync::Mutex::new(None);
	static ref ALIAS_PID: sync::Mutex<Option<u32>> = sync::Mutex::new(None);
	}

	/// Set a tube to delegate duplicate handle calls.
	pub fn set_duplicate_handle_tube(dh_tube: DuplicateHandleTube) {
	DH_TUBE.lock().replace(dh_tube);
	}

	/// Set alias pid for use with a DuplicateHandleTube.
	pub fn set_alias_pid(alias_pid: u32) {
	ALIAS_PID.lock().replace(alias_pid);
	}

	impl Tube {
	/// Create a pair of connected tubes. Request is sent in one direction while response is
	/// received in the other direction.
	/// The result is in the form (server, client).
	pub fn pair() -> Result<(Tube, Tube)> {
	let (socket1, socket2) = StreamChannel::pair(BlockingMode::Blocking, FramingMode::Message)
	.map_err(\|e\| Error::Pair(io::Error::from_raw_os_error(e.errno())))?;

	Ok((Tube::new(socket1), Tube::new(socket2)))
	}

	/// Create a pair of connected tubes with the specified buffer size.
	/// Request is sent in one direction while response is received in the other direction.
	/// The result is in the form (server, client).
	pub fn pair_with_buffer_size(buffer_size: usize) -> Result<(Tube, Tube)> {
	let (socket1, socket2) = StreamChannel::pair_with_buffer_size(
	BlockingMode::Blocking,
	FramingMode::Message,
	buffer_size,
	)
	.map_err(\|e\| Error::Pair(io::Error::from_raw_os_error(e.errno())))?;
	let tube1 = Tube::new(socket1);
	let tube2 = Tube::new(socket2);
	Ok((tube1, tube2))
	}

	// Create a new `Tube`.
	pub fn new(socket: StreamChannel) -> Tube {
	Tube {
	socket,
	target_pid: None,
	}
	}

	pub(super) fn try_clone(&self) -> Result<Self> {
	Ok(Tube {
	socket: self.socket.try_clone().map_err(Error::Clone)?,
	target_pid: self.target_pid,
	})
	}

	pub fn send<T: Serialize>(&self, msg: &T) -> Result<()> {
	serialize_and_send(\|buf\| self.socket.write_immutable(buf), msg, self.target_pid)
	}

	pub fn recv<T: DeserializeOwned>(&self) -> Result<T> {
	deserialize_and_recv(\|buf\| (&self.socket).read(buf))
	}

	/// NOTE: On Windows this will only succeed if called on a server pipe. See #pair
	/// documentation to ensure you have a server pipe before calling.
	#[cfg(windows)]
	pub fn flush_blocking(&mut self) -> Result<()> {
	self.socket.flush_blocking().map_err(Error::Flush)
	}

	/// For Tubes that span processes, this method must be used to set the PID of the other end
	/// of the Tube, otherwise sending handles to the other end won't work.
	pub fn set_target_pid(&mut self, target_pid: u32) {
	self.target_pid = Some(target_pid);
	}

	/// Returns the PID of the process at the other end of the Tube, if any is set.
	pub fn target_pid(&self) -> Option<u32> {
	self.target_pid
	}

	/// 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.");
	}

	/// 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.");
	}
	}

	pub fn serialize_and_send<T: Serialize, F: Fn(&[u8]) -> io::Result<usize>>(
	write_fn: F,
	msg: &T,
	target_pid: Option<u32>,
	) -> Result<()> {
	let msg_serialize = SerializeDescriptors::new(&msg);
	let msg_json = serde_json::to_vec(&msg_serialize).map_err(Error::Json)?;
	let msg_descriptors = msg_serialize.into_descriptors();

	let mut duped_descriptors = Vec::with_capacity(msg_descriptors.len());
	for desc in msg_descriptors {
	// Safe because these handles are guaranteed to be valid. Details:
	// 1. They come from sys_util::descriptor_reflection::with_as_descriptor.
	// 2. with_as_descriptor is intended to be applied to owned descriptor types (e.g. File,
	// SafeDescriptor).
	// 3. The owning object is borrowed by msg until sending is complete.
	duped_descriptors.push(duplicate_handle(desc, target_pid)? as usize)
	}

	let descriptor_json = if duped_descriptors.is_empty() {
	None
	} else {
	Some(serde_json::to_vec(&duped_descriptors).map_err(Error::Json)?)
	};

	let header = MsgHeader {
	msg_json_size: msg_json.len(),
	descriptor_json_size: descriptor_json.as_ref().map_or(0, \|json\| json.len()),
	};

	let mut data_packet = Cursor::new(Vec::with_capacity(
	header.as_slice().len() + header.msg_json_size + header.descriptor_json_size,
	));
	data_packet
	.write(header.as_slice())
	.map_err(Error::SendIoBuf)?;
	data_packet
	.write(msg_json.as_slice())
	.map_err(Error::SendIoBuf)?;
	if let Some(descriptor_json) = descriptor_json {
	data_packet
	.write(descriptor_json.as_slice())
	.map_err(Error::SendIoBuf)?;
	}

	// Multiple writers (producers) are safe because each write is atomic.
	let data_bytes = data_packet.into_inner();

	write_fn(&data_bytes).map_err(Error::SendIo)?;
	Ok(())
	}

	fn duplicate_handle(desc: RawHandle, target_pid: Option<u32>) -> Result<RawHandle> {
	match target_pid {
	Some(pid) => match &*DH_TUBE.lock() {
	Some(tube) => tube.request_duplicate_handle(pid, desc),
	None => {
	win_util::duplicate_handle_with_target_pid(desc, pid).map_err(Error::DupDescriptor)
	}
	},
	None => win_util::duplicate_handle(desc).map_err(Error::DupDescriptor),
	}
	}

	/// Reads a part of a Tube packet asserting that it was correctly read. This means:
	/// * Treats partial "message" (transport framing) reads are Ok, as long as we filled our buffer.
	/// We use this to ignore errors when reading the message header, which has the lengths we need
	/// to allocate our buffers for the remainder of the message.
	/// * We filled the supplied buffer.
	fn perform_read<F: Fn(&mut [u8]) -> io::Result<usize>>(
	read_fn: &F,
	buf: &mut [u8],
	) -> io::Result<usize> {
	let res = match read_fn(buf) {
	Ok(s) => Ok(s),
	Err(e)
	if e.raw_os_error()
	.map_or(false, \|errno\| errno == ERROR_MORE_DATA as i32) =>
	{
	Ok(buf.len())
	}
	Err(e) => Err(e),
	};

	let bytes_read = res?;
	if bytes_read != buf.len() {
	Err(io::Error::new(
	io::ErrorKind::UnexpectedEof,
	"failed to fill whole buffer",
	))
	} else {
	Ok(bytes_read)
	}
	}

	/// Deserializes a Tube packet by calling the supplied read function. This function MUST
	/// assert that the buffer was filled.
	pub fn deserialize_and_recv<T: DeserializeOwned, F: Fn(&mut [u8]) -> io::Result<usize>>(
	read_fn: F,
	) -> Result<T> {
	let mut header_bytes = vec![0u8; mem::size_of::<MsgHeader>()];
	perform_read(&read_fn, header_bytes.as_mut_slice()).map_err(Error::Recv)?;

	// Safe because the header is always written by the send function, and only that function
	// writes to this channel.
	let header =
	MsgHeader::from_slice(header_bytes.as_slice()).expect("Tube header failed to deserialize.");

	let mut msg_json = vec![0u8; header.msg_json_size];
	perform_read(&read_fn, msg_json.as_mut_slice()).map_err(Error::Recv)?;

	if msg_json.is_empty() {
	// This means we got a message header, but there is no json body (due to a zero size in
	// the header). This should never happen because it means the receiver is getting no
	// data whatsoever from the sender.
	return Err(Error::RecvUnexpectedEmptyBody);
	}

	let msg_descriptors: Vec<RawDescriptor> = if header.descriptor_json_size > 0 {
	let mut msg_descriptors_json = vec![0u8; header.descriptor_json_size];
	perform_read(&read_fn, msg_descriptors_json.as_mut_slice()).map_err(Error::Recv)?;
	let descriptor_usizes: Vec<usize> =
	serde_json::from_slice(msg_descriptors_json.as_slice()).map_err(Error::Json)?;

	// Safe because the usizes are RawDescriptors that were converted to usize in the send
	// method.
	descriptor_usizes
	.iter()
	.map(\|item\| *item as RawDescriptor)
	.collect()
	} else {
	Vec::new()
	};

	let mut msg_descriptors_safe = msg_descriptors
	.into_iter()
	.map(\|v\| {
	Some(unsafe {
	// Safe because the socket returns new fds that are owned locally by this scope.
	SafeDescriptor::from_raw_descriptor(v)
	})
	})
	.collect();

	deserialize_with_descriptors(
	\|\| serde_json::from_slice(&msg_json),
	&mut msg_descriptors_safe,
	)
	.map_err(Error::Json)
	}

	#[derive(PollToken, Eq, PartialEq, Copy, Clone)]
	enum Token {
	SocketReady,
	}

	impl AsRawDescriptor for Tube {
	fn as_raw_descriptor(&self) -> RawDescriptor {
	self.socket.as_raw_descriptor()
	}
	}

	impl AsRawHandle for Tube {
	fn as_raw_handle(&self) -> RawHandle {
	self.as_raw_descriptor()
	}
	}

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

	impl CloseNotifier for Tube {
	fn get_close_notifier(&self) -> &dyn AsRawDescriptor {
	self.socket.get_close_notifier()
	}
	}

	impl AsRawHandle for SendTube {
	fn as_raw_handle(&self) -> RawHandle {
	self.0.as_raw_descriptor()
	}
	}

	impl AsRawHandle for RecvTube {
	fn as_raw_handle(&self) -> RawHandle {
	self.0.as_raw_descriptor()
	}
	}

	/// A request to duplicate a handle to a target process.
	#[derive(Serialize, Deserialize, Debug)]
	pub struct DuplicateHandleRequest {
	pub target_alias_pid: u32,
	pub handle: usize,
	}

	/// Contains a duplicated handle or None if an error occurred.
	#[derive(Serialize, Deserialize, Debug)]
	pub struct DuplicateHandleResponse {
	pub handle: Option<usize>,
	}

	/// Wrapper for tube which is used to delegate DuplicateHandle function calls to
	/// the broker process.
	#[derive(Serialize, Deserialize, Debug)]
	pub struct DuplicateHandleTube(Tube);

	impl DuplicateHandleTube {
	pub fn new(tube: Tube) -> Self {
	Self(tube)
	}

	pub fn request_duplicate_handle(
	&self,
	target_alias_pid: u32,
	handle: RawHandle,
	) -> Result<RawHandle> {
	let req = DuplicateHandleRequest {
	target_alias_pid,
	handle: handle as usize,
	};
	self.0.send(&req)?;
	let res: DuplicateHandleResponse = self.0.recv()?;
	res.handle
	.map(\|h\| h as RawHandle)
	.ok_or(Error::BrokerDupDescriptor)
	}
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use crate::{EventContext, EventTrigger, PollToken, ReadNotifier};
	use std::time;

	const EVENT_WAIT_TIME: time::Duration = time::Duration::from_secs(10);

	#[derive(PollToken, Debug, Eq, PartialEq, Copy, Clone)]
	enum Token {
	ReceivedData,
	}

	#[test]
	fn test_serialize_tube() {
	let (tube_1, tube_2) = Tube::pair().unwrap();
	let event_ctx: EventContext<Token> = EventContext::build_with(&[EventTrigger::from(
	tube_2.get_read_notifier(),
	Token::ReceivedData,
	)])
	.unwrap();

	// Serialize the Tube
	let msg_serialize = SerializeDescriptors::new(&tube_1);
	let serialized = serde_json::to_vec(&msg_serialize).unwrap();
	let msg_descriptors = msg_serialize.into_descriptors();

	// Deserialize the Tube
	let mut msg_descriptors_safe = msg_descriptors
	.into_iter()
	.map(\|v\| Some(unsafe { SafeDescriptor::from_raw_descriptor(v) }))
	.collect();
	let tube_deserialized: Tube = deserialize_with_descriptors(
	\|\| serde_json::from_slice(&serialized),
	&mut msg_descriptors_safe,
	)
	.unwrap();

	// Send a message through deserialized Tube
	tube_deserialized.send(&"hi".to_string()).unwrap();

	assert_eq!(event_ctx.wait_timeout(EVENT_WAIT_TIME).unwrap().len(), 1);
	assert_eq!(tube_2.recv::<String>().unwrap(), "hi");
	}
	}