fixed read-parse-print slowdowns in clients

Author: BartMassey

Cargo.lock

@@ -26,3 +26,4 @@ features = ["net", "rt", "io-std", "io-util", "rt-multi-thread"]
 
 [dependencies.async-std]
 version = "1.10.0"
+features = ["unstable"]

Cargo.toml

@@ -52,6 +52,8 @@ Threaded server will use an amortized counter by default. To
 use an atomic counter, use the `--alt` argument.  They are
 roughly the same speed.
 
+## Notes
+
 On my modern Linux box I need to run these things as root to
 get decent performance. I also occasionally need to
 
@@ -60,8 +62,6 @@ get decent performance. I also occasionally need to
 to get SYN cookies turned off on localhost. (Sigh. Working
 on reporting this.) Don't forget to undo this when you're done!
 
-## Notes
-
 Various optimizations that have been tried:
 
 * Make the seq server use an amortized counter construction
@@ -73,6 +73,14 @@ Various optimizations that have been tried:
 * Make the async server use amortized counter: roughly 3×
   slowdown, ditched.
 
+* Replace the read-parse-print stupidity in the clients with
+  simply writing the received bytes. Some speedup, kept.
+
+* Use `tokio::io::stdout` in the async client. Sadly, since
+  this is not locked it's not guaranteed to work reliably
+  100% of the time. Replaced with `spawn_blocking` writes to
+  `std::io::stdout` for about a 20% slowdown, but correctness.
+
 ## Acknowledgements
 
 Thanks to Josh Triplett for the scoped-threads server and

README.md

@@ -1,30 +1,38 @@
+use std::io::Write;
+
 pub mod rt_async_std {
-    pub use async_std::io::{self, ReadExt, WriteExt};
-    pub use async_std::net::TcpStream;
-    pub use async_std::task;
+
+    use super::*;
+
+    use async_std::io::ReadExt;
+    use async_std::net::TcpStream;
+    use async_std::task;
 
     async fn drop_handle(h: task::JoinHandle<()>) {
         h.await;
     }
 
-    super::ccasync!();
+    ccasync!();
 
     pub fn start(n: usize, m: usize) {
         task::block_on(send(n, m));
     }
 }
 
 pub mod rt_tokio {
-    pub use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
-    pub use tokio::net::TcpStream;
-    pub use tokio::runtime::Runtime;
-    pub use tokio::task;
+
+    use super::*;
+
+    use tokio::io::AsyncReadExt;
+    use tokio::net::TcpStream;
+    use tokio::runtime::Runtime;
+    use tokio::task;
 
     async fn drop_handle(h: task::JoinHandle<()>) {
         h.await.unwrap();
     }
 
-    super::ccasync!();
+    ccasync!();
 
     pub fn start(n: usize, m: usize) {
         let rt = Runtime::new().unwrap();
@@ -34,23 +42,24 @@ pub mod rt_tokio {
 
 macro_rules! ccasync {
     () => {
-        async fn get_count() -> u64 {
+        async fn get_count() -> Vec<u8> {
             let mut stream = TcpStream::connect("127.0.0.1:10123").await.unwrap();
-            let mut buf = String::with_capacity(22);
-            stream.read_to_string(&mut buf).await.unwrap();
+            let mut buf = Vec::with_capacity(22);
+            stream.read_to_end(&mut buf).await.unwrap();
             drop(stream);
-            buf.trim_end().parse().unwrap()
+            buf
         }
 
         async fn send(n: usize, m: usize) {
             let mut handles = Vec::with_capacity(m);
             for _ in 0..n {
                 let h = task::spawn(async {
-                    let c = get_count().await;
-                    io::stdout()
-                        .write_all(format!("{}\n", c).as_bytes())
-                        .await
-                        .unwrap();
+                    let mut buf = get_count().await;
+                    let nbuf = buf.len();
+                    buf[nbuf - 2] = b'\n';
+                    task::spawn_blocking(move || {
+                        std::io::stdout().write_all(&buf[..nbuf-1]).unwrap();
+                    });
                 });
                 handles.push(h);
                 if handles.len() >= m {

src/ccasync.rs

@@ -1,13 +1,16 @@
-use std::io::Read;
+use std::io::{Read, Write, stdout};
 use std::net;
 
 pub fn send(n: usize) {
-    let mut buf = String::with_capacity(22);
+    let stdout = stdout();
+    let mut stdout = stdout.lock();
+    let mut buf = vec![0u8; 22];
     for _ in 0..n {
         let mut stream = net::TcpStream::connect("127.0.0.1:10123").unwrap();
         buf.clear();
-        stream.read_to_string(&mut buf).unwrap();
-        let count: u64 = buf.trim_end().parse().unwrap();
-        println!("{}", count);
+        stream.read_to_end(&mut buf).unwrap();
+        let nbuf = buf.len();
+        buf[nbuf - 2] = b'\n';
+        stdout.write_all(&buf[..nbuf-1]).unwrap();
     }
 }

src/ccseq.rs

@@ -1,13 +1,20 @@
 use std::collections::VecDeque;
-use std::io::Read;
+use std::io::{Read, Write, stdout};
 use std::net;
 use std::thread;
 
-fn get_count() -> u64 {
+fn get_count() -> Vec<u8> {
     let mut stream = net::TcpStream::connect("127.0.0.1:10123").unwrap();
-    let mut buf = String::with_capacity(22);
-    stream.read_to_string(&mut buf).unwrap();
-    buf.trim_end().parse().unwrap()
+    let mut buf = vec![0u8; 22];
+    stream.read_to_end(&mut buf).unwrap();
+    buf
+}
+
+fn print_count(h: thread::JoinHandle<Vec<u8>>) {
+    let mut buf = h.join().unwrap();
+    let nbuf = buf.len();
+    buf[nbuf - 2] = b'\n';
+    stdout().write_all(&buf[..nbuf-1]).unwrap();
 }
 
 pub fn send(n: usize, m: usize) {
@@ -16,11 +23,11 @@ pub fn send(n: usize, m: usize) {
         handles.push_front(thread::spawn(get_count));
         if handles.len() > m {
             for h in handles.drain(m..) {
-                println!("{:?}", h.join().unwrap());
+                print_count(h);
             }
         }
     }
     for h in handles.into_iter() {
-        println!("{:?}", h.join().unwrap());
+        print_count(h);
     }
 }