core::rt: Make TCP servers work

Author: brson

src/libcore/rt/io/net/tcp.rs

@@ -79,7 +79,7 @@ impl Writer for TcpStream {
 }
 
 pub struct TcpListener {
-    rtlistener: ~RtioTcpListenerObject
+    rtlistener: ~RtioTcpListenerObject,
 }
 
 impl TcpListener {
@@ -116,6 +116,8 @@ impl Listener<TcpStream> for TcpListener {
 #[cfg(test)]
 mod test {
     use super::*;
+    use int;
+    use cell::Cell;
     use rt::test::*;
     use rt::io::net::ip::Ipv4;
     use rt::io::*;
@@ -172,11 +174,11 @@ mod test {
         }
     }
 
-    #[test] #[ignore]
+    #[test]
     fn multiple_connect_serial() {
         do run_in_newsched_task {
             let addr = next_test_ip4();
-            let max = 100;
+            let max = 10;
 
             do spawntask_immediately {
                 let mut listener = TcpListener::bind(addr);
@@ -197,4 +199,82 @@ mod test {
         }
     }
 
+    #[test]
+    fn multiple_connect_interleaved_greedy_schedule() {
+        do run_in_newsched_task {
+            let addr = next_test_ip4();
+            static MAX: int = 10;
+
+            do spawntask_immediately {
+                let mut listener = TcpListener::bind(addr);
+                for int::range(0, MAX) |i| {
+                    let stream = Cell(listener.accept());
+                    rtdebug!("accepted");
+                    // Start another task to handle the connection
+                    do spawntask_immediately {
+                        let mut stream = stream.take();
+                        let mut buf = [0];
+                        stream.read(buf);
+                        assert!(buf[0] == i as u8);
+                        rtdebug!("read");
+                    }
+                }
+            }
+
+            connect(0, addr);
+
+            fn connect(i: int, addr: IpAddr) {
+                if i == MAX { return }
+
+                do spawntask_immediately {
+                    rtdebug!("connecting");
+                    let mut stream = TcpStream::connect(addr);
+                    // Connect again before writing
+                    connect(i + 1, addr);
+                    rtdebug!("writing");
+                    stream.write([i as u8]);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn multiple_connect_interleaved_lazy_schedule() {
+        do run_in_newsched_task {
+            let addr = next_test_ip4();
+            static MAX: int = 10;
+
+            do spawntask_immediately {
+                let mut listener = TcpListener::bind(addr);
+                for int::range(0, MAX) |_| {
+                    let stream = Cell(listener.accept());
+                    rtdebug!("accepted");
+                    // Start another task to handle the connection
+                    do spawntask_later {
+                        let mut stream = stream.take();
+                        let mut buf = [0];
+                        stream.read(buf);
+                        assert!(buf[0] == 99);
+                        rtdebug!("read");
+                    }
+                }
+            }
+
+            connect(0, addr);
+
+            fn connect(i: int, addr: IpAddr) {
+                if i == MAX { return }
+
+                do spawntask_later {
+                    rtdebug!("connecting");
+                    let mut stream = TcpStream::connect(addr);
+                    // Connect again before writing
+                    connect(i + 1, addr);
+                    rtdebug!("writing");
+                    stream.write([99]);
+                }
+            }
+        }
+    }
+
 }

src/libcore/rt/test.rs

@@ -64,6 +64,46 @@ pub fn spawntask_immediately(f: ~fn()) {
     }
 }
 
+/// Create a new task and run it right now. Aborts on failure
+pub fn spawntask_later(f: ~fn()) {
+    use super::sched::*;
+
+    let mut sched = local_sched::take();
+    let task = ~Task::with_local(&mut sched.stack_pool,
+                                 LocalServices::without_unwinding(),
+                                 f);
+
+    sched.task_queue.push_front(task);
+    local_sched::put(sched);
+}
+
+/// Spawn a task and either run it immediately or run it later
+pub fn spawntask_random(f: ~fn()) {
+    use super::sched::*;
+    use rand::{Rand, rng};
+
+    let mut rng = rng();
+    let run_now: bool = Rand::rand(&mut rng);
+
+    let mut sched = local_sched::take();
+    let task = ~Task::with_local(&mut sched.stack_pool,
+                                 LocalServices::without_unwinding(),
+                                 f);
+
+    if run_now {
+        do sched.switch_running_tasks_and_then(task) |task| {
+            let task = Cell(task);
+            do local_sched::borrow |sched| {
+                sched.task_queue.push_front(task.take());
+            }
+        }
+    } else {
+        sched.task_queue.push_front(task);
+        local_sched::put(sched);
+    }
+}
+
+
 /// Spawn a task and wait for it to finish, returning whether it completed successfully or failed
 pub fn spawntask_try(f: ~fn()) -> Result<(), ()> {
     use cell::Cell;

src/libcore/rt/uv/uvio.rs

@@ -21,6 +21,7 @@ use rt::uv::idle::IdleWatcher;
 use rt::rtio::*;
 use rt::sched::{Scheduler, local_sched};
 use rt::io::{standard_error, OtherIoError};
+use rt::tube::Tube;
 
 #[cfg(test)] use uint;
 #[cfg(test)] use unstable::run_in_bare_thread;
@@ -149,7 +150,7 @@ impl IoFactory for UvIoFactory {
     fn tcp_bind(&mut self, addr: IpAddr) -> Result<~RtioTcpListenerObject, IoError> {
         let mut watcher = TcpWatcher::new(self.uv_loop());
         match watcher.bind(addr) {
-            Ok(_) => Ok(~UvTcpListener { watcher: watcher }),
+            Ok(_) => Ok(~UvTcpListener::new(watcher)),
             Err(uverr) => {
                 // XXX: Should we wait until close completes?
                 watcher.as_stream().close(||());
@@ -161,10 +162,20 @@ impl IoFactory for UvIoFactory {
 
 // FIXME #6090: Prefer newtype structs but Drop doesn't work
 pub struct UvTcpListener {
-    watcher: TcpWatcher
+    watcher: TcpWatcher,
+    listening: bool,
+    incoming_streams: Tube<Result<~RtioTcpStreamObject, IoError>>
 }
 
 impl UvTcpListener {
+    fn new(watcher: TcpWatcher) -> UvTcpListener {
+        UvTcpListener {
+            watcher: watcher,
+            listening: false,
+            incoming_streams: Tube::new()
+        }
+    }
+
     fn watcher(&self) -> TcpWatcher { self.watcher }
 }
 
@@ -179,41 +190,37 @@ impl RtioTcpListener for UvTcpListener {
 
     fn accept(&mut self) -> Result<~RtioTcpStreamObject, IoError> {
         rtdebug!("entering listen");
-        let result_cell = empty_cell();
-        let result_cell_ptr: *Cell<Result<~RtioTcpStreamObject, IoError>> = &result_cell;
 
-        let server_tcp_watcher = self.watcher();
+        if self.listening {
+            return self.incoming_streams.recv();
+        }
 
-        let scheduler = local_sched::take();
-        assert!(scheduler.in_task_context());
+        self.listening = true;
 
-        do scheduler.deschedule_running_task_and_then |task| {
-            let task_cell = Cell(task);
-            let mut server_tcp_watcher = server_tcp_watcher;
-            do server_tcp_watcher.listen |server_stream_watcher, status| {
-                let maybe_stream = if status.is_none() {
-                    let mut server_stream_watcher = server_stream_watcher;
-                    let mut loop_ = server_stream_watcher.event_loop();
-                    let mut client_tcp_watcher = TcpWatcher::new(&mut loop_);
-                    let client_tcp_watcher = client_tcp_watcher.as_stream();
-                    // XXX: Need's to be surfaced in interface
-                    server_stream_watcher.accept(client_tcp_watcher);
-                    Ok(~UvTcpStream { watcher: client_tcp_watcher })
-                } else {
-                    Err(standard_error(OtherIoError))
-                };
-
-                unsafe { (*result_cell_ptr).put_back(maybe_stream); }
+        let server_tcp_watcher = self.watcher();
+        let incoming_streams_cell = Cell(self.incoming_streams.clone());
+
+        let incoming_streams_cell = Cell(incoming_streams_cell.take());
+        let mut server_tcp_watcher = server_tcp_watcher;
+        do server_tcp_watcher.listen |server_stream_watcher, status| {
+            let maybe_stream = if status.is_none() {
+                let mut server_stream_watcher = server_stream_watcher;
+                let mut loop_ = server_stream_watcher.event_loop();
+                let mut client_tcp_watcher = TcpWatcher::new(&mut loop_);
+                let client_tcp_watcher = client_tcp_watcher.as_stream();
+                // XXX: Need's to be surfaced in interface
+                server_stream_watcher.accept(client_tcp_watcher);
+                Ok(~UvTcpStream { watcher: client_tcp_watcher })
+            } else {
+                Err(standard_error(OtherIoError))
+            };
 
-                rtdebug!("resuming task from listen");
-                // Context switch
-                let scheduler = local_sched::take();
-                scheduler.resume_task_immediately(task_cell.take());
-            }
+            let mut incoming_streams = incoming_streams_cell.take();
+            incoming_streams.send(maybe_stream);
+            incoming_streams_cell.put_back(incoming_streams);
         }
 
-        assert!(!result_cell.is_empty());
-        return result_cell.take();
+        return self.incoming_streams.recv();
     }
 }