Add windowsize option

README.md

@@ -39,18 +39,26 @@ To enable a client with IP `X.X.X.X` to receive files from a remote NBD disk, cr
 
 ---
 
+If no directory named `<tftp_root>/x.x.x.x` or corresponding NBD config `<tftp_root>/x.x.x.x.nbd>` is found, the system attempts to read the requested file from `<tftp_root>/default>`. This allows all peers to be served with a single file or enables RTFTP to function as a standard TFTP server.
+
+
+Additionally, RTFTP supports proactive setup of NBD connections upon the appearance of an NBD configuration file by utilizing [**inotify**](https://man7.org/linux/man-pages/man7/inotify.7.html) subsystem. With this approach, the remote filesystem is already up and running before the first TFTP request arrives.
+
+---
+
 ## Example
 
 TFTP root directory layout:
 
 ```
 tftp_root/
 ├── 192.168.10.10/
-│   ├── efi/
-│   │   └── grubnetaa64.efi.signed
 │   └── grub/
 │       └── grub.cfg
-└── 192.168.10.10.nbd
+├── 192.168.10.10.nbd
+└── default/
+    └── efi/
+        └── grubnetaa64.efi.signed
 ```
 
 Contents of `192.168.10.10.nbd`:
@@ -74,15 +82,17 @@ Contents of `192.168.10.10.nbd`:
 
 In this example:
 
-- The client with IP `192.168.10.10` will receive `efi/grubnetaa64.efi.signed` and `grub/grub.cfg` from the **local filesystem**.
-- Any other files will be retrieved from the **remote NBD disk** at `nbd://10.10.10.10:25000/server_root` from inside `/boot` directory from the **first** partition.
-
+- The client with IP `192.168.10.10` will receive `efi/grubnetaa64.efi.signed` from the **local filesystem** from the `tftp_root/default` directory
+- The client with IP `192.168.10.10` will `grub/grub.cfg` from the **local filesystem** from a specific `tftp_root/192.168.10.10` directory.
+- Any other files will be retrieved by the client with IP `192.168.10.10` from the **remote NBD disk** at `nbd://10.10.10.10:25000/server_root` from inside `/boot` directory from the **first** partition.
+- Clients with any other IPs will be able to download only `efi/grubnetaa64.efi.signed` from the `tftp_root/192.168.10.10` directory. 
 ---
 
 ### Notes:
 
 - Only Read Request (RRQ) is supported.
 - If a file exists in both the local directory and the NBD-based filesystem, the **local file takes precedence**.
+- If a file exists in both the `default` directory and a client directory, the latter is downloaded.
 - Initial setup of the virtual NBD filesystem takes **1.5 to 3 seconds**, so the first request usually need to be retried automatically by the client.
 - The NBD disk is either:
   - Connected proactively when config is created to avoid the first read request delay.
@@ -92,6 +102,7 @@ In this example:
     - timeout 
     - blksize
     - tsize
+    - windowsize
 - The daemon is intended to run without root privileges. To allow RTFTP to bind to UDP port 69, one of following workarounds may be applied:
     - Add **CAP_NET_BIND_SERVICE** capability to RTFTP: `setcap 'cap_net_bind_service=+ep' /path/to/rtftp`
     - Start RTFTP via `authbind` with port 69 allowed for the RTFTP user: `touch /etc/authbind/byport/69 && chown <rtftp_user>:<rtftp_group> /etc/authbind/byport/69`

src/cursor/mod.rs

@@ -70,7 +70,8 @@ impl<'a> WriteCursor<'a> {
         if end_index > self.buffer.len() {
             return Err(BufferError::new("Too little data left to write u16"));
         }
-        self.buffer[self.offset..end_index].copy_from_slice(&value.to_be_bytes());
+        self.buffer[self.offset] = ((value & 0xFF00) >> 8) as u8;
+        self.buffer[self.offset + 1] = (value & 0xFF) as u8;
         self.offset = end_index;
         Ok(self.offset)
     }

src/datagram_stream.rs

@@ -0,0 +1,75 @@
+use std::fmt;
+use std::fmt::{Debug, Display, Formatter};
+use std::io::ErrorKind;
+use std::net::SocketAddr;
+use tokio::net::UdpSocket;
+
+pub(super) struct DatagramStream {
+    local_socket: UdpSocket,
+    peer_address: SocketAddr,
+    display: String,
+}
+
+impl DatagramStream {
+    pub(super) fn new(local_socket: UdpSocket, peer_address: SocketAddr) -> Self {
+        let local_address = local_socket.local_addr().unwrap();
+        let local_ip = local_address.ip().to_string();
+        let local_port = local_address.port().to_string();
+        let remote_ip = peer_address.ip().to_string();
+        let remote_port = peer_address.port().to_string();
+        let display = format!("{local_ip}:{local_port} <=> {remote_ip}:{remote_port}");
+        Self {
+            local_socket,
+            peer_address,
+            display,
+        }
+    }
+
+    pub(super) fn remote_port(&self) -> u16 {
+        self.peer_address.port()
+    }
+
+    pub(super) async fn send(&self, buffer: &[u8]) -> std::io::Result<()> {
+        match self.local_socket.send_to(buffer, self.peer_address).await {
+            Ok(sent) => {
+                if sent != buffer.len() {
+                    Err(ErrorKind::ConnectionReset.into())
+                } else {
+                    Ok(())
+                }
+            }
+            Err(error) => Err(error),
+        }
+    }
+
+    pub(super) async fn recv(&self, buffer: &mut [u8], min_size: usize) -> std::io::Result<usize> {
+        loop {
+            match self.local_socket.recv_from(buffer).await {
+                Ok((recv_size, remote_address)) => {
+                    if remote_address != self.peer_address {
+                        eprintln!(
+                            "{self}: Ignore datagram {recv_size} long from alien {remote_address}"
+                        );
+                    } else if recv_size < min_size {
+                        eprintln!("{self}: Ignore runt datagram {recv_size} long");
+                    } else {
+                        return Ok(recv_size);
+                    }
+                }
+                Err(error) => return Err(error),
+            }
+        }
+    }
+}
+
+impl Debug for DatagramStream {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "<{}>", self.display)
+    }
+}
+
+impl Display for DatagramStream {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "<{}>", self.display)
+    }
+}

