Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor to use new ResourceId from reactor crate #36

Merged
merged 4 commits into from
Jan 10, 2024
Merged

Refactor to use new ResourceId from reactor crate #36

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0fc9bb5
refactor to use new ResourceId from reactor crate
dr-orlovsky Dec 12, 2023
19ea32e
add information about raw fd to the SessionEvent::Established
dr-orlovsky Dec 19, 2023
f86a594
remove unnecessary set_resource_id method
dr-orlovsky Dec 20, 2023
b20dd94
chore: update dependencies
dr-orlovsky Jan 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 4 additions & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ readme = "README.md"

[dependencies]
amplify = "4.0.0"
io-reactor = { version = "0.2.0", optional = true }
io-reactor = { version = "0.2.1", optional = true }
cyphernet = { version = "0.4.1", features = ["ed25519", "pem", "multibase", "noise_sha2", "noise_x25519", "noise-framework", "mixnets", "dns"] }
mio = { version = "0.8.6", optional = true }
socket2 = { version = "0.5.2", optional = true, features = ["all"] }
Expand All @@ -38,3 +38,6 @@ log = ["log_crate", "io-reactor/log"]
[package.metadata.docs.rs]
all-features = true
rustc-args = ["--cfg", "docsrs"]

[patch.crates-io]
io-reactor = { git = "https://github.com/rust-amplify/io-reactor", branch = "resourceid" }
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note to remove this once io-reactor patch is merged

49 changes: 35 additions & 14 deletions src/resource.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ use std::time::Duration;
use std::{fmt, io, net};

use reactor::poller::IoType;
use reactor::{Io, Resource, WriteAtomic, WriteError};
use reactor::{Io, Resource, ResourceId, WriteAtomic, WriteError};

use crate::{Direction, NetConnection, NetListener, NetSession, READ_BUFFER_SIZE};

Expand Down Expand Up @@ -151,11 +151,8 @@ impl<L: NetListener<Stream = S::Connection>, S: NetSession> NetAccept<S, L> {
impl<L: NetListener<Stream = S::Connection>, S: NetSession> Resource for NetAccept<S, L>
where S: Send
{
type Id = net::SocketAddr;
type Event = ListenerEvent<S>;

fn id(&self) -> Self::Id { self.listener.local_addr() }

fn interests(&self) -> IoType { IoType::read_only() }

fn handle_io(&mut self, io: Io) -> Option<Self::Event> {
Expand Down Expand Up @@ -202,6 +199,11 @@ pub enum TransportState {
Terminated,
}

/// Error indicating that method [`NetTransport::set_resource_id`] was called more than once.
#[derive(Copy, Clone, Eq, PartialEq, Debug, Display, Error)]
#[display("an attempt to re-assign resource id to {new} for net transport {current}.")]
pub struct ResIdReassigned { current: ResourceId, new: ResourceId }

/// Net transport is an adaptor around specific [`NetSession`] (implementing
/// session management, including optional handshake, encoding etc) to be used
/// as a transport resource in a [`reactor::Reactor`].
Expand All @@ -213,6 +215,8 @@ pub struct NetTransport<S: NetSession> {
write_intent: bool,
read_buffer: Box<[u8; HEAP_BUFFER_SIZE]>,
write_buffer: VecDeque<u8>,
/// Resource id assigned by the reactor
id: Option<ResourceId>
}

impl<S: NetSession> Display for NetTransport<S> {
Expand Down Expand Up @@ -256,6 +260,7 @@ impl<S: NetSession> NetTransport<S> {
write_intent: true,
read_buffer: Box::new([0u8; READ_BUFFER_SIZE]),
write_buffer: empty!(),
id: None,
})
}

Expand Down Expand Up @@ -295,9 +300,29 @@ impl<S: NetSession> NetTransport<S> {
write_intent: false,
read_buffer: Box::new([0u8; READ_BUFFER_SIZE]),
write_buffer: empty!(),
id: None,
})
}

pub fn display(&self) -> impl Display {
match self.id {
None => self.session.display(),
Some(id) => id.to_string()
}
}

pub fn resource_id(&self) -> Option<ResourceId> {
self.id
}

