Merge pull request #11 from Covertness/windows

fix the problem on the Windows

Cargo.toml

@@ -13,7 +13,6 @@ keywords = ["CoAP"]
 bincode = "0.3.0"
 rustc-serialize = "0.3"
 mio = "0.5"
-threadpool = "0.1"
 url = "0.2.36"
 num = "0.1"
 rand = "0.3"

README.md

@@ -72,10 +72,7 @@ fn main() {
 ```
 
 ## Benchmark
-### Using one thread
-![image](benches/one_thread_summary.png)
-
-### Using eight thread
-![image](benches/eight_thread_summary.png)
-
-Tests were performed using [basho_bench](https://github.com/basho/basho_bench).
+```bash
+$ cargo run --example server
+$ cargo bench
+```

benches/client.rs

@@ -4,23 +4,24 @@ extern crate test;
 extern crate coap;
 
 use test::Bencher;
-use coap::packet::*;
-use coap::CoAPClient;
+use coap::{CoAPClient, CoAPRequest, IsMessage, MessageType, CoAPOption};
 
 #[bench]
 fn bench_client_request(b: &mut Bencher) {
 	let addr = "127.0.0.1:5683";
-	let request = "test";
-	let mut packet = Packet::new();
-	packet.header.set_version(1);
-	packet.header.set_type(MessageType::Confirmable);
-	packet.header.set_code("0.01");
-	packet.header.set_message_id(1);
-	packet.set_token(vec!(0x51, 0x55, 0x77, 0xE8));
-	packet.add_option(CoAPOption::UriPath, request.to_string().into_bytes());
+	let endpoint = "test";
+
+	let mut request = CoAPRequest::new();
+	request.set_version(1);
+	request.set_type(MessageType::Confirmable);
+	request.set_code("0.01");
+	request.set_message_id(1);
+	request.set_token(vec!(0x51, 0x55, 0x77, 0xE8));
+	request.add_option(CoAPOption::UriPath, endpoint.to_string().into_bytes());
 
 	b.iter(|| {
 		let client = CoAPClient::new(addr).unwrap();
-		client.send(&packet).unwrap();
+		client.send(&request).unwrap();
+		client.receive().unwrap();
 	});
 }
\ No newline at end of file

src/lib.rs

@@ -70,7 +70,6 @@
 extern crate bincode;
 extern crate rustc_serialize;
 extern crate mio;
-extern crate threadpool;
 extern crate url;
 extern crate num;
 extern crate rand;

src/server.rs

@@ -8,9 +8,7 @@ use mio::udp::UdpSocket;
 use message::packet::Packet;
 use message::request::CoAPRequest;
 use message::response::CoAPResponse;
-use threadpool::ThreadPool;
 
-const DEFAULT_WORKER_NUM: usize = 4;
 type TxQueue = mpsc::Sender<QueuedResponse>;
 type RxQueue = mpsc::Receiver<QueuedResponse>;
 
@@ -42,77 +40,133 @@ impl<F> CoAPHandler for F
 
 struct UdpHandler<H: CoAPHandler + 'static> {
     socket: UdpSocket,
-    thread_pool: ThreadPool,
     tx_sender: TxQueue,
+    rx_recv: RxQueue,
     coap_handler: H,
 }
 
 impl<H: CoAPHandler + 'static> UdpHandler<H> {
     fn new(socket: UdpSocket,
-           thread_pool: ThreadPool,
            tx_sender: TxQueue,
+           rx_recv: RxQueue,
            coap_handler: H)
            -> UdpHandler<H> {
         UdpHandler {
             socket: socket,
-            thread_pool: thread_pool,
             tx_sender: tx_sender,
+            rx_recv: rx_recv,
             coap_handler: coap_handler,
         }
     }
-}
 
-impl<H: CoAPHandler + 'static> Handler for UdpHandler<H> {
-    type Timeout = usize;
-    type Message = ();
+    fn request_handler(&self) -> Option<QueuedResponse> {
+        match self.requset_recv() {
+            Some(rqst) => {
+                let src = rqst.source.unwrap();
+                match self.coap_handler.handle(rqst) {
+                    Some(response) => {
+                        debug!("Response: {:?}", response);
+                        return Some(QueuedResponse {
+                            address: src,
+                            response: response,
+                        });
+                    }
+                    None => {
+                        debug!("No response");
+                        return None;
+                    }
+                }
+            }
+            None => {
+                return None;
+            }
+        }
+    }
 
-    fn ready(&mut self, _: &mut EventLoop<UdpHandler<H>>, _: Token, events: EventSet) {
-        if !events.is_readable() {
-            warn!("Unreadable Event");
-            return;
+    fn response_handler(&self) {
+        loop {
+            match self.rx_recv.try_recv() {
+                Ok(q_res) => {
+                    self.response_send(q_res);
+                }
+                Err(mpsc::TryRecvError::Empty) => {
+                    break;
+                }
+                Err(mpsc::TryRecvError::Disconnected) => {
+                    error!("The RxQueue become disconnected");
+                    break;
+                }
+            }
         }
+    }
 
-        let coap_handler = self.coap_handler;
+    fn requset_recv(&self) -> Option<CoAPRequest> {
         let mut buf = [0; 1500];
 
         match self.socket.recv_from(&mut buf) {
             Ok(Some((nread, src))) => {
                 debug!("Handling request from {}", src);
-                let response_q = self.tx_sender.clone();
-
-                self.thread_pool.execute(move || {
-                    match Packet::from_bytes(&buf[..nread]) {
-                        Ok(packet) => {
-                            // Dispatch user handler, if there is a response packet
-                            //   send the reply via the TX thread
-                            let rqst = CoAPRequest::from_packet(packet, &src);
-                            match coap_handler.handle(rqst) {
-                                Some(response) => {
-                                    debug!("Response: {:?}", response);
-                                    response_q.send(QueuedResponse {
-                                            address: src,
-                                            response: response,
-                                        })
-                                        .unwrap();
-                                }
-                                None => {
-                                    debug!("No response");
-                                }
-                            };
-                        }
-                        Err(_) => {
-                            error!("Failed to parse request");
-                            return;
-                        }
-                    };
-                });
+
+                match Packet::from_bytes(&buf[..nread]) {
+                    Ok(packet) => {
+                        return Some(CoAPRequest::from_packet(packet, &src));
+                    }
+                    Err(_) => {
+                        error!("Failed to parse request");
+                        return None;
+                    }
+                }
             }
             _ => {
                 error!("Failed to read from socket");
                 panic!("unexpected error");
             }
         }
+    }
+
+    fn response_send(&self, q_res: QueuedResponse) {
+        match q_res.response.message.to_bytes() {
+            Ok(bytes) => {
+                match self.socket.send_to(&bytes[..], &q_res.address) {
+                    Ok(None) => {
+                        // Look at https://github.com/carllerche/mio/issues/411 in detail
+                        error!("Failed to complete the response");
+                    }
+                    Ok(_) => {}
+                    Err(_) => {
+                        error!("Failed to send response");
+                    }
+                }
+            }
+            Err(_) => {
+                error!("Failed to decode response");
+            }
+        }
+    }
+}
+
+impl<H: CoAPHandler + 'static> Handler for UdpHandler<H> {
+    type Timeout = usize;
+    type Message = ();
 
+    fn ready(&mut self, event_loop: &mut EventLoop<UdpHandler<H>>, _: Token, events: EventSet) {
+        if events.is_writable() {
+            // handle the response
+            self.response_handler();
+            event_loop.reregister(&self.socket, Token(0), EventSet::readable(), PollOpt::edge()).unwrap();
+            return;
+        } else if events.is_readable() {
+            // handle the request
+            match self.request_handler() {
+                Some(response) => {
+                    self.tx_sender.send(response).unwrap();
+                    event_loop.reregister(&self.socket, Token(0), EventSet::writable(), PollOpt::edge()).unwrap();
+                }
+                None => {}
+            }
+        } else {
+            warn!("Unreadable Event, {:?}", events);
+        }
     }
 
     fn notify(&mut self, event_loop: &mut EventLoop<UdpHandler<H>>, _: ()) {
@@ -125,8 +179,6 @@ pub struct CoAPServer {
     socket: UdpSocket,
     event_sender: Option<Sender<()>>,
     event_thread: Option<thread::JoinHandle<()>>,
-    tx_thread: Option<thread::JoinHandle<()>>,
-    worker_num: usize,
 }
 
 impl CoAPServer {
@@ -140,8 +192,6 @@ impl CoAPServer {
                             socket: s,
                             event_sender: None,
                             event_thread: None,
-                            tx_thread: None,
-                            worker_num: DEFAULT_WORKER_NUM,
                         })
                     })
                 }
@@ -168,26 +218,18 @@ impl CoAPServer {
         }
 
         // Create resources
-        let worker_num = self.worker_num;
         let (tx, rx) = mpsc::channel();
         let (tx_send, tx_recv): (TxQueue, RxQueue) = mpsc::channel();
-        let tx_only = self.socket.try_clone().unwrap();
-
-        // Setup and spawn single TX thread
-        let tx_thread = thread::spawn(move || {
-            transmit_handler(tx_recv, tx_only);
-        });
 
         // Setup and spawn event loop thread, which will spawn
         //   children threads which handle incomining requests
         let thread = thread::spawn(move || {
-            let thread_pool = ThreadPool::new(worker_num);
             let mut event_loop = EventLoop::new().unwrap();
             event_loop.register(&socket, Token(0), EventSet::readable(), PollOpt::edge()).unwrap();
 
             tx.send(event_loop.channel()).unwrap();
 
-            event_loop.run(&mut UdpHandler::new(socket, thread_pool, tx_send, handler)).unwrap();
+            event_loop.run(&mut UdpHandler::new(socket, tx_send, tx_recv, handler)).unwrap();
         });
 
         // Ensure threads started successfully
@@ -195,7 +237,6 @@ impl CoAPServer {
             Ok(event_sender) => {
                 self.event_sender = Some(event_sender);
                 self.event_thread = Some(thread);
-                self.tx_thread = Some(tx_thread);
                 Ok(())
             }
             Err(_) => Err(CoAPServerError::EventLoopError),
@@ -213,37 +254,8 @@ impl CoAPServer {
             _ => {}
         }
     }
-
-    /// Set the number of threads for handling requests
-    pub fn set_worker_num(&mut self, worker_num: usize) {
-        self.worker_num = worker_num;
-    }
 }
 
-fn transmit_handler(tx_recv: RxQueue, tx_only: UdpSocket) {
-    // Note! We should only transmit with this UDP Socket
-    // TODO: Add better support for failure detection or logging
-    loop {
-        match tx_recv.recv() {
-            Ok(q_res) => {
-                match q_res.response.message.to_bytes() {
-                    Ok(bytes) => {
-                        let _ = tx_only.send_to(&bytes[..], &q_res.address);
-                    }
-                    Err(_) => {
-                        error!("Failed to decode response");
-                    }
-                }
-            }
-            // recv error occurs when all transmitters are terminited
-            //   (when all UDP Handlers are closed)
-            Err(_) => {
-                info!("Shutting down Transmit Handler");
-                break;
-            }
-        }
-    }
-}
 
 impl Drop for CoAPServer {
     fn drop(&mut self) {