split actix to a simple and orm benchmarks; added pg raw benchmark (#3243)

frameworks/Rust/actix/Cargo.toml

@@ -3,6 +3,18 @@ name = "actix"
 version = "0.1.1"
 build = "build.rs"
 
+[[bin]]
+name = "actix"
+path = "src/simple.rs"
+
+[[bin]]
+name = "actix-diesel"
+path = "src/main.rs"
+
+[[bin]]
+name = "actix-pg"
+path = "src/pg.rs"
+
 [dependencies]
 askama = "0.5"
 serde = "1.0"
@@ -15,6 +27,7 @@ http = "0.1"
 actix = "^0.4.3"
 actix-web = "=0.3.1"
 diesel = { version = "1.1", features = ["postgres"] }
+postgres = "0.15"
 
 [build-dependencies]
 askama = "0.5"

frameworks/Rust/actix/benchmark_config.json

@@ -5,6 +5,23 @@
       "setup_file": "setup",
       "json_url": "/json",
       "plaintext_url": "/plaintext",
+      "port": 8080,
+      "approach": "Realistic",
+      "classification": "Micro",
+      "database": "Postgres",
+      "framework": "actix",
+      "language": "Rust",
+      "orm": "Raw",
+      "platform": "None",
+      "webserver": "actix-web",
+      "os": "Linux",
+      "database_os": "Linux",
+      "display_name": "actix",
+      "notes": "",
+      "versus": ""
+    },
+    "diesel": {
+      "setup_file": "setup_diesel",
       "db_url": "/db",
       "fortune_url": "/fortune",
       "query_url": "/queries?q=",
@@ -20,7 +37,28 @@
       "webserver": "actix-web",
       "os": "Linux",
       "database_os": "Linux",
-      "display_name": "actix",
+      "display_name": "actix-diesel",
+      "notes": "",
+      "versus": ""
+    },
+    "pg": {
+      "setup_file": "setup_pg",
+      "db_url": "/db",
+      "fortune_url": "/fortune",
+      "query_url": "/queries?q=",
+      "update_url": "/updates?q=",
+      "port": 8080,
+      "approach": "Realistic",
+      "classification": "Micro",
+      "database": "Postgres",
+      "framework": "actix",
+      "language": "Rust",
+      "orm": "Raw",
+      "platform": "None",
+      "webserver": "actix-web",
+      "os": "Linux",
+      "database_os": "Linux",
+      "display_name": "actix-pg",
       "notes": "",
       "versus": ""
     }

frameworks/Rust/actix/setup_diesel.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+fw_depends postgresql rust
+
+cargo clean
+RUSTFLAGS="-C target-cpu=native" cargo build --release
+./target/release/actix-diesel &

frameworks/Rust/actix/setup_pg.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+fw_depends postgresql rust
+
+cargo clean
+RUSTFLAGS="-C target-cpu=native" cargo build --release
+./target/release/actix-pg &

frameworks/Rust/actix/src/main.rs

@@ -26,26 +26,6 @@ struct State {
     db: SyncAddress<db::DbExecutor>
 }
 