pub fn set_resource_id(&mut self, id: ResourceId) -> Result<(), ResIdReassigned> {
if let Some(current) = self.id {
return Err(ResIdReassigned { current, new: id })
}
self.id = Some(id);
Ok(())
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Where is this used?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the logging above

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean inside the new display method, which is called by each logging line in the file

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean set_resource_id


pub fn state(&self) -> TransportState { self.state }
pub fn is_active(&self) -> bool { self.state == TransportState::Active }

Expand Down Expand Up @@ -345,7 +370,7 @@ impl<S: NetSession> NetTransport<S> {
.contains(&err.kind()) =>
{
#[cfg(feature = "log")]
log::warn!(target: "transport", "Resource {} was not able to consume any data even though it has announced its write readiness", self.id());
log::warn!(target: "transport", "Resource {} was not able to consume any data even though it has announced its write readiness", self.display());
self.write_intent = true;
None
}
Expand Down Expand Up @@ -379,7 +404,7 @@ impl<S: NetSession> NetTransport<S> {
fn flush_buffer(&mut self) -> io::Result<()> {
let orig_len = self.write_buffer.len();
#[cfg(feature = "log")]
log::trace!(target: "transport", "Resource {} is flushing its buffer of {orig_len} bytes", self.id());
log::trace!(target: "transport", "Resource {} is flushing its buffer of {orig_len} bytes", self.display());
let len =
self.session.write(self.write_buffer.make_contiguous()).or_else(|err| {
match err.kind() {
Expand All @@ -388,23 +413,23 @@ impl<S: NetSession> NetTransport<S> {
| io::ErrorKind::WriteZero
| io::ErrorKind::Interrupted => {
#[cfg(feature = "log")]
log::warn!(target: "transport", "Resource {} kernel buffer is fulled (system message is '{err}')", self.id());
log::warn!(target: "transport", "Resource {} kernel buffer is fulled (system message is '{err}')", self.display());
Ok(0)
},
_ => {
#[cfg(feature = "log")]
log::error!(target: "transport", "Resource {} failed write operation with message '{err}'", self.id());
log::error!(target: "transport", "Resource {} failed write operation with message '{err}'", self.display());
Err(err)
},
}
})?;
if orig_len > len {
#[cfg(feature = "log")]
log::debug!(target: "transport", "Resource {} was able to consume only a part of the buffered data ({len} of {orig_len} bytes)", self.id());
log::debug!(target: "transport", "Resource {} was able to consume only a part of the buffered data ({len} of {orig_len} bytes)", self.display());
self.write_intent = true;
} else {
#[cfg(feature = "log")]
log::trace!(target: "transport", "Resource {} was able to consume all of the buffered data ({len} of {orig_len} bytes)", self.id());
log::trace!(target: "transport", "Resource {} was able to consume all of the buffered data ({len} of {orig_len} bytes)", self.display());
self.write_intent = false;
}
self.write_buffer.drain(..len);
Expand All @@ -413,12 +438,8 @@ impl<S: NetSession> NetTransport<S> {
}

impl<S: NetSession> Resource for NetTransport<S> {
// TODO: Use S::Artifact instead
type Id = RawFd;
type Event = SessionEvent<S>;

fn id(&self) -> Self::Id { self.session.as_connection().as_raw_fd() }

fn interests(&self) -> IoType {
match self.state {
TransportState::Init => IoType::write_only(),
Expand Down
18 changes: 9 additions & 9 deletions src/tunnel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,8 @@ impl<S: NetSession> Tunnel<S> {

let int_fd = stream.as_raw_fd();
let ext_fd = self.session.as_connection().as_raw_fd();
poller.register(&int_fd, IoType::read_only());
poller.register(&ext_fd, IoType::read_only());
let int_id = poller.register(&int_fd, IoType::read_only());
let ext_id = poller.register(&ext_fd, IoType::read_only());

let mut in_buf = VecDeque::<u8>::new();
let mut out_buf = VecDeque::<u8>::new();
Expand Down Expand Up @@ -116,7 +116,7 @@ impl<S: NetSession> Tunnel<S> {
log::warn!(target: "tunnel", "Tunnel {listener_addr} timed out with client {socket_addr}");
return Err(io::ErrorKind::TimedOut.into());
}
while let Some((fd, res)) = poller.next() {
while let Some((id, res)) = poller.next() {
let ev = match res {
Ok(ev) => ev,
Err(IoFail::Connectivity(code)) => {
Expand All @@ -130,7 +130,7 @@ impl<S: NetSession> Tunnel<S> {
return Err(io::ErrorKind::BrokenPipe.into());
}
};
if fd == int_fd {
if id == int_id {
if ev.write {
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "attempting to write {} bytes received from the remote {socket_addr}", in_buf.len());
Expand All @@ -140,7 +140,7 @@ impl<S: NetSession> Tunnel<S> {
in_buf.drain(..written);
in_count += written;
if in_buf.is_empty() {
poller.set_interest(&int_fd, IoType::read_only());
poller.set_interest(int_id, IoType::read_only());
}
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "{socket_addr} received {written} bytes from local out of {} buffered", in_buf.len());
Expand All @@ -152,12 +152,12 @@ impl<S: NetSession> Tunnel<S> {

handle!(stream.read(&mut buf), |read| {
out_buf.extend(&buf[..read]);
poller.set_interest(&ext_fd, IoType::read_write());
poller.set_interest(ext_id, IoType::read_write());
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "{socket_addr} read {read} bytes from local ({} total in the buffer)", out_buf.len());
});
}
} else if fd == ext_fd {
} else if id == ext_id {
if ev.write {
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "attempting to write {} bytes received from {socket_addr} to remote", out_buf.len());
Expand All @@ -167,7 +167,7 @@ impl<S: NetSession> Tunnel<S> {
out_buf.drain(..written);
out_count += written;
if out_buf.is_empty() {
poller.set_interest(&ext_fd, IoType::read_only());
poller.set_interest(ext_id, IoType::read_only());
}
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "{socket_addr} sent {written} bytes to remote out of {} buffered", out_buf.len());
Expand All @@ -179,7 +179,7 @@ impl<S: NetSession> Tunnel<S> {

handle!(self.session.read(&mut buf), |read| {
in_buf.extend(&buf[..read]);
poller.set_interest(&int_fd, IoType::read_write());
poller.set_interest(int_id, IoType::read_write());
#[cfg(feature = "log")]
log::trace!(target: "tunnel", "{socket_addr} read {read} bytes from remote ({} total in the buffer)", in_buf.len());
});
Expand Down
Loading