node: Constrain worker channel size

To avoid buffering large amounts of data in the process, we set the
worker channel size to `64`. Keep in mind that this is one
`ChannelEvent`, not one byte. `ChannelEvent::Data` can already contain
an arbitrary amount of data via its `Vec<u8>`, but in practice the
maximum is 8192, due to the use of `io::copy`.

We also remove an unused function, and move the flushing to the
`ChannelFlushWriter`.
This commit is contained in:
cloudhead 2024-06-26 13:36:00 +02:00
parent 38592955b2
commit a79ca5e8fc
No known key found for this signature in database
2 changed files with 5 additions and 20 deletions

View File

@ -10,7 +10,9 @@ use crate::runtime::Handle;
use crate::wire::StreamId;
/// Maximum size of channel used to communicate with a worker.
pub const MAX_WORKER_CHANNEL_SIZE: usize = 4096;
/// Note that as long as we're using [`std::io::copy`] to copy data from the
/// upload-pack's stdout, the data chunks are of a maximum size of 8192 bytes.
pub const MAX_WORKER_CHANNEL_SIZE: usize = 64;
/// A reader and writer pair that can be used in the fetch protocol.
///
@ -184,7 +186,7 @@ struct ChannelWriter<T = Vec<u8>> {
/// Wraps a [`ChannelWriter`] alongside the associated [`Handle`] and [`NodeId`].
///
/// This allows the channel to [`Write::flush`] when calling
/// [`Write::write_all`], which is necessary to signal to the
/// [`Write::write`], which is necessary to signal to the
/// controller to send the wire data.
pub struct ChannelFlushWriter<T = Vec<u8>> {
writer: ChannelWriter<T>,
@ -206,29 +208,13 @@ impl Write for ChannelFlushWriter<Vec<u8>> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let n = buf.len();
self.writer.send(buf.to_vec())?;
self.flush()?;
Ok(n)
}
fn flush(&mut self) -> io::Result<()> {
self.handle.flush(self.remote, self.stream)
}
fn write_all(&mut self, mut buf: &[u8]) -> io::Result<()> {
while !buf.is_empty() {
match self.write(buf) {
Ok(0) => {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"failed to write whole buffer",
));
}
Ok(n) => buf = &buf[n..],
Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
}
}
self.flush()
}
}
impl<T: AsRef<[u8]>> ChannelWriter<T> {

View File

@ -201,7 +201,6 @@ where
{
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let n = self.send.write(buf)?;
self.send.flush()?;
self.emit(buf);
Ok(n)
}