Implementation of the IP loopback. (#19)

This patch lets users connect to `ip:127.0.0.1/*`. Bytes written to
this connection can only be read from the same connection.

schemes/ipd/src/main.rs

@@ -13,7 +13,10 @@
 //! To open a IP connection, use `ip:[host]/protocol`.
 //!
 //! * If `host` is specified, it must be an ipv4 number (e.g. `192.168.0.1`)
-//! and the connection may be used immediately to send/receive data.
+//! and the connection may be used immediately to send/receive data. Ip v4 number
+//! `127.0.0.1` is hardwired to the loopback device (i.e. localhost), which doesn't
+//! access any physical device and in which data can only be read by the same
+//! connection that has written it.
 //! * If `host` is omitted, this connectino will wait for a distant peer to
 //! connect.
 //! * The `protocol` is the hex-based number of the ip protocol
@@ -47,3 +50,43 @@ fn main() {
         }
     });
 }
+
+#[cfg(test)]
+fn test() {
+    use scheme::IpScheme;
+
+    println!("* Test that we can read a simple packet from the same connection.");
+    let bytes = "TEST".as_bytes();
+    let mut scheme = IpScheme::new();
+    let a = scheme.open("ip:127.0.0.1/11").unwrap();
+    let b = scheme.open("ip:127.0.0.1/11").unwrap();
+    let num_bytes_written = scheme.write(a, bytes).unwrap();
+    assert_eq!(num_bytes_written, bytes.len());
+
+    let mut buf = [0;65536];
+    let num_bytes_read = scheme.read(a, &mut buf).unwrap();
+    assert_eq!(num_bytes_read, num_bytes_written);
+
+    let bytes_read = &buf[0..num_bytes_read];
+    assert_eq!(bytes, bytes_read);
+
+    println!("* Test that we cannot read the same packet from a different connection.");
+    let num_bytes_read = scheme.read(b, &mut buf).unwrap();
+    assert_eq!(num_bytes_read, 0);
+
+    println!("* Push a number of packets, check that we get them in the right order.");
+    let mut payloads : Vec<String> = 0..100.map(|i| format!("TEST {}", i)).collect();
+    for payload in &payloads {
+        let bytes = payload.into_bytes();
+        let num_bytes_written = scheme.write(a, &bytes).unwrap();
+        assert_eq!(bytes.len(), num_bytes_written);
+    }
+    for payload in &payloads {
+        let bytes = payload.into_bytes();
+        let mut buf = [0;65536];
+        let num_bytes_read = scheme.read(a, &mut buf).unwrap();
+        assert_eq!(bytes.len(), num_bytes_read);
+        let bytes_read = &buf[0..num_bytes_read];
+        assert_eq!(bytes, bytes_read);
+    }
+}

schemes/ipd/src/resource.rs

@@ -1,21 +1,20 @@
 use std::{cmp, mem};
+use std::collections::VecDeque;
 
 use netutils::{n16, Ipv4Addr, Checksum, Ipv4Header, Ipv4};
 use resource_scheme::Resource;
 use syscall;
 use syscall::error::*;
 
