renet/renet
renet
/
renet
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Extract socket handle into a new SocketMeta structure. 

This paves way for adding more metadata apart from handles,
such as caches and so on.

This commit also removes SocketHandle::EMPTY in favor of
SocketHandle::default() since they are functionally identical.
This commit is contained in:
whitequark 2017-11-22 03:50:09 +00:00
parent 715aa4b217
commit 6c83936872
6 changed files with 76 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/socket/icmp.rs
View File

@ -3,7 +3,7 @@ use managed::Managed;
use {Error, Result}; use {Error, Result};
use phy::{ChecksumCapabilities, DeviceCapabilities}; use phy::{ChecksumCapabilities, DeviceCapabilities};
use socket::{Socket, SocketHandle}; use socket::{Socket, SocketMeta, SocketHandle};
use storage::{Resettable, RingBuffer}; use storage::{Resettable, RingBuffer};
use wire::{IpAddress, IpEndpoint, IpProtocol, IpRepr}; use wire::{IpAddress, IpEndpoint, IpProtocol, IpRepr};
use wire::{Ipv4Address, Ipv4Repr}; use wire::{Ipv4Address, Ipv4Repr};
@ -92,7 +92,7 @@ pub type SocketBuffer<'a, 'b: 'a> = RingBuffer<'a, PacketBuffer<'b>>;
/// [bind]: #method.bind /// [bind]: #method.bind
#[derive(Debug)] #[derive(Debug)]
pub struct IcmpSocket<'a, 'b: 'a> { pub struct IcmpSocket<'a, 'b: 'a> {
handle: SocketHandle, pub(crate) meta: SocketMeta,
rx_buffer: SocketBuffer<'a, 'b>, rx_buffer: SocketBuffer<'a, 'b>,
tx_buffer: SocketBuffer<'a, 'b>, tx_buffer: SocketBuffer<'a, 'b>,
/// The endpoint this socket is communicating with /// The endpoint this socket is communicating with
@ -105,7 +105,7 @@ impl<'a, 'b> IcmpSocket<'a, 'b> {
/// Create an ICMPv4 socket with the given buffers. /// Create an ICMPv4 socket with the given buffers.
pub fn new(rx_buffer: SocketBuffer<'a, 'b>, tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> { pub fn new(rx_buffer: SocketBuffer<'a, 'b>, tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> {
Socket::Icmp(IcmpSocket { Socket::Icmp(IcmpSocket {
handle: SocketHandle::EMPTY, meta: SocketMeta::default(),
rx_buffer: rx_buffer, rx_buffer: rx_buffer,
tx_buffer: tx_buffer, tx_buffer: tx_buffer,
endpoint: Endpoint::default(), endpoint: Endpoint::default(),
@ -116,12 +116,7 @@ impl<'a, 'b> IcmpSocket<'a, 'b> {
/// Return the socket handle. /// Return the socket handle.
#[inline] #[inline]
pub fn handle(&self) -> SocketHandle { pub fn handle(&self) -> SocketHandle {
self.handle self.meta.handle
}
/// Set the socket handle.
pub(in super) fn set_handle(&mut self, handle: SocketHandle) {
self.handle = handle;
} }
/// Return the time-to-live (IPv4) or hop limit (IPv6) value used in outgoing packets. /// Return the time-to-live (IPv4) or hop limit (IPv6) value used in outgoing packets.
@ -256,7 +251,7 @@ impl<'a, 'b> IcmpSocket<'a, 'b> {
let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?; let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?;
packet_buf.endpoint = endpoint; packet_buf.endpoint = endpoint;
net_trace!("{}:{}: buffer to send {} octets", net_trace!("{}:{}: buffer to send {} octets",
self.handle, packet_buf.endpoint, size); self.meta.handle, packet_buf.endpoint, size);
Ok(&mut packet_buf.as_mut()[..size]) Ok(&mut packet_buf.as_mut()[..size])
} }
@ -276,7 +271,7 @@ impl<'a, 'b> IcmpSocket<'a, 'b> {
pub fn recv(&mut self) -> Result<(&[u8], IpAddress)> { pub fn recv(&mut self) -> Result<(&[u8], IpAddress)> {
let packet_buf = self.rx_buffer.dequeue_one()?; let packet_buf = self.rx_buffer.dequeue_one()?;
net_trace!("{}:{}: receive {} buffered octets", net_trace!("{}:{}: receive {} buffered octets",
self.handle, packet_buf.endpoint, packet_buf.size); self.meta.handle, packet_buf.endpoint, packet_buf.size);
Ok((&packet_buf.as_ref(), packet_buf.endpoint)) Ok((&packet_buf.as_ref(), packet_buf.endpoint))
} }
@ -322,14 +317,14 @@ impl<'a, 'b> IcmpSocket<'a, 'b> {
packet_buf.as_mut().copy_from_slice(ip_payload); packet_buf.as_mut().copy_from_slice(ip_payload);
packet_buf.endpoint = ip_repr.src_addr(); packet_buf.endpoint = ip_repr.src_addr();
net_trace!("{}:{}: receiving {} octets", net_trace!("{}:{}: receiving {} octets",
self.handle, packet_buf.endpoint, packet_buf.size); self.meta.handle, packet_buf.endpoint, packet_buf.size);
Ok(()) Ok(())
} }
pub(crate) fn dispatch<F>(&mut self, caps: &DeviceCapabilities, emit: F) -> Result<()> pub(crate) fn dispatch<F>(&mut self, caps: &DeviceCapabilities, emit: F) -> Result<()>
where F: FnOnce((IpRepr, Icmpv4Repr)) -> Result<()> where F: FnOnce((IpRepr, Icmpv4Repr)) -> Result<()>
{ {
let handle = self.handle; let handle = self.meta.handle;
let ttl = self.ttl.unwrap_or(64); let ttl = self.ttl.unwrap_or(64);
let checksum = &caps.checksum; let checksum = &caps.checksum;
self.tx_buffer.dequeue_one_with(|packet_buf| { self.tx_buffer.dequeue_one_with(|packet_buf| {

19
src/socket/mod.rs
View File

@ -74,6 +74,15 @@ pub enum Socket<'a, 'b: 'a> {
__Nonexhaustive(PhantomData<(&'a (), &'b ())>) __Nonexhaustive(PhantomData<(&'a (), &'b ())>)
} }
/// Network socket metadata.
///
/// This includes things that only external (to the socket, that is) code
/// is interested in, but which are more conveniently stored inside the socket itself.
#[derive(Debug, Default)]
pub(crate) struct SocketMeta {
handle: SocketHandle,
}
macro_rules! dispatch_socket { macro_rules! dispatch_socket {
($self_:expr, |$socket:ident [$( $mut_:tt )*]| $code:expr) => ({ ($self_:expr, |$socket:ident [$( $mut_:tt )*]| $code:expr) => ({
match $self_ { match $self_ {
@ -93,11 +102,15 @@ macro_rules! dispatch_socket {
impl<'a, 'b> Socket<'a, 'b> { impl<'a, 'b> Socket<'a, 'b> {
/// Return the socket handle. /// Return the socket handle.
pub fn handle(&self) -> SocketHandle { pub fn handle(&self) -> SocketHandle {
dispatch_socket!(self, |socket []| socket.handle()) self.meta().handle
} }
pub(crate) fn set_handle(&mut self, handle: SocketHandle) { pub(crate) fn meta(&self) -> &SocketMeta {
dispatch_socket!(self, |socket [mut]| socket.set_handle(handle)) dispatch_socket!(self, |socket []| &socket.meta)
}
pub(crate) fn meta_mut(&mut self) -> &mut SocketMeta {
dispatch_socket!(self, |socket [mut]| &mut socket.meta)
} }
pub(crate) fn poll_at(&self) -> Option<u64> { pub(crate) fn poll_at(&self) -> Option<u64> {

23
src/socket/raw.rs
View File

@ -4,7 +4,7 @@ use managed::Managed;
use {Error, Result}; use {Error, Result};
use phy::ChecksumCapabilities; use phy::ChecksumCapabilities;
use wire::{IpVersion, IpRepr, IpProtocol, Ipv4Repr, Ipv4Packet}; use wire::{IpVersion, IpRepr, IpProtocol, Ipv4Repr, Ipv4Packet};
use socket::{Socket, SocketHandle}; use socket::{Socket, SocketMeta, SocketHandle};
use storage::{Resettable, RingBuffer}; use storage::{Resettable, RingBuffer};
/// A buffered raw IP packet. /// A buffered raw IP packet.
@ -57,7 +57,7 @@ pub type SocketBuffer<'a, 'b: 'a> = RingBuffer<'a, PacketBuffer<'b>>;
/// transmit and receive packet buffers. /// transmit and receive packet buffers.
#[derive(Debug)] #[derive(Debug)]
pub struct RawSocket<'a, 'b: 'a> { pub struct RawSocket<'a, 'b: 'a> {
handle: SocketHandle, pub(crate) meta: SocketMeta,
ip_version: IpVersion, ip_version: IpVersion,
ip_protocol: IpProtocol, ip_protocol: IpProtocol,
rx_buffer: SocketBuffer<'a, 'b>, rx_buffer: SocketBuffer<'a, 'b>,
@ -71,7 +71,7 @@ impl<'a, 'b> RawSocket<'a, 'b> {
rx_buffer: SocketBuffer<'a, 'b>, rx_buffer: SocketBuffer<'a, 'b>,
tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> { tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> {
Socket::Raw(RawSocket { Socket::Raw(RawSocket {
handle: SocketHandle::EMPTY, meta: SocketMeta::default(),
ip_version, ip_version,
ip_protocol, ip_protocol,
rx_buffer, rx_buffer,
@ -82,12 +82,7 @@ impl<'a, 'b> RawSocket<'a, 'b> {
/// Return the socket handle. /// Return the socket handle.
#[inline] #[inline]
pub fn handle(&self) -> SocketHandle { pub fn handle(&self) -> SocketHandle {
self.handle self.meta.handle
}
/// Set the socket handle.
pub(in super) fn set_handle(&mut self, handle: SocketHandle) {
self.handle = handle;
} }
/// Return the IP version the socket is bound to. /// Return the IP version the socket is bound to.
@ -127,7 +122,7 @@ impl<'a, 'b> RawSocket<'a, 'b> {
pub fn send(&mut self, size: usize) -> Result<&mut [u8]> { pub fn send(&mut self, size: usize) -> Result<&mut [u8]> {
let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?; let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?;
net_trace!("{}:{}:{}: buffer to send {} octets", net_trace!("{}:{}:{}: buffer to send {} octets",
self.handle, self.ip_version, self.ip_protocol, self.meta.handle, self.ip_version, self.ip_protocol,
packet_buf.size); packet_buf.size);
Ok(packet_buf.as_mut()) Ok(packet_buf.as_mut())
} }
@ -149,7 +144,7 @@ impl<'a, 'b> RawSocket<'a, 'b> {
pub fn recv(&mut self) -> Result<&[u8]> { pub fn recv(&mut self) -> Result<&[u8]> {
let packet_buf = self.rx_buffer.dequeue_one()?; let packet_buf = self.rx_buffer.dequeue_one()?;
net_trace!("{}:{}:{}: receive {} buffered octets", net_trace!("{}:{}:{}: receive {} buffered octets",
self.handle, self.ip_version, self.ip_protocol, self.meta.handle, self.ip_version, self.ip_protocol,
packet_buf.size); packet_buf.size);
Ok(&packet_buf.as_ref()) Ok(&packet_buf.as_ref())
} }
@ -176,12 +171,12 @@ impl<'a, 'b> RawSocket<'a, 'b> {
debug_assert!(self.accepts(ip_repr)); debug_assert!(self.accepts(ip_repr));
let header_len = ip_repr.buffer_len(); let header_len = ip_repr.buffer_len();
let total_len = header_len + payload.len(); let total_len = header_len + payload.len();
let packet_buf = self.rx_buffer.enqueue_one_with(|buf| buf.resize(total_len))?; let packet_buf = self.rx_buffer.enqueue_one_with(|buf| buf.resize(total_len))?;
ip_repr.emit(&mut packet_buf.as_mut()[..header_len], &checksum_caps); ip_repr.emit(&mut packet_buf.as_mut()[..header_len], &checksum_caps);
packet_buf.as_mut()[header_len..].copy_from_slice(payload); packet_buf.as_mut()[header_len..].copy_from_slice(payload);
net_trace!("{}:{}:{}: receiving {} octets", net_trace!("{}:{}:{}: receiving {} octets",
self.handle, self.ip_version, self.ip_protocol, self.meta.handle, self.ip_version, self.ip_protocol,
packet_buf.size); packet_buf.size);
Ok(()) Ok(())
} }
@ -211,7 +206,7 @@ impl<'a, 'b> RawSocket<'a, 'b> {
} }
} }
let handle = self.handle; let handle = self.meta.handle;
let ip_protocol = self.ip_protocol; let ip_protocol = self.ip_protocol;
let ip_version = self.ip_version; let ip_version = self.ip_version;
self.tx_buffer.dequeue_one_with(|packet_buf| { self.tx_buffer.dequeue_one_with(|packet_buf| {

8
src/socket/set.rs
View File

@ -16,13 +16,9 @@ pub struct Item<'a, 'b: 'a> {
} }
/// A handle, identifying a socket in a set. /// A handle, identifying a socket in a set.
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy)] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default)]
pub struct Handle(usize); pub struct Handle(usize);
impl Handle {
pub(crate) const EMPTY: Handle = Handle(0);
}
impl fmt::Display for Handle { impl fmt::Display for Handle {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "#{}", self.0) write!(f, "#{}", self.0)
@ -56,7 +52,7 @@ impl<'a, 'b: 'a, 'c: 'a + 'b> Set<'a, 'b, 'c> {
mut socket: Socket<'b, 'c>) -> Handle { mut socket: Socket<'b, 'c>) -> Handle {
net_trace!("[{}]: adding", index); net_trace!("[{}]: adding", index);
let handle = Handle(index); let handle = Handle(index);
socket.set_handle(handle); socket.meta_mut().handle = handle;
*slot = Some(Item { socket: socket, refs: 1 }); *slot = Some(Item { socket: socket, refs: 1 });
handle handle
} }

71
src/socket/tcp.rs
View File

@ -7,7 +7,7 @@ use core::{cmp, fmt};
use {Error, Result}; use {Error, Result};
use phy::DeviceCapabilities; use phy::DeviceCapabilities;
use wire::{IpProtocol, IpRepr, IpAddress, IpEndpoint, TcpSeqNumber, TcpRepr, TcpControl}; use wire::{IpProtocol, IpRepr, IpAddress, IpEndpoint, TcpSeqNumber, TcpRepr, TcpControl};
use socket::{Socket, SocketHandle}; use socket::{Socket, SocketMeta, SocketHandle};
use storage::{Assembler, RingBuffer}; use storage::{Assembler, RingBuffer};
pub type SocketBuffer<'a> = RingBuffer<'a, u8>; pub type SocketBuffer<'a> = RingBuffer<'a, u8>;
@ -176,7 +176,7 @@ impl Timer {
/// attempts will be reset. /// attempts will be reset.
#[derive(Debug)] #[derive(Debug)]
pub struct TcpSocket<'a> { pub struct TcpSocket<'a> {
handle: SocketHandle, pub(crate) meta: SocketMeta,
state: State, state: State,
timer: Timer, timer: Timer,
assembler: Assembler, assembler: Assembler,
@ -237,7 +237,7 @@ impl<'a> TcpSocket<'a> {
} }
Socket::Tcp(TcpSocket { Socket::Tcp(TcpSocket {
handle: SocketHandle::EMPTY, meta: SocketMeta::default(),
state: State::Closed, state: State::Closed,
timer: Timer::default(), timer: Timer::default(),
assembler: Assembler::new(rx_buffer.capacity()), assembler: Assembler::new(rx_buffer.capacity()),
@ -263,12 +263,7 @@ impl<'a> TcpSocket<'a> {
/// Return the socket handle. /// Return the socket handle.
#[inline] #[inline]
pub fn handle(&self) -> SocketHandle { pub fn handle(&self) -> SocketHandle {
self.handle self.meta.handle
}
/// Set the socket handle.
pub(in super) fn set_handle(&mut self, handle: SocketHandle) {
self.handle = handle;
} }
/// Return the timeout duration. /// Return the timeout duration.
@ -614,7 +609,7 @@ impl<'a> TcpSocket<'a> {
if size > 0 { if size > 0 {
#[cfg(any(test, feature = "verbose"))] #[cfg(any(test, feature = "verbose"))]
net_trace!("{}:{}:{}: tx buffer: enqueueing {} octets (now {})", net_trace!("{}:{}:{}: tx buffer: enqueueing {} octets (now {})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
size, _old_length + size); size, _old_length + size);
} }
Ok(result) Ok(result)
@ -658,7 +653,7 @@ impl<'a> TcpSocket<'a> {
if size > 0 { if size > 0 {
#[cfg(any(test, feature = "verbose"))] #[cfg(any(test, feature = "verbose"))]
net_trace!("{}:{}:{}: rx buffer: dequeueing {} octets (now {})", net_trace!("{}:{}:{}: rx buffer: dequeueing {} octets (now {})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
size, _old_length - size); size, _old_length - size);
} }
Ok(result) Ok(result)
@ -702,7 +697,7 @@ impl<'a> TcpSocket<'a> {
if buffer.len() > 0 { if buffer.len() > 0 {
#[cfg(any(test, feature = "verbose"))] #[cfg(any(test, feature = "verbose"))]
net_trace!("{}:{}:{}: rx buffer: peeking at {} octets", net_trace!("{}:{}:{}: rx buffer: peeking at {} octets",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
buffer.len()); buffer.len());
} }
Ok(buffer) Ok(buffer)
@ -737,11 +732,11 @@ impl<'a> TcpSocket<'a> {
if self.state != state { if self.state != state {
if self.remote_endpoint.addr.is_unspecified() { if self.remote_endpoint.addr.is_unspecified() {
net_trace!("{}:{}: state={}=>{}", net_trace!("{}:{}: state={}=>{}",
self.handle, self.local_endpoint, self.meta.handle, self.local_endpoint,
self.state, state); self.state, state);
} else { } else {
net_trace!("{}:{}:{}: state={}=>{}", net_trace!("{}:{}:{}: state={}=>{}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
self.state, state); self.state, state);
} }
} }
@ -846,7 +841,7 @@ impl<'a> TcpSocket<'a> {
}) => { }) => {
net_debug!("{}:{}:{}: unacceptable RST (expecting RST|ACK) \ net_debug!("{}:{}:{}: unacceptable RST (expecting RST|ACK) \
in response to initial SYN", in response to initial SYN",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
return Err(Error::Dropped) return Err(Error::Dropped)
} }
(State::SynSent, &TcpRepr { (State::SynSent, &TcpRepr {
@ -854,7 +849,7 @@ impl<'a> TcpSocket<'a> {
}) => { }) => {
if ack_number != self.local_seq_no + 1 { if ack_number != self.local_seq_no + 1 {
net_debug!("{}:{}:{}: unacceptable RST|ACK in response to initial SYN", net_debug!("{}:{}:{}: unacceptable RST|ACK in response to initial SYN",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
return Err(Error::Dropped) return Err(Error::Dropped)
} }
} }
@ -867,7 +862,7 @@ impl<'a> TcpSocket<'a> {
// Every packet after the initial SYN must be an acknowledgement. // Every packet after the initial SYN must be an acknowledgement.
(_, &TcpRepr { ack_number: None, .. }) => { (_, &TcpRepr { ack_number: None, .. }) => {
net_debug!("{}:{}:{}: expecting an ACK", net_debug!("{}:{}:{}: expecting an ACK",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
return Err(Error::Dropped) return Err(Error::Dropped)
} }
// Every acknowledgement must be for transmitted but unacknowledged data. // Every acknowledgement must be for transmitted but unacknowledged data.
@ -876,7 +871,7 @@ impl<'a> TcpSocket<'a> {
if ack_number < self.local_seq_no { if ack_number < self.local_seq_no {
net_debug!("{}:{}:{}: duplicate ACK ({} not in {}...{})", net_debug!("{}:{}:{}: duplicate ACK ({} not in {}...{})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
ack_number, self.local_seq_no, self.local_seq_no + unacknowledged); ack_number, self.local_seq_no, self.local_seq_no + unacknowledged);
// FIXME: implement fast retransmit // FIXME: implement fast retransmit
return Err(Error::Dropped) return Err(Error::Dropped)
@ -884,7 +879,7 @@ impl<'a> TcpSocket<'a> {
if ack_number > self.local_seq_no + unacknowledged { if ack_number > self.local_seq_no + unacknowledged {
net_debug!("{}:{}:{}: unacceptable ACK ({} not in {}...{})", net_debug!("{}:{}:{}: unacceptable ACK ({} not in {}...{})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
ack_number, self.local_seq_no, self.local_seq_no + unacknowledged); ack_number, self.local_seq_no, self.local_seq_no + unacknowledged);
return Ok(Some(self.ack_reply(ip_repr, &repr))) return Ok(Some(self.ack_reply(ip_repr, &repr)))
} }
@ -908,7 +903,7 @@ impl<'a> TcpSocket<'a> {
if window_start == window_end && segment_start != segment_end { if window_start == window_end && segment_start != segment_end {
net_debug!("{}:{}:{}: non-zero-length segment with zero receive window, \ net_debug!("{}:{}:{}: non-zero-length segment with zero receive window, \
will only send an ACK", will only send an ACK",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
segment_in_window = false; segment_in_window = false;
} }
@ -916,7 +911,7 @@ impl<'a> TcpSocket<'a> {
(window_start <= segment_end && segment_end <= window_end)) { (window_start <= segment_end && segment_end <= window_end)) {
net_debug!("{}:{}:{}: segment not in receive window \ net_debug!("{}:{}:{}: segment not in receive window \
({}..{} not intersecting {}..{}), will send challenge ACK", ({}..{} not intersecting {}..{}), will send challenge ACK",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
segment_start, segment_end, window_start, window_end); segment_start, segment_end, window_start, window_end);
segment_in_window = false; segment_in_window = false;
} }
@ -953,7 +948,7 @@ impl<'a> TcpSocket<'a> {
if sent_fin && self.tx_buffer.len() + 1 == ack_len { if sent_fin && self.tx_buffer.len() + 1 == ack_len {
ack_len -= 1; ack_len -= 1;
net_trace!("{}:{}:{}: received ACK of FIN", net_trace!("{}:{}:{}: received ACK of FIN",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
ack_of_fin = true; ack_of_fin = true;
} }
} }
@ -968,7 +963,7 @@ impl<'a> TcpSocket<'a> {
// RSTs in SYN-RECEIVED flip the socket back to the LISTEN state. // RSTs in SYN-RECEIVED flip the socket back to the LISTEN state.
(State::SynReceived, TcpControl::Rst) => { (State::SynReceived, TcpControl::Rst) => {
net_trace!("{}:{}:{}: received RST", net_trace!("{}:{}:{}: received RST",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
self.local_endpoint.addr = self.listen_address; self.local_endpoint.addr = self.listen_address;
self.remote_endpoint = IpEndpoint::default(); self.remote_endpoint = IpEndpoint::default();
self.set_state(State::Listen); self.set_state(State::Listen);
@ -978,7 +973,7 @@ impl<'a> TcpSocket<'a> {
// RSTs in any other state close the socket. // RSTs in any other state close the socket.
(_, TcpControl::Rst) => { (_, TcpControl::Rst) => {
net_trace!("{}:{}:{}: received RST", net_trace!("{}:{}:{}: received RST",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
self.set_state(State::Closed); self.set_state(State::Closed);
self.local_endpoint = IpEndpoint::default(); self.local_endpoint = IpEndpoint::default();
self.remote_endpoint = IpEndpoint::default(); self.remote_endpoint = IpEndpoint::default();
@ -988,7 +983,7 @@ impl<'a> TcpSocket<'a> {
// SYN packets in the LISTEN state change it to SYN-RECEIVED. // SYN packets in the LISTEN state change it to SYN-RECEIVED.
(State::Listen, TcpControl::Syn) => { (State::Listen, TcpControl::Syn) => {
net_trace!("{}:{}: received SYN", net_trace!("{}:{}: received SYN",
self.handle, self.local_endpoint); self.meta.handle, self.local_endpoint);
self.local_endpoint = IpEndpoint::new(ip_repr.dst_addr(), repr.dst_port); self.local_endpoint = IpEndpoint::new(ip_repr.dst_addr(), repr.dst_port);
self.remote_endpoint = IpEndpoint::new(ip_repr.src_addr(), repr.src_port); self.remote_endpoint = IpEndpoint::new(ip_repr.src_addr(), repr.src_port);
// FIXME: use something more secure here // FIXME: use something more secure here
@ -1020,7 +1015,7 @@ impl<'a> TcpSocket<'a> {
// SYN|ACK packets in the SYN-SENT state change it to ESTABLISHED. // SYN|ACK packets in the SYN-SENT state change it to ESTABLISHED.
(State::SynSent, TcpControl::Syn) => { (State::SynSent, TcpControl::Syn) => {
net_trace!("{}:{}:{}: received SYN|ACK", net_trace!("{}:{}:{}: received SYN|ACK",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
self.local_endpoint = IpEndpoint::new(ip_repr.dst_addr(), repr.dst_port); self.local_endpoint = IpEndpoint::new(ip_repr.dst_addr(), repr.dst_port);
self.remote_seq_no = repr.seq_number + 1; self.remote_seq_no = repr.seq_number + 1;
self.remote_last_seq = self.local_seq_no + 1; self.remote_last_seq = self.local_seq_no + 1;
@ -1101,7 +1096,7 @@ impl<'a> TcpSocket<'a> {
_ => { _ => {
net_debug!("{}:{}:{}: unexpected packet {}", net_debug!("{}:{}:{}: unexpected packet {}",
self.handle, self.local_endpoint, self.remote_endpoint, repr); self.meta.handle, self.local_endpoint, self.remote_endpoint, repr);
return Err(Error::Dropped) return Err(Error::Dropped)
} }
} }
@ -1114,7 +1109,7 @@ impl<'a> TcpSocket<'a> {
// Dequeue acknowledged octets. // Dequeue acknowledged octets.
debug_assert!(self.tx_buffer.len() >= ack_len); debug_assert!(self.tx_buffer.len() >= ack_len);
net_trace!("{}:{}:{}: tx buffer: dequeueing {} octets (now {})", net_trace!("{}:{}:{}: tx buffer: dequeueing {} octets (now {})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
ack_len, self.tx_buffer.len() - ack_len); ack_len, self.tx_buffer.len() - ack_len);
self.tx_buffer.dequeue_allocated(ack_len); self.tx_buffer.dequeue_allocated(ack_len);
} }
@ -1136,13 +1131,13 @@ impl<'a> TcpSocket<'a> {
debug_assert!(self.assembler.total_size() == self.rx_buffer.capacity()); debug_assert!(self.assembler.total_size() == self.rx_buffer.capacity());
// Place payload octets into the buffer. // Place payload octets into the buffer.
net_trace!("{}:{}:{}: rx buffer: receiving {} octets at offset {}", net_trace!("{}:{}:{}: rx buffer: receiving {} octets at offset {}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
payload_len, payload_offset); payload_len, payload_offset);
self.rx_buffer.write_unallocated(payload_offset, repr.payload); self.rx_buffer.write_unallocated(payload_offset, repr.payload);
} }
Err(()) => { Err(()) => {
net_debug!("{}:{}:{}: assembler: too many holes to add {} octets at offset {}", net_debug!("{}:{}:{}: assembler: too many holes to add {} octets at offset {}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
payload_len, payload_offset); payload_len, payload_offset);
return Err(Error::Dropped) return Err(Error::Dropped)
} }
@ -1152,7 +1147,7 @@ impl<'a> TcpSocket<'a> {
debug_assert!(self.assembler.total_size() == self.rx_buffer.capacity()); debug_assert!(self.assembler.total_size() == self.rx_buffer.capacity());
// Enqueue the contiguous data octets in front of the buffer. // Enqueue the contiguous data octets in front of the buffer.
net_trace!("{}:{}:{}: rx buffer: enqueueing {} octets (now {})", net_trace!("{}:{}:{}: rx buffer: enqueueing {} octets (now {})",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
contig_len, self.rx_buffer.len() + contig_len); contig_len, self.rx_buffer.len() + contig_len);
self.rx_buffer.enqueue_unallocated(contig_len); self.rx_buffer.enqueue_unallocated(contig_len);
} }
@ -1160,7 +1155,7 @@ impl<'a> TcpSocket<'a> {
if !self.assembler.is_empty() { if !self.assembler.is_empty() {
// Print the ranges recorded in the assembler. // Print the ranges recorded in the assembler.
net_trace!("{}:{}:{}: assembler: {}", net_trace!("{}:{}:{}: assembler: {}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
self.assembler); self.assembler);
} }
@ -1236,13 +1231,13 @@ impl<'a> TcpSocket<'a> {
if self.timed_out(timestamp) { if self.timed_out(timestamp) {
// If a timeout expires, we should abort the connection. // If a timeout expires, we should abort the connection.
net_debug!("{}:{}:{}: timeout exceeded", net_debug!("{}:{}:{}: timeout exceeded",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
self.set_state(State::Closed); self.set_state(State::Closed);
} else if !self.seq_to_transmit() { } else if !self.seq_to_transmit() {
if let Some(retransmit_delta) = self.timer.should_retransmit(timestamp) { if let Some(retransmit_delta) = self.timer.should_retransmit(timestamp) {
// If a retransmit timer expired, we should resend data starting at the last ACK. // If a retransmit timer expired, we should resend data starting at the last ACK.
net_debug!("{}:{}:{}: retransmitting at t+{}ms", net_debug!("{}:{}:{}: retransmitting at t+{}ms",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
retransmit_delta); retransmit_delta);
self.remote_last_seq = self.local_seq_no; self.remote_last_seq = self.local_seq_no;
} }
@ -1343,7 +1338,7 @@ impl<'a> TcpSocket<'a> {
State::TimeWait => { State::TimeWait => {
if self.timer.should_close(timestamp) { if self.timer.should_close(timestamp) {
net_trace!("{}:{}:{}: TIME-WAIT timeout", net_trace!("{}:{}:{}: TIME-WAIT timeout",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
self.reset(); self.reset();
return Err(Error::Exhausted) return Err(Error::Exhausted)
} }
@ -1366,10 +1361,10 @@ impl<'a> TcpSocket<'a> {
// Trace a summary of what will be sent. // Trace a summary of what will be sent.
if is_keep_alive { if is_keep_alive {
net_trace!("{}:{}:{}: sending a keep-alive", net_trace!("{}:{}:{}: sending a keep-alive",
self.handle, self.local_endpoint, self.remote_endpoint); self.meta.handle, self.local_endpoint, self.remote_endpoint);
} else if repr.payload.len() > 0 { } else if repr.payload.len() > 0 {
net_trace!("{}:{}:{}: tx buffer: sending {} octets at offset {}", net_trace!("{}:{}:{}: tx buffer: sending {} octets at offset {}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
repr.payload.len(), self.remote_last_seq - self.local_seq_no); repr.payload.len(), self.remote_last_seq - self.local_seq_no);
} }
if repr.control != TcpControl::None || repr.payload.len() == 0 { if repr.control != TcpControl::None || repr.payload.len() == 0 {
@ -1384,7 +1379,7 @@ impl<'a> TcpSocket<'a> {
_ => "<unreachable>" _ => "<unreachable>"
}; };
net_trace!("{}:{}:{}: sending {}", net_trace!("{}:{}:{}: sending {}",
self.handle, self.local_endpoint, self.remote_endpoint, self.meta.handle, self.local_endpoint, self.remote_endpoint,
flags); flags);
} }

21
src/socket/udp.rs
View File

@ -3,7 +3,7 @@ use managed::Managed;
use {Error, Result}; use {Error, Result};
use wire::{IpProtocol, IpRepr, IpEndpoint, UdpRepr}; use wire::{IpProtocol, IpRepr, IpEndpoint, UdpRepr};
use socket::{Socket, SocketHandle}; use socket::{Socket, SocketMeta, SocketHandle};
use storage::{Resettable, RingBuffer}; use storage::{Resettable, RingBuffer};
/// A buffered UDP packet. /// A buffered UDP packet.
@ -59,7 +59,7 @@ pub type SocketBuffer<'a, 'b: 'a> = RingBuffer<'a, PacketBuffer<'b>>;
/// packet buffers. /// packet buffers.
#[derive(Debug)] #[derive(Debug)]
pub struct UdpSocket<'a, 'b: 'a> { pub struct UdpSocket<'a, 'b: 'a> {
handle: SocketHandle, pub(crate) meta: SocketMeta,
endpoint: IpEndpoint, endpoint: IpEndpoint,
rx_buffer: SocketBuffer<'a, 'b>, rx_buffer: SocketBuffer<'a, 'b>,
tx_buffer: SocketBuffer<'a, 'b>, tx_buffer: SocketBuffer<'a, 'b>,
@ -72,7 +72,7 @@ impl<'a, 'b> UdpSocket<'a, 'b> {
pub fn new(rx_buffer: SocketBuffer<'a, 'b>, pub fn new(rx_buffer: SocketBuffer<'a, 'b>,
tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> { tx_buffer: SocketBuffer<'a, 'b>) -> Socket<'a, 'b> {
Socket::Udp(UdpSocket { Socket::Udp(UdpSocket {
handle: SocketHandle::EMPTY, meta: SocketMeta::default(),
endpoint: IpEndpoint::default(), endpoint: IpEndpoint::default(),
rx_buffer: rx_buffer, rx_buffer: rx_buffer,
tx_buffer: tx_buffer, tx_buffer: tx_buffer,
@ -83,12 +83,7 @@ impl<'a, 'b> UdpSocket<'a, 'b> {
/// Return the socket handle. /// Return the socket handle.
#[inline] #[inline]
pub fn handle(&self) -> SocketHandle { pub fn handle(&self) -> SocketHandle {
self.handle self.meta.handle
}
/// Set the socket handle.
pub(in super) fn set_handle(&mut self, handle: SocketHandle) {
self.handle = handle;
} }
/// Return the bound endpoint. /// Return the bound endpoint.
@ -171,7 +166,7 @@ impl<'a, 'b> UdpSocket<'a, 'b> {
let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?; let packet_buf = self.tx_buffer.enqueue_one_with(|buf| buf.resize(size))?;
packet_buf.endpoint = endpoint; packet_buf.endpoint = endpoint;
net_trace!("{}:{}:{}: buffer to send {} octets", net_trace!("{}:{}:{}: buffer to send {} octets",
self.handle, self.endpoint, packet_buf.endpoint, size); self.meta.handle, self.endpoint, packet_buf.endpoint, size);
Ok(&mut packet_buf.as_mut()[..size]) Ok(&mut packet_buf.as_mut()[..size])
} }
@ -190,7 +185,7 @@ impl<'a, 'b> UdpSocket<'a, 'b> {
pub fn recv(&mut self) -> Result<(&[u8], IpEndpoint)> { pub fn recv(&mut self) -> Result<(&[u8], IpEndpoint)> {
let packet_buf = self.rx_buffer.dequeue_one()?; let packet_buf = self.rx_buffer.dequeue_one()?;
net_trace!("{}:{}:{}: receive {} buffered octets", net_trace!("{}:{}:{}: receive {} buffered octets",
self.handle, self.endpoint, self.meta.handle, self.endpoint,
packet_buf.endpoint, packet_buf.size); packet_buf.endpoint, packet_buf.size);
Ok((&packet_buf.as_ref(), packet_buf.endpoint)) Ok((&packet_buf.as_ref(), packet_buf.endpoint))
} }
@ -221,14 +216,14 @@ impl<'a, 'b> UdpSocket<'a, 'b> {
packet_buf.as_mut().copy_from_slice(repr.payload); packet_buf.as_mut().copy_from_slice(repr.payload);
packet_buf.endpoint = IpEndpoint { addr: ip_repr.src_addr(), port: repr.src_port }; packet_buf.endpoint = IpEndpoint { addr: ip_repr.src_addr(), port: repr.src_port };
net_trace!("{}:{}:{}: receiving {} octets", net_trace!("{}:{}:{}: receiving {} octets",
self.handle, self.endpoint, self.meta.handle, self.endpoint,
packet_buf.endpoint, packet_buf.size); packet_buf.endpoint, packet_buf.size);
Ok(()) Ok(())
} }
pub(crate) fn dispatch<F>(&mut self, emit: F) -> Result<()> pub(crate) fn dispatch<F>(&mut self, emit: F) -> Result<()>
where F: FnOnce((IpRepr, UdpRepr)) -> Result<()> { where F: FnOnce((IpRepr, UdpRepr)) -> Result<()> {
let handle = self.handle; let handle = self.handle();
let endpoint = self.endpoint; let endpoint = self.endpoint;
let ttl = self.ttl.unwrap_or(64); let ttl = self.ttl.unwrap_or(64);
self.tx_buffer.dequeue_one_with(|packet_buf| { self.tx_buffer.dequeue_one_with(|packet_buf| {