Actix: update deps and plaintext/json impl (#4777)

* update deps and remove core plaintext and json benchs

* read randomNumber from row #4771

* add custom plaintext and json benchs

* update worlds random number

frameworks/Rust/actix/Cargo.toml

@@ -19,13 +19,18 @@ path = "src/main_diesel.rs"
 name = "actix-pg"
 path = "src/main_pg.rs"
 
+[[bin]]
+name = "actix-raw"
+path = "src/main_raw.rs"
+
 [dependencies]
-actix = { version="0.8.1", features=["http"] }
-actix-web = { version="1.0.0-beta.3", default-features = false }
-actix-http = { version="0.1.5", default-features = false }
+actix = { version="0.8.2", features=["http"] }
+actix-web = { version="1.0.0-beta.5", default-features = false }
+actix-http = { version="0.2.0", default-features = false }
 actix-rt = "0.2.2"
-actix-server = "0.4.2"
-actix-service = "0.3.6"
+actix-codec = "0.1.2"
+actix-server = "0.5.0"
+actix-service = "0.4.0"
 jemallocator = "0.3.0"
 askama = "0.8"
 markup = "0.3.1"
@@ -42,6 +47,7 @@ diesel = { version = "1.2", features = ["postgres"] }
 url = { version="1.7", features=["query_encoding"] }
 log = { version = "0.4", features = ["release_max_level_off"] }
 v_htmlescape = "0.4"
+tokio-tcp = "0.1"
 tokio-postgres = { git="https://github.com/fafhrd91/rust-postgres.git" }
 
 [build-dependencies]
@@ -54,3 +60,4 @@ codegen-units = 1
 
 [patch.crates-io]
 tokio-reactor = { git="https://github.com/fafhrd91/tokio.git" }
+actix-http = { git="https://github.com/actix/actix-web.git", default-features = false }

frameworks/Rust/actix/actix-core.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.34.0
+FROM rust:1.34.1
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-diesel.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.34.0
+FROM rust:1.34.1
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-pg.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.34.0
+FROM rust:1.34.1
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-raw.dockerfile

@@ -0,0 +1,9 @@
+FROM rust:1.34.1
+
+ADD ./ /actix
+WORKDIR /actix
+
+RUN cargo clean
+RUN RUSTFLAGS="-C target-cpu=native" cargo build --release
+
+CMD ./target/release/actix-raw

frameworks/Rust/actix/actix.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.34.0
+FROM rust:1.34.1
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/benchmark_config.json

@@ -21,8 +21,6 @@
     },
     "core": {
       "fortune_url": "/fortune",
-      "json_url": "/json",
-      "plaintext_url": "/plaintext",
       "db_url": "/db",
       "query_url": "/queries?q=",
       "update_url": "/updates?q=",
@@ -41,6 +39,24 @@
       "notes": "",
       "versus": ""
     },