+/// Max number of bytes in a packet.
+const MAX_PACKET_LENGTH : usize = 65536;
+
 /// A IP (internet protocol) resource.
 ///
 /// Each instance represents a connection (~ a IP socket).
 pub struct IpResource {
-    /// Link to the underlying device (typically, an Ethernet card).
+    /// The underlying mechanism ensured to connect to the peer.
-    pub link: usize,
+    pub connection: Connection,
-
-    /// If this connection was opened waiting for a peer (i.e. `ip:/protocol`),
-    /// the data received when the peer actually connected. Otherwise, empty.
-    /// Emptied during the first call to `read()`.
-    pub init_data: Vec<u8>,
 
     /// The IP address of the host (i.e. this machine).
     pub host_addr: Ipv4Addr,
@@ -33,6 +32,25 @@ pub struct IpResource {
     pub id: u16,
 }
 
+pub enum Connection {
+    Device {
+        /// Link to the underlying device (typically, an Ethernet card).
+        link: usize,
+
+        /// If this connection was opened waiting for a peer (i.e. `ip:/protocol`),
+        /// the data received when the peer actually connected. Otherwise, empty.
+        /// Emptied during the first call to `read()`.
+        init_data: Vec<u8>,
+    },
+    Loopback {
+        /// FIFO queue of packets written to the connection and waiting to be read.
+        ///
+        /// The data stored contains the exact data that has been added by the client
+        /// calling `write()`, without adding any headers.
+        packets: VecDeque<Vec<u8>>
+    }
+}
+
 impl Resource for IpResource {
     /// Duplicate the connection.
     ///
@@ -43,14 +61,26 @@ impl Resource for IpResource {
     /// Fails if the `link` to the underlying device cannot be
     /// duplicated.
     fn dup(&self) -> Result<Box<Self>> {
-        let link = try!(syscall::dup(self.link));
+        use self::Connection::*;
-        Ok(Box::new(IpResource {
+        let connection = match self.connection {
+            Loopback { ref packets } => Loopback {
+                packets: packets.clone()
+            },
+            Device { link, ref init_data } => {
+                let link = try!(syscall::dup(link));
+                let init_data = init_data.clone();
+                Device {
                     link: link,
-            init_data: self.init_data.clone(),
+                    init_data: init_data
+                }
+            }
+        };
+        Ok(Box::new(IpResource {
             host_addr: self.host_addr,
             peer_addr: self.peer_addr,
             proto: self.proto,
             id: self.id,
+            connection: connection,
         }))
     }
 
@@ -79,13 +109,35 @@ impl Resource for IpResource {
     /// can happen only if the connection was waiting for a remote peer to connect, i.e.
     /// with a url `ip:/protocol`, without host.
     ///
+    /// If this connection is a loopback, oldest unread packet written is read.
+    ///
     /// # Errors
     ///
     /// Fails if the call to `syscall::read()` fails for this device.
+    ///
+    /// # Data loss
+    ///
+    /// If `buf` is too small, *exceeding data is discarded*. To be sure that you read
+    /// all data, you should provide a 64kb `buf`.
     fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
-        if !self.init_data.is_empty() {
+        use self::Connection::*;
+        match self.connection {
+            Loopback { ref mut packets } => {
+                match packets.pop_front() {
+                    None => Ok(0),
+                    Some(data) => {
+                        for (b, d) in buf.iter_mut().zip(data.iter()) {
+                            *b = *d;
+                        }
+                        // Note: We're discarding excess `data`.
+                        Ok(cmp::min(buf.len(), data.len()))
+                    }
+                }
+            }
+            Device { ref mut init_data, link } => {
+                if !init_data.is_empty() {
                     let mut data: Vec<u8> = Vec::new();
-            mem::swap(&mut self.init_data, &mut data);
+                    mem::swap(init_data, &mut data);
 
                     for (b, d) in buf.iter_mut().zip(data.iter()) {
                         *b = *d;
@@ -94,8 +146,8 @@ impl Resource for IpResource {
                     return Ok(cmp::min(buf.len(), data.len()));
                 }
 
-        let mut bytes = [0; 65536];
+                let mut bytes = [0; MAX_PACKET_LENGTH];
-        let count = try!(syscall::read(self.link, &mut bytes));
+                let count = try!(syscall::read(link, &mut bytes));
 
                 if let Some(packet) = Ipv4::from_bytes(&bytes[..count]) {
                     if packet.header.proto == self.proto &&
@@ -111,6 +163,8 @@ impl Resource for IpResource {
 
                 Ok(0)
             }
+        }
+    }
 
     /// Send data to the peer.
     ///
@@ -118,8 +172,22 @@ impl Resource for IpResource {
     ///
     /// Fails if the call to `syscall::write()` fails for this device.
     fn write(&mut self, buf: &[u8]) -> Result<usize> {
-        let ip_data = Vec::from(buf);
+        use self::Connection::*;
 
+        let ip_data = Vec::from(buf);
+        match self.connection {
+            Loopback { ref mut packets } => {
+                // Make sure that we're not going to store data that can't be read.
+                let buf =
+                    if buf.len() > MAX_PACKET_LENGTH {
+                        &buf[0..MAX_PACKET_LENGTH]
+                    } else {
+                        buf
+                    };
+                packets.push_back(buf.to_vec());
+                return Ok(buf.len())
+            }
+            Device { link, .. } => {
                 self.id += 1;
                 let mut ip = Ipv4 {
                     header: Ipv4Header {
@@ -145,19 +213,27 @@ impl Resource for IpResource {
                                           Checksum::sum(ip.options.as_ptr() as usize, ip.options.len()));
                 }
 
-        match syscall::write(self.link, &ip.to_bytes()) {
+                match syscall::write(link, &ip.to_bytes()) {
                     Ok(_) => Ok(buf.len()),
                     Err(err) => Err(err),
                 }
             }
+        }
+    }
 
     fn sync(&mut self) -> Result<usize> {
-        syscall::fsync(self.link)
+        if let Connection::Device { link, .. } = self.connection {
+            syscall::fsync(link)
+        } else {
+            Ok(0)
+        }
     }
 }
 
 impl Drop for IpResource {
     fn drop(&mut self) {
-        let _ = syscall::close(self.link);
+        if let Connection::Device { link, .. } = self.connection {
+            let _ = syscall::close(link);
+        }
     }
 }

schemes/ipd/src/scheme.rs

@@ -1,4 +1,5 @@
 use std::cell::RefCell;
+use std::collections::VecDeque;
 use std::rand;
 use std::{str, u16};
 
@@ -8,7 +9,10 @@ use syscall;
 use syscall::error::{Error, Result, EACCES, ENOENT, EINVAL};
 use syscall::flag::O_RDWR;
 
-use resource::IpResource;
+use resource::*;
+
+/// The IP address of the localhost.
+const LOCALHOST: Ipv4Addr = Ipv4Addr { bytes: [127, 0, 0, 1] };
 
 /// A ARP entry (MAC + IP)
 pub struct ArpEntry {
@@ -48,6 +52,18 @@ impl ResourceScheme<IpResource> for IpScheme {
                         let mut route_mac = MacAddr::BROADCAST;
 
                         if ! peer_addr.equals(Ipv4Addr::BROADCAST) {
+                            if peer_addr.equals(LOCALHOST) {
+                                return Ok(Box::new(IpResource {
+                                    connection: Connection::Loopback {
+                                        packets: VecDeque::new()
+                                    },
+                                    host_addr: ip_addr,
+                                    peer_addr: peer_addr,
+                                    proto: proto,
+                                    id: (rand() % 65536) as u16,
+                                }));
+                            }
+
                             let mut needs_routing = false;
 
                             for octet in 0..4 {
@@ -116,8 +132,10 @@ impl ResourceScheme<IpResource> for IpScheme {
 
                         if let Ok(link) = syscall::open(&format!("ethernet:{}/800", &route_mac.to_string()), O_RDWR) {
                             return Ok(Box::new(IpResource {
+                                connection: Connection::Device {
                                     link: link,
                                     init_data: Vec::new(),
+                                },
                                 host_addr: ip_addr,
                                 peer_addr: peer_addr,
                                 proto: proto,
@@ -134,8 +152,10 @@ impl ResourceScheme<IpResource> for IpScheme {
                                         if packet.header.proto == proto &&
                                            (packet.header.dst.equals(ip_addr) || packet.header.dst.equals(Ipv4Addr::BROADCAST)) {
                                             return Ok(Box::new(IpResource {
+                                                connection: Connection::Device {
                                                     link: link,
                                                     init_data: packet.data,
+                                                },
                                                 host_addr: ip_addr,
                                                 peer_addr: packet.header.src,
                                                 proto: proto,