src/guestfs/mod.rs

@@ -72,14 +72,6 @@ struct guestfs_stat {
     ctime: i64,
 }
 
-impl Drop for guestfs_stat {
-    fn drop(&mut self) {
-        unsafe {
-            guestfs_free_stat(self);
-        }
-    }
-}
-
 #[link(name = "guestfs")]
 unsafe extern "C" {
     fn guestfs_create() -> *const guestfs_h;
@@ -223,9 +215,13 @@ impl GuestFS {
         }
     }
 
-    pub(super) fn add_qemu_option(&self, key: &str, value: &str) -> Result<(), GuestFSError> {
-        let c_str_key = CString::new(key).expect("CString::new failed");
-        let c_str_value = CString::new(value).expect("CString::new failed");
+    pub(super) fn add_qemu_option<S: AsRef<str>>(
+        &self,
+        key: S,
+        value: S,
+    ) -> Result<(), GuestFSError> {
+        let c_str_key = CString::new(key.as_ref()).expect("CString::new failed");
+        let c_str_value = CString::new(value.as_ref()).expect("CString::new failed");
         if unsafe { guestfs_config(self.handle, c_str_key.as_ptr(), c_str_value.as_ptr()) } == 0 {
             Ok(())
         } else {
@@ -270,9 +266,13 @@ impl GuestFS {
         Ok(partitions_list)
     }
 
-    pub(super) fn mount_ro(&self, device: &str, mountpoint: &str) -> Result<(), GuestFSError> {
-        let c_str_device = CString::new(device).expect("CString::new failed");
-        let c_str_mountpoint = CString::new(mountpoint).expect("CString::new failed");
+    pub(super) fn mount_ro<S: AsRef<str>>(
+        &self,
+        device: S,
+        mountpoint: S,
+    ) -> Result<(), GuestFSError> {
+        let c_str_device = CString::new(device.as_ref()).expect("CString::new failed");
+        let c_str_mountpoint = CString::new(mountpoint.as_ref()).expect("CString::new failed");
         if unsafe {
             guestfs_mount_ro(
                 self.handle,
@@ -287,8 +287,8 @@ impl GuestFS {
         }
     }
 
-    pub(super) fn get_size(&self, path: &str) -> Result<usize, GuestFSError> {
-        let c_str_path = CString::new(path).expect("CString::new failed");
+    pub(super) fn get_size<S: AsRef<str>>(&self, path: S) -> Result<usize, GuestFSError> {
+        let c_str_path = CString::new(path.as_ref()).expect("CString::new failed");
         let size = unsafe {
             let result = guestfs_stat(self.handle, c_str_path.as_ptr());
             if result.is_null() {
@@ -301,8 +301,8 @@ impl GuestFS {
         Ok(size as usize)
     }
 
-    pub(super) fn set_append(&self, string: &str) -> Result<(), GuestFSError> {
-        let c_str = CString::new(string).expect("CString::new failed");
+    pub(super) fn set_append<S: AsRef<str>>(&self, string: S) -> Result<(), GuestFSError> {
+        let c_str = CString::new(string.as_ref()).expect("CString::new failed");
         let result = unsafe { guestfs_set_append(self.handle, c_str.as_ptr()) };
         if result == 0 {
             Ok(())
@@ -311,8 +311,12 @@ impl GuestFS {
         }
     }
 
-    pub(super) fn read_chunk(&self, path: &str, offset: usize) -> Result<Vec<u8>, GuestFSError> {
-        let c_str_path = CString::new(path).expect("CString::new failed");
+    pub(super) fn read_chunk<S: AsRef<str>>(
+        &self,
+        path: S,
+        offset: usize,
+    ) -> Result<Vec<u8>, GuestFSError> {
+        let c_str_path = CString::new(path.as_ref()).expect("CString::new failed");
         unsafe {
             let mut size_r: libc::size_t = 0;
             let read_buffer = guestfs_pread(

src/main.rs

@@ -1,4 +1,9 @@
+#[cfg(windows)]
+compile_error!(
+    "This project does not support building on Windows due to its reliance on libguestfs and inotify."
+);
 mod cursor;
+mod datagram_stream;
 mod fs;
 mod fs_watch;
 mod guestfs;

src/messages/mod.rs

@@ -127,9 +127,9 @@ impl OptionsAcknowledge {
         }
     }
 
-    pub(super) fn serialize(&self, buffer: &mut [u8]) -> Result<(usize, u16), BufferError> {
+    pub(super) fn serialize(&self, buffer: &mut [u8]) -> Result<usize, BufferError> {
         if buffer.is_empty() {
-            return Ok((0, 0));
+            return Ok(0);
         }
         let mut datagram = WriteCursor::new(buffer);
         datagram.put_ushort(OACK)?;
@@ -141,7 +141,7 @@ impl OptionsAcknowledge {
             }
             offset
         };
-        Ok((offset, 0))
+        Ok(offset)
     }
     pub fn push(&mut self, option: (String, String)) {
         self.options.push(option)

src/options/mod.rs

@@ -14,9 +14,13 @@ static TIMEOUT: &str = "timeout";
 
 static BLKSIZE: &str = "blksize";
 
+const WINDOW_SIZE: &str = "windowsize";
+
 const BLOCK_SIZE_LIMIT: usize = u16::MAX as usize;
 
-const ACK_TIMEOUT_LIMIT: usize = 60;
+const ACK_TIMEOUT_LIMIT: usize = 255;
+
+const WINDOW_SIZE_LIMIT: usize = u16::MAX as usize;
 
 #[derive(Clone)]
 pub(super) struct Blksize {
@@ -28,13 +32,10 @@ impl Blksize {
         if let Some(block_size_string) = options.get(BLKSIZE)
             && let Ok(block_size) = block_size_string.parse::<usize>()
         {
-            if block_size < BLOCK_SIZE_LIMIT {
+            if (8..=BLOCK_SIZE_LIMIT).contains(&block_size) {
                 return Some(Self { block_size });
             } else {
-                eprintln!(
-                    "Requested block size {block_size} exceeds \
-                        maximum allowed block size {BLOCK_SIZE_LIMIT}"
-                );
+                eprintln!("Requested {block_size} doesn't fit in range 8 .. ={BLOCK_SIZE_LIMIT}");
             }
         }
         None
@@ -44,16 +45,8 @@ impl Blksize {
         (String::from(BLKSIZE), self.block_size.to_string())
     }
 
-    pub(super) fn is_last(&self, chunk_size: usize) -> bool {
-        chunk_size < self.block_size
-    }
-
-    pub(super) fn read_chunk(
-        &self,
-        opened_file: &mut dyn OpenedFile,
-        buffer: &mut [u8],
-    ) -> Result<usize, FileError> {
-        opened_file.read_to(&mut buffer[..self.block_size])
+    pub(super) fn get_size(&self) -> usize {
+        self.block_size
     }
 }
 
@@ -86,11 +79,11 @@ impl AckTimeout {
         if let Some(timeout_string) = options.get(TIMEOUT)
             && let Ok(timeout) = timeout_string.parse::<usize>()
         {
-            if timeout <= ACK_TIMEOUT_LIMIT {
+            if (1..=ACK_TIMEOUT_LIMIT).contains(&timeout) {
                 return Some(Self { timeout });
             } else {
                 eprintln!(
-                    "Requested timeout {timeout} exceeds maximum allowed {ACK_TIMEOUT_LIMIT}"
+                    "Requested timeout {timeout} doesn't fit in range 1 .. ={ACK_TIMEOUT_LIMIT}"
                 );
             }
         }
@@ -126,3 +119,41 @@ impl TSize {
         (String::from(TSIZE), self.file_size.to_string())
     }
 }
+
+pub(super) struct WindowSize(usize);
+
+impl Display for WindowSize {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "[blocks_count: {}]", self.0)
+    }
+}
+
+impl WindowSize {
+    pub(super) fn find_in(options: &HashMap<String, String>) -> Option<Self> {
+        if let Some(window_size) = options.get(WINDOW_SIZE)
+            && let Ok(window_size) = window_size.parse::<usize>()
+        {
+            if (1..=WINDOW_SIZE_LIMIT).contains(&window_size) {
+                return Some(Self(window_size));
+            } else {
+                eprintln!(
+                    "Requested window size {window_size} doesn't fit in range 1 .. ={WINDOW_SIZE_LIMIT}"
+                );
+            }
+        }
+        None
+    }
+
+    pub(super) fn get_size(&self) -> usize {
+        self.0
+    }
+    pub(super) fn as_key_pair(&self) -> (String, String) {
+        (String::from(WINDOW_SIZE), self.0.to_string())
+    }
+}
+
+impl Default for WindowSize {
+    fn default() -> Self {
+        Self(1)
+    }
+}