+    "raw": {
+      "json_url": "/json",
+      "plaintext_url": "/plaintext",
+      "port": 8080,
+      "approach": "Realistic",
+      "classification": "Platform",
+      "database": "Postgres",
+      "framework": "actix",
+      "language": "Rust",
+      "orm": "Raw",
+      "platform": "None",
+      "webserver": "actix-web",
+      "os": "Linux",
+      "database_os": "Linux",
+      "display_name": "Actix [Raw]",
+      "notes": "",
+      "versus": ""
+    },
     "diesel": {
       "db_url": "/db",
       "fortune_url": "/fortune",

frameworks/Rust/actix/src/db_pg.rs

@@ -1,3 +1,4 @@
+use std::fmt::Write;
 use std::io;
 
 use actix::fut;
@@ -151,7 +152,7 @@ impl Handler<UpdateWorld> for PgConnection {
     fn handle(&mut self, msg: UpdateWorld, _: &mut Self::Context) -> Self::Result {
         let mut worlds = Vec::with_capacity(msg.0 as usize);
         for _ in 0..msg.0 {
-            let id: i32 = self.rng.gen_range(1, 10_001);
+            let id = self.rng.gen_range(1, 10_001);
             let w_id: i32 = self.rng.gen_range(1, 10_001);
             worlds.push(
                 self.cl
@@ -164,10 +165,12 @@ impl Handler<UpdateWorld> for PgConnection {
                     })
                     .map(move |(row, _)| {
                         let row = row.unwrap();
-                        World {
+                        let mut world = World {
                             id: row.get(0),
-                            randomnumber: id,
-                        }
+                            randomnumber: row.get(1),
+                        };
+                        world.randomnumber = id;
+                        world
                     }),
             );
         }
@@ -182,7 +185,7 @@ impl Handler<UpdateWorld> for PgConnection {
                         .push_str("UPDATE world SET randomnumber = temp.randomnumber FROM (VALUES ");
 
                     for w in &worlds {
-                        update.push_str(&format!("({}, {}),", w.id, w.randomnumber));
+                        let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
                     }
                     worlds.sort_by_key(|w| w.id);
 

frameworks/Rust/actix/src/db_pg_direct.rs

@@ -1,3 +1,4 @@
+use std::fmt::Write;
 use std::io;
 
 use actix_http::Error;
@@ -114,15 +115,18 @@ impl PgConnection {
                     })
                     .map(move |(row, _)| {
                         let row = row.unwrap();
-                        World {
+                        let mut world = World {
                             id: row.get(0),
-                            randomnumber: id,
-                        }
+                            randomnumber: row.get(1),
+                        };
+                        world.randomnumber = id;
+                        world
                     }),
             );
         }
 
         let mut cl = self.cl.clone();
