1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
use std::{collections::hash_map::Entry, sync::Arc};
use timestamped_socket::interface::InterfaceName;
use crate::{
ConnectedSocket, NetworkManagerData, OpenSocket, PACKET_BUFFER_SIZE, PtpAddressFamily,
SocketData, TimestampSource,
};
/// A handle for a network interface
///
/// Describes the configured network interface and way of timestamping for its
/// sockets. It also ensures the requested timestamping clock remains available
/// on the interface during its existence.
pub struct Interface<A: PtpAddressFamily> {
pub(crate) state: Arc<NetworkManagerData<A>>,
pub(crate) name: Option<InterfaceName>,
pub(crate) timestamp_source: TimestampSource,
}
impl<A: PtpAddressFamily> Clone for Interface<A> {
fn clone(&self) -> Self {
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut interfaces = self.state.interfaces.write().unwrap();
// The interface will always be present because of the refcounting.
let entry = interfaces.get_mut(&self.name).unwrap();
// An overflowing reference count is an error condition from which we
// cannot reasonably recover. A panic here is the best solution.
entry.refcount = entry.refcount.checked_add(1).unwrap();
Self {
state: self.state.clone(),
name: self.name,
timestamp_source: self.timestamp_source,
}
}
}
impl<A: PtpAddressFamily> Drop for Interface<A> {
fn drop(&mut self) {
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut interfaces = self.state.interfaces.write().unwrap();
// The interface will always be present because of the refcounting.
let entry = interfaces.get_mut(&self.name).unwrap();
// An underflowing refcount should never happen. If it does, something
// is seriously wrong and aborting is the best solution.
entry.refcount = entry.refcount.checked_sub(1).unwrap();
if entry.refcount == 0 {
interfaces.remove(&self.name);
}
}
}
impl<A: PtpAddressFamily> Interface<A> {
/// Create an open socket intended for listening to the network.
///
/// Creates a socket that is intended to receive traffic from any remote,
/// potentially providing responses to that traffic.
#[must_use]
#[expect(
clippy::missing_panics_doc,
reason = "Function will only panic if there is an implementation bug in this crate."
)]
pub fn listen_socket(&self) -> OpenSocket<A> {
let (socket_channel_tx, socket_channel_rx) = tokio::sync::mpsc::channel(PACKET_BUFFER_SIZE);
let socket = SocketData {
remote_filter: None,
local_filter: None,
interface_filter: self.name,
timestamp_source: self.timestamp_source,
message_channel: socket_channel_tx,
};
{
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut interfaces = self.state.interfaces.write().unwrap();
// The interface will always be present because of the refcounting.
let entry = interfaces.get_mut(&self.name).unwrap();
// An overflowing reference count is an error condition from which we
// cannot reasonably recover. A panic here is the best solution.
entry.refcount = entry.refcount.checked_add(1).unwrap();
}
let socket_id = {
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut sockets = self.state.sockets.write().unwrap();
loop {
let socket_id = self
.state
.next_socket_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if let Entry::Vacant(vacant_entry) = sockets.entry(socket_id) {
vacant_entry.insert(socket);
break socket_id;
}
}
};
OpenSocket {
state: self.state.clone(),
socket_id,
packet_receiver: socket_channel_rx,
}
}
/// Create a socket connected to a specific remote address.
///
/// This creates a socket intended for use in communicating with a single
/// remote party over the interface. The `remote` address specifies the
/// network address of that party.
///
/// The `local` address specifies which address should be used as the
/// sender address on the messages. If this is not specified, the operating
/// system will select a send address automatically.
#[expect(
clippy::missing_panics_doc,
reason = "Function will only panic if there is an implementation bug in this crate."
)]
pub fn connected_socket(&self, local: Option<A>, remote: A) -> ConnectedSocket<A> {
let (socket_channel_tx, socket_channel_rx) = tokio::sync::mpsc::channel(PACKET_BUFFER_SIZE);
let socket = SocketData {
remote_filter: Some(remote),
local_filter: local,
interface_filter: self.name,
timestamp_source: self.timestamp_source,
message_channel: socket_channel_tx,
};
{
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut interfaces = self.state.interfaces.write().unwrap();
// The interface will always be present because of the refcounting.
let entry = interfaces.get_mut(&self.name).unwrap();
// An overflowing reference count is an error condition from which we
// cannot reasonably recover. A panic here is the best solution.
entry.refcount = entry.refcount.checked_add(1).unwrap();
}
let socket_id = {
// The mutex can only be poisoned from an earlier panic. It is ok for
// us to propagate that to all the threads.
let mut sockets = self.state.sockets.write().unwrap();
loop {
let socket_id = self
.state
.next_socket_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if let Entry::Vacant(vacant_entry) = sockets.entry(socket_id) {
vacant_entry.insert(socket);
break socket_id;
}
}
};
ConnectedSocket {
state: self.state.clone(),
socket_id,
packet_receiver: socket_channel_rx,
}
}
}