-fn json(_: HttpRequest<State>) -> Result<HttpResponse> {
-    let message = models::Message {
-        message: "Hello, World!"
-    };
-    let body = serde_json::to_string(&message)?;
-
-    Ok(httpcodes::HTTPOk
-       .build()
-       .header(header::SERVER, "Actix")
-       .content_type("application/json")
-       .body(body)?)
-}
-
-fn plaintext(_: HttpRequest<State>) -> Result<HttpResponse> {
-    Ok(httpcodes::HTTPOk.build()
-       .header(header::SERVER, "Actix")
-       .content_type("text/plain")
-       .body("Hello, World!")?)
-}
-
 fn world_row(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
     req.state().db.call_fut(db::RandomWorld)
         .from_err()
@@ -193,8 +173,6 @@ fn main() {
     // start http server
     HttpServer::new(
         move || Application::with_state(State{db: addr.clone()})
-            .resource("/json", |r| r.f(json))
-            .resource("/plaintext", |r| r.f(plaintext))
             .resource("/db", |r| r.route().a(world_row))
             .resource("/queries", |r| r.route().a(queries))
             .resource("/fortune", |r| r.route().a(fortune))

frameworks/Rust/actix/src/pg.rs

@@ -0,0 +1,293 @@
+extern crate actix;
+extern crate actix_web;
+extern crate http;
+extern crate rand;
+extern crate num_cpus;
+extern crate futures;
+extern crate postgres;
+extern crate serde;
+extern crate serde_json;
+#[macro_use] extern crate serde_derive;
+#[macro_use] extern crate askama;
+
+use std::{io, cmp};
+use actix_web::*;
+use actix::prelude::*;
+use askama::Template;
+use http::header;
+use postgres::{Connection, TlsMode};
+use rand::{thread_rng, Rng, ThreadRng};
+use rand::distributions::{Range, IndependentSample};
+use futures::{Future, Stream, stream};
+
+#[derive(Serialize)]
+pub struct World {
+    pub id: i32,
+    pub randomnumber: i32,
+}
+
+#[derive(Serialize)]
+pub struct Fortune {
+    pub id: i32,
+    pub message: String,
+}
+
+
+struct State {
+    db: SyncAddress<PgConnection>
+}
+
+fn world_row(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
+    req.state().db.call_fut(RandomWorld)
+        .from_err()
+        .and_then(|res| {
+            match res {
+                Ok(row) => {
+                    let body = serde_json::to_string(&row).unwrap();
+                    Ok(httpcodes::HTTPOk.build()
+                       .header(header::SERVER, "Actix")
+                       .content_type("application/json")
+                       .body(body)?)
+                },
+                Err(_) => Ok(httpcodes::HTTPInternalServerError.into()),
+            }
+        })
+        .responder()
+}
+
+fn queries(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
+    // get queries parameter
+    let q = if let Some(q) = req.query().get("q") {
+        q.parse::<usize>().ok().unwrap_or(1)
+    } else {
+        1
+    };
+    let q = cmp::min(500, cmp::max(1, q));
+
+    // run sql queries
+    let stream = (0..q).map(|_| req.state().db.call_fut(RandomWorld));
+    stream::futures_unordered(stream)
+        .from_err()
+        .fold(Vec::with_capacity(q), |mut list, val|
+              match val {
+                  Ok(val) => {
+                      list.push(val);
+                      Ok(list)
+                  },
+                  Err(e) => Err(e)
+              }
+        )
+        .and_then(|res| {
+            let body = serde_json::to_string(&res).unwrap();
+            Ok(httpcodes::HTTPOk.build()
+               .header(header::SERVER, "Actix")
+               .content_type("application/json")
+               .content_encoding(headers::ContentEncoding::Identity)
+               .body(body)?)
+        })
+        .responder()
+}
+
+fn updates(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
+    // get queries parameter
+    let q = if let Some(q) = req.query().get("q") {
+        q.parse::<usize>().ok().unwrap_or(1)
+    } else {
+        1
+    };
+    let q = cmp::min(500, cmp::max(1, q));
+
+    // get rundom world objects
+    let mut stream = Vec::with_capacity(q);
+    for _ in 0..q {
+        stream.push(req.state().db.call_fut(RandomWorld));
+    }
+    stream::futures_unordered(stream)
+        .from_err()
+        .fold(Vec::with_capacity(q), |mut list, val|
+              match val {
+                  Ok(val) => {
+                      list.push(val);
+                      Ok(list)
+                  },
+                  Err(e) => Err(e)
+              }
+        )
+        .and_then(move |mut worlds| {
+            // update worlds
+            let mut rng = rand::thread_rng();
+            let between = Range::new(1, 10_000);
+            for world in &mut worlds {
+                world.randomnumber = between.ind_sample(&mut rng);
+            }
+            let body = serde_json::to_string(&worlds).unwrap();
+
+            // persist to db
+            req.state().db.call_fut(UpdateWorld(worlds))
+                .from_err()
+                .and_then(move |res| {
+                    if res.is_ok() {
+                        Ok(httpcodes::HTTPOk.build()
+                           .header(header::SERVER, "Actix")
+                           .content_type("application/json")
+                           .content_encoding(headers::ContentEncoding::Identity)
+                           .body(body)?)
+                    } else {
+                        Ok(httpcodes::HTTPInternalServerError.into())
+                    }
+                })
+        })
+        .responder()
+}
+
+#[derive(Template)]
+#[template(path = "fortune.html")]
+struct FortuneTemplate<'a> {
+    items: &'a Vec<Fortune>,
+}
+
+fn fortune(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
+    req.state().db.call_fut(TellFortune)
+        .from_err()
+        .and_then(|res| {
+            match res {
+                Ok(rows) => {
+                    let tmpl = FortuneTemplate { items: &rows };
+                    let res = tmpl.render().unwrap();
+
+                    Ok(httpcodes::HTTPOk.build()
+                       .header(header::SERVER, "Actix")
+                       .content_type("text/html; charset=utf-8")
+                       .content_encoding(headers::ContentEncoding::Identity)
+                       .body(res)?)
+                },
+                Err(_) => Ok(httpcodes::HTTPInternalServerError.into())
+            }
+        })
+        .responder()
+}
+
+/// Postgres interface
+struct PgConnection {
+    conn: Connection,
+    rng: ThreadRng,
+}
+
+//unsafe impl Send for PgConnection {}
+//unsafe impl Sync for PgConnection {}
+
+impl Actor for PgConnection {
+    type Context = SyncContext<Self>;
+}
+
+impl PgConnection {
+    pub fn new(db_url: &str) -> PgConnection {
+        let conn = Connection::connect(db_url, TlsMode::None)
+            .expect(&format!("Error connecting to {}", db_url));
+        PgConnection{
+            conn: conn,
+            rng: thread_rng(),
+        }
+    }
+}
+
+unsafe impl Send for PgConnection {}
+
+
+pub struct RandomWorld;
+
+impl ResponseType for RandomWorld {
+    type Item = World;
+    type Error = io::Error;
+}
+
+impl Handler<RandomWorld> for PgConnection {
+    type Result = MessageResult<RandomWorld>;
+
+    fn handle(&mut self, _: RandomWorld, _: &mut Self::Context) -> Self::Result {
+        let random_world = self.conn.prepare_cached(
+            "SELECT id, randomnumber FROM World WHERE id=$1").unwrap();
+
+        let random_id = self.rng.gen_range::<i32>(1, 10_000);
+        for row in &random_world.query(&[&random_id]).unwrap() {
+            return Ok(World {id: row.get(0), randomnumber: row.get(1)})
+        }
+
+        Err(io::Error::new(io::ErrorKind::Other, format!("Database error")))
+    }
+}
+
+pub struct UpdateWorld(pub Vec<World>);
+
+impl ResponseType for UpdateWorld {
+    type Item = ();
+    type Error = ();
+}
+
+impl Handler<UpdateWorld> for PgConnection {
+    type Result = ();
+
+    fn handle(&mut self, msg: UpdateWorld, _: &mut Self::Context) {
+        let update_world = self.conn.prepare_cached(
+            "UPDATE World SET randomnumber=$1 WHERE id=$2").unwrap();
+
+        for world in &msg.0 {
+            let _ = update_world.execute(&[&world.randomnumber, &world.id]);
+        }
+    }
+}
+
+pub struct TellFortune;
+
+impl ResponseType for TellFortune {
+    type Item = Vec<Fortune>;
+    type Error = io::Error;
+}
+
+impl Handler<TellFortune> for PgConnection {
+    type Result = io::Result<Vec<Fortune>>;
+
+    fn handle(&mut self, _: TellFortune, _: &mut Self::Context) -> Self::Result {
+        let fortune = self.conn.prepare_cached("SELECT id, message FROM Fortune").unwrap();
+
+        let mut items = Vec::with_capacity(13);
+        items.push(
+            Fortune{id: 0,
+                    message: "Additional fortune added at request time.".to_string()});
+
+        for row in &fortune.query(&[])? {
+            items.push(Fortune{id: row.get(0), message: row.get(1)});
+        }
+        items.sort_by(|it, next| it.message.cmp(&next.message));
+        Ok(items)
+    }
+}
+
+
+fn main() {
+    let sys = System::new("techempower");
+    let dbhost = match option_env!("DBHOST") {
+        Some(it) => it,
+        _ => "127.0.0.1"
+    };
+    let db_url = format!(
+        "postgres://benchmarkdbuser:benchmarkdbpass@{}/hello_world", dbhost);
+
+    // Start db executor actors
+    let addr = SyncArbiter::start(
+        num_cpus::get() * 3, move || PgConnection::new(&db_url));
+
+    // start http server
+    HttpServer::new(
+        move || Application::with_state(State{db: addr.clone()})
+            .resource("/db", |r| r.route().a(world_row))
+            .resource("/queries", |r| r.route().a(queries))
+            .resource("/fortune", |r| r.route().a(fortune))
+            .resource("/updates", |r| r.route().a(updates)))
+        .backlog(8192)
+        .bind("0.0.0.0:8080").unwrap()
+        .start();
+
+    println!("Started http server: 127.0.0.1:8080");
+    let _ = sys.run();
+}

frameworks/Rust/actix/src/simple.rs

@@ -0,0 +1,57 @@
+extern crate actix;
+extern crate actix_web;
+extern crate http;
+extern crate futures;
+extern crate serde;
+extern crate serde_json;
+#[macro_use] extern crate serde_derive;
+
+use actix_web::*;
+use actix::prelude::*;
+use http::StatusCode;
+use http::header::{self, HeaderValue};
+
+
+#[derive(Serialize, Deserialize)]
+pub struct Message {
+    pub message: &'static str,
+}
+
+fn json(_: HttpRequest) -> HttpResponse {
+    let message = Message {
+        message: "Hello, World!"
+    };
+    let body = serde_json::to_string(&message).unwrap();
+
+    let mut resp = HttpResponse::new(StatusCode::OK, body.into());
+    resp.headers_mut().insert(
+        header::SERVER, HeaderValue::from_static("Actix"));
+    resp.headers_mut().insert(
+        header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
+    resp
+}
+
+fn plaintext(_: HttpRequest) -> HttpResponse {
+    let mut resp = HttpResponse::new(StatusCode::OK, "Hello, World!".into());
+    resp.headers_mut().insert(
+        header::SERVER, HeaderValue::from_static("Actix"));
+    resp.headers_mut().insert(
+        header::CONTENT_TYPE, HeaderValue::from_static("text/plain"));
+    resp
+}
+
+fn main() {
+    let sys = System::new("techempower");
+
+    // start http server
+    HttpServer::new(
+        move || Application::new()
+            .resource("/json", |r| r.f(json))
+            .resource("/plaintext", |r| r.f(plaintext)))
+        .backlog(8192)
+        .bind("0.0.0.0:8080").unwrap()
+        .start();
+
+    println!("Started http server: 127.0.0.1:8080");
+    let _ = sys.run();
+}