+        let mut rng = self.rng.clone();
         stream::futures_unordered(worlds)
             .collect()
             .and_then(move |worlds| {
@@ -132,7 +136,7 @@ impl PgConnection {
                 );
 
                 for w in &worlds {
-                    update.push_str(&format!("({}, {}),", w.id, w.randomnumber));
+                    let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
                 }
                 update.pop();
                 update.push_str(

frameworks/Rust/actix/src/main_platform.rs

@@ -14,8 +14,8 @@ use actix_http::http::{HeaderValue, StatusCode};
 use actix_http::{Error, HttpService, KeepAlive, Request, Response};
 use actix_server::{Server, ServerConfig};
 use actix_service::{NewService, Service};
-use bytes::{Bytes, BytesMut};
-use futures::future::{join_all, ok, Either, FutureResult};
+use bytes::BytesMut;
+use futures::future::ok;
 use futures::{Async, Future, Poll};
 use serde_json::to_writer;
 
@@ -24,33 +24,20 @@ mod models;
 mod utils;
 
 use crate::db_pg_direct::PgConnection;
-use crate::utils::{FortunesTemplate, Message, Writer, SIZE};
+use crate::utils::{FortunesTemplate, Writer};
 
 struct App {
-    dbs: Vec<PgConnection>,
-    useall: bool,
-    next: usize,
-}
-
-impl App {
-    fn get_db(&mut self) -> &mut PgConnection {
-        if self.useall {
-            self.next = (self.next + 1) % 4;
-            &mut self.dbs[self.next]
-        } else {
-            &mut self.dbs[0]
-        }
-    }
+    hdr_srv: HeaderValue,
+    hdr_ctjson: HeaderValue,
+    hdr_cthtml: HeaderValue,
+    db: PgConnection,
 }
 
 impl Service for App {
     type Request = Request;
     type Response = Response;
     type Error = Error;
-    type Future = Either<
-        FutureResult<Self::Response, Self::Error>,
-        Box<Future<Item = Response, Error = Error>>,
-    >;
+    type Future = Box<Future<Item = Response, Error = Error>>;
 
     #[inline]
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@@ -60,100 +47,71 @@ impl Service for App {
     fn call(&mut self, req: Request) -> Self::Future {
         let path = req.path();
         match path.len() {
-            10 if path == "/plaintext" => {
-                let mut res = Response::with_body(
-                    StatusCode::OK,
-                    Body::Bytes(Bytes::from_static(b"Hello, World!")),
-                );
-                res.headers_mut()
-                    .insert(SERVER, HeaderValue::from_static("Actix"));
-                res.headers_mut()
-                    .insert(CONTENT_TYPE, HeaderValue::from_static("text/plain"));
-                Either::A(ok(res))
-            }
-            5 if path == "/json" => {
-                let message = Message {
-                    message: "Hello, World!",
-                };
-                let mut body = BytesMut::with_capacity(SIZE);
-                to_writer(Writer(&mut body), &message).unwrap();
-                let mut res =
-                    Response::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
-                res.headers_mut()
-                    .insert(SERVER, HeaderValue::from_static("Actix"));
-                res.headers_mut()
-                    .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-                Either::A(ok(res))
-            }
             3 if path == "/db" => {
-                let fut = self.dbs[0].get_world();
+                let fut = self.db.get_world();
+                let h_srv = self.hdr_srv.clone();
+                let h_ct = self.hdr_ctjson.clone();
 
-                Either::B(Box::new(fut.map(move |body| {
+                Box::new(fut.map(move |body| {
                     let mut res = Response::with_body(StatusCode::OK, Body::Bytes(body));
-                    res.headers_mut()
-                        .insert(SERVER, HeaderValue::from_static("Actix"));
-                    res.headers_mut().insert(
-                        CONTENT_TYPE,
-                        HeaderValue::from_static("application/json"),
-                    );
+                    let hdrs = res.headers_mut();
+                    hdrs.insert(SERVER, h_srv);
+                    hdrs.insert(CONTENT_TYPE, h_ct);
                     res
-                })))
+                }))
             }
             8 if path == "/fortune" => {
-                let fut = self.dbs[0].tell_fortune();
+                let fut = self.db.tell_fortune();
+                let h_srv = self.hdr_srv.clone();
+                let h_ct = self.hdr_cthtml.clone();
 
-                Either::B(Box::new(fut.from_err().map(move |fortunes| {
+                Box::new(fut.from_err().map(move |fortunes| {
                     let mut body = BytesMut::with_capacity(2048);
                     let mut writer = Writer(&mut body);
                     let _ = write!(writer, "{}", FortunesTemplate { fortunes });
                     let mut res =
                         Response::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
-                    res.headers_mut()
-                        .insert(SERVER, HeaderValue::from_static("Actix"));
-                    res.headers_mut().insert(
-                        CONTENT_TYPE,
-                        HeaderValue::from_static("text/html; charset=utf-8"),
-                    );
+                    let hdrs = res.headers_mut();
+                    hdrs.insert(SERVER, h_srv);
+                    hdrs.insert(CONTENT_TYPE, h_ct);
                     res
-                })))
+                }))
             }
             8 if path == "/queries" => {
                 let q = utils::get_query_param(req.uri().query().unwrap_or("")) as usize;
-                let fut = self.dbs[0].get_worlds(q);
+                let fut = self.db.get_worlds(q);
+                let h_srv = self.hdr_srv.clone();
+                let h_ct = self.hdr_ctjson.clone();
 
-                Either::B(Box::new(fut.from_err().map(move |worlds| {
+                Box::new(fut.from_err().map(move |worlds| {
                     let mut body = BytesMut::with_capacity(35 * worlds.len());
                     to_writer(Writer(&mut body), &worlds).unwrap();
                     let mut res =
                         Response::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
-                    res.headers_mut()
-                        .insert(SERVER, HeaderValue::from_static("Actix"));
-                    res.headers_mut().insert(
-                        CONTENT_TYPE,
-                        HeaderValue::from_static("application/json"),
-                    );
+                    let hdrs = res.headers_mut();
+                    hdrs.insert(SERVER, h_srv);
+                    hdrs.insert(CONTENT_TYPE, h_ct);
                     res
-                })))
+                }))
             }
             8 if path == "/updates" => {
                 let q = utils::get_query_param(req.uri().query().unwrap_or("")) as usize;
-                let fut = self.dbs[0].update(q);
+                let fut = self.db.update(q);
+                let h_srv = self.hdr_srv.clone();
+                let h_ct = self.hdr_ctjson.clone();
 
-                Either::B(Box::new(fut.from_err().map(move |worlds| {
+                Box::new(fut.from_err().map(move |worlds| {
                     let mut body = BytesMut::with_capacity(35 * worlds.len());
                     to_writer(Writer(&mut body), &worlds).unwrap();
                     let mut res =
                         Response::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
-                    res.headers_mut()
-                        .insert(SERVER, HeaderValue::from_static("Actix"));
-                    res.headers_mut().insert(
-                        CONTENT_TYPE,
-                        HeaderValue::from_static("application/json"),
-                    );
+                    let hdrs = res.headers_mut();
+                    hdrs.insert(SERVER, h_srv);
+                    hdrs.insert(CONTENT_TYPE, h_ct);
                     res
-                })))
+                }))
             }
-            _ => Either::A(ok(Response::new(http::StatusCode::NOT_FOUND))),
+            _ => Box::new(ok(Response::new(http::StatusCode::NOT_FOUND))),
         }
     }
 }
@@ -161,7 +119,8 @@ impl Service for App {
 #[derive(Clone)]
 struct AppFactory;
 
-impl NewService<ServerConfig> for AppFactory {
+impl NewService for AppFactory {
+    type Config = ServerConfig;
     type Request = Request;
     type Response = Response;
     type Error = Error;
@@ -173,14 +132,11 @@ impl NewService<ServerConfig> for AppFactory {
         const DB_URL: &str =
             "postgres://benchmarkdbuser:benchmarkdbpass@tfb-database/hello_world";
 
-        let mut conns = Vec::new();
-        for _ in 0..4 {
-            conns.push(PgConnection::connect(DB_URL));
-        }
-        Box::new(join_all(conns).map(|dbs| App {
-            dbs,
-            next: 0,
-            useall: num_cpus::get() > 4,
+        Box::new(PgConnection::connect(DB_URL).map(|db| App {
+            db,
+            hdr_srv: HeaderValue::from_static("Actix"),
+            hdr_ctjson: HeaderValue::from_static("application/json"),
+            hdr_cthtml: HeaderValue::from_static("text/html; charset=utf-8"),
         }))
     }
 }

frameworks/Rust/actix/src/main_raw.rs

@@ -0,0 +1,160 @@
+#[global_allocator]
+static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
+
+#[macro_use]
+extern crate serde_derive;
+#[macro_use]
+extern crate diesel;
+
+use std::{io, io::Write};
+
+use actix_codec::{AsyncRead, AsyncWrite, Decoder};
+use actix_http::h1;
+use actix_http::Request;
+use actix_server::{Io, Server};
+use actix_service::service_fn;
+use bytes::{BufMut, BytesMut};
+use futures::{Async, Future, Poll};
+use serde_json::to_writer;
+use tokio_tcp::TcpStream;
+
+mod models;
+mod utils;
+
+use crate::utils::{Message, Writer};
+
+const HTTPOK: &[u8] = b"HTTP/1.1 200 OK\r\n";
+const HTTPNFOUND: &[u8] = b"HTTP/1.1 400 OK\r\n";
+const HDR_SERVER: &[u8] = b"Server: Actix\r\n";
+const HDR_CTPLAIN: &[u8] = b"Content-Type: text/plain\r\n";
+const HDR_CTJSON: &[u8] = b"Content-Type: application/json\r\n";
+const HDR_CLEN: &[u8] = b"Content-Length: ";
+const BODY: &[u8] = b"Hello, World!";
+
+struct App<T> {
+    io: T,
+    read_buf: BytesMut,
+    write_buf: BytesMut,
+    codec: h1::Codec,
+}
+
+impl<T: AsyncRead + AsyncWrite> App<T> {
+    fn handle_request(&mut self, req: Request) {
+        let path = req.path();
+        match path.len() {
+            10 if path == "/plaintext" => {
+                self.write_buf.extend_from_slice(HTTPOK);
+                self.write_buf.extend_from_slice(HDR_SERVER);
+                self.write_buf.extend_from_slice(HDR_CTPLAIN);
+                self.write_buf.extend_from_slice(HDR_CLEN);
+                let _ = write!(Writer(&mut self.write_buf), "{}\r\n", 13);
+                self.codec.config().set_date(&mut self.write_buf);
+                self.write_buf.extend_from_slice(BODY);
+            }
+            5 if path == "/json" => {
+                let message = Message {
+                    message: "Hello, World!",
+                };
+
+                self.write_buf.extend_from_slice(HTTPOK);
+                self.write_buf.extend_from_slice(HDR_SERVER);
+                self.write_buf.extend_from_slice(HDR_CTJSON);
+                self.write_buf.extend_from_slice(HDR_CLEN);
+                let _ = write!(Writer(&mut self.write_buf), "{}\r\n", 27);
+                self.codec.config().set_date(&mut self.write_buf);
+                to_writer(Writer(&mut self.write_buf), &message).unwrap();
+            }
+            _ => {
+                self.write_buf.extend_from_slice(HTTPNFOUND);
+                self.write_buf.extend_from_slice(HDR_SERVER);
+                self.write_buf.extend_from_slice(HDR_CLEN);
+                let _ = write!(Writer(&mut self.write_buf), "0\r\n");
+            }
+        }
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite> Future for App<T> {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        if self.write_buf.remaining_mut() < 1024 {
+            self.write_buf.reserve(32_768);
+        }
+
+        loop {
+            if self.read_buf.remaining_mut() < 4096 {
+                self.read_buf.reserve(32_768);
+            }
+            let read = unsafe { self.io.read(self.read_buf.bytes_mut()) };
+            match read {
+                Ok(0) => return Ok(Async::Ready(())),
+                Ok(n) => unsafe { self.read_buf.advance_mut(n) },
+                Err(e) => {
+                    if e.kind() == io::ErrorKind::WouldBlock {
+                        break;
+                    } else {
+                        return Err(());
+                    }
+                }
+            }
+        }
+
+        loop {
+            match self.codec.decode(&mut self.read_buf) {
+                Ok(Some(h1::Message::Item(req))) => self.handle_request(req),
+                Ok(None) => break,
+                _ => return Err(()),
+            }
+        }
+
+        if !self.write_buf.is_empty() {
+            let len = self.write_buf.len();
+            let mut written = 0;
+            while written < len {
+                match self.io.write(&self.write_buf[written..]) {
+                    Ok(0) => return Ok(Async::Ready(())),
+                    Ok(n) => {
+                        written += n;
+                    }
+                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                        if written > 0 {
+                            let _ = self.write_buf.split_to(written);
+                        }
+                        break;
+                    }
+                    Err(_) => return Err(()),
+                }
+            }
+            if written > 0 {
+                if written == len {
+                    unsafe { self.write_buf.set_len(0) }
+                } else {
+                    let _ = self.write_buf.split_to(written);
+                }
+            }
+        }
+        Ok(Async::NotReady)
+    }
+}
+
+fn main() -> io::Result<()> {
+    let sys = actix_rt::System::builder().stop_on_panic(false).build();
+
+    // start http server
+    Server::build()
+        .backlog(1024)
+        .bind("techempower", "0.0.0.0:8080", || {
+            service_fn(|io: Io<TcpStream, _>| App {
+                io: io.into_parts().0,
+                read_buf: BytesMut::with_capacity(32_768),
+                write_buf: BytesMut::with_capacity(32_768),
+                codec: h1::Codec::default(),
+            })
+        })?
+        .start();
+
+    println!("Started http server: 127.0.0.1:8080");
+    sys.run()
+}