[WIP] plaintext benchmark

asyncdispatch2.nimble

@@ -15,7 +15,6 @@ task test, "Run all tests":
   exec "nim c -r --gc:markAndSweep tests/testsync"
   exec "nim c -r -d:release tests/testsync"
 
-
   exec "nim c -r -d:useSysAssert -d:useGcAssert tests/testsoon"
   exec "nim c -r tests/testsoon"
   exec "nim c -r --gc:markAndSweep tests/testsoon"
@@ -60,3 +59,7 @@ task test, "Run all tests":
   exec "nim c -r tests/testbugs"
   exec "nim c -r --gc:markAndSweep tests/testbugs"
   exec "nim c -r -d:release tests/testbugs"
+
+task benchmark, "compile bench-bot":
+  exec "nim c -d:release tests/benchmark/bot"
+  exec "tests/benchmark/bot all"

tests/benchmark/.gitignore

@@ -0,0 +1,3 @@
+bot
+wrk
+wrksrc/

tests/benchmark/actix-raw/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "actix"
+version = "0.7.1"
+
+[dependencies]
+futures = "0.1"
+actix = "0.7"
+actix-web = { version="0.7", default-features = false }
+
+[profile.release]
+lto = true
+opt-level = 3
+codegen-units = 1

tests/benchmark/actix-raw/plaintext.dockerfile

@@ -0,0 +1,19 @@
+FROM rust:1.28-slim
+
+RUN apt update -yqq && \
+    apt install -yqq libpq-dev
+
+RUN mkdir -p /opt/actix/src
+RUN touch /opt/actix/src/lib.rs
+
+ADD ./Cargo.toml /opt/actix/
+WORKDIR /opt/actix
+RUN RUSTFLAGS="-C target-cpu=native" cargo build --release
+
+RUN rm -rf /opt/actix/src/
+ADD ./src /opt/actix/src
+
+RUN RUSTFLAGS="-C target-cpu=native" cargo build --release
+
+CMD ./target/release/actix
+# CMD USE_THREADS=1 ./target/release/actix

tests/benchmark/actix-raw/src/main.rs

@@ -0,0 +1,34 @@
+extern crate actix;
+extern crate actix_web;
+extern crate futures;
+
+use std::env;
+
+use actix::prelude::*;
+use actix_web::{http, server, App, HttpRequest, HttpResponse};
+
+fn plaintext(req: &HttpRequest) -> HttpResponse {
+    HttpResponse::build_from(req)
+        .header(http::header::SERVER, "Actix")
+        .header(http::header::CONTENT_TYPE, "text/plain")
+        .body("Hello, World!")
+}
+
+fn main() {
+    let sys = System::new("techempower");
+
+    // start http server
+    let srv = server::new(move || {
+        App::new()
+            .resource("/plaintext", |r| r.f(plaintext))
+    }).backlog(8192);
+
+    if env::var_os("USE_THREADS").is_none() {
+        srv.workers(1)
+    } else {
+        srv
+    }.bind("0.0.0.0:8080").unwrap().start();
+
+    println!("Started http server: 127.0.0.1:8080");
+    let _ = sys.run();
+}

tests/benchmark/asyncdispatch2/build.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+if [ ! -f server ]; then
+  echo "building asyncdispatch2..."
+  nimble install -y https://github.com/status-im/nim-asyncdispatch2 > /dev/null
+  nim c -d:release --verbosity:0 --hints:off --threads:on server.nim
+fi

tests/benchmark/asyncdispatch2/plaintext.dockerfile

@@ -0,0 +1,6 @@
+FROM statusteam/nim-base
+RUN nimble install -y https://github.com/status-im/nim-asyncdispatch2
+WORKDIR /server
+COPY server.nim server.nim
+RUN nim c -d:release --threads:on server.nim
+CMD ["./server"]

tests/benchmark/asyncdispatch2/server.nim

@@ -0,0 +1,161 @@
+import strutils, posix, deques, times, strformat
+import asyncdispatch2, threadpool, os, osproc
+
+proc getFormattedTime*(): string =
+  result = format(getTime().inZone(utc()), "ddd, dd MMM yyyy hh:mm:ss 'GMT'")
+
+type
+  IncomingCtx = ref object
+    buf: string
+    bufLen: int
+    resp: string
+    respLen: int
+
+  ServerCtx = ref object
+    ctxQueue: Deque[IncomingCtx]
+    serverTime: string
+
+proc newIncomingCtx(readSize: int, writeSize: int): IncomingCtx =
+  result = IncomingCtx(
+    buf: newString(readSize),
+    bufLen: 0,
+    resp: newString(writeSize),
+    respLen: 0
+  )
+
+proc newServerCtx*(readSize, writeSize: int, cap: int): ServerCtx =
+  new(result)
+  result.ctxQueue = initDeque[IncomingCtx](cap)
+  var ctxArray = newSeq[IncomingCtx](cap)
+  for i in 0 ..< cap:
+    ctxArray[i] = newIncomingCtx(readSize, writeSize)
+    GC_ref(ctxArray[i])
+    result.ctxQueue.addFirst(ctxArray[i])
+
+proc getServerTime*(srv: ServerCtx): string =
+  result = srv.serverTime
+
+proc updateServerTime*(srv: ServerCtx) =
+  srv.serverTime = getFormattedTime()
+
+proc createCtx*(readSize, writeSize: int): IncomingCtx =
+  result = newIncomingCtx(readSize, writeSize)
+  GC_ref(result)
+
+proc getIncomingCtx*(srv: ServerCtx, readSize, writeSize: int): IncomingCtx =
+  if srv.ctxQueue.len > 0:
+    return srv.ctxQueue.popFirst()
+  else:
+    return createCtx(readSize, writeSize)
+
+proc freeCtx*(srv: ServerCtx, ctx: IncomingCtx) =
+  srv.ctxQueue.addLast(ctx)
+
+proc resetBuffer(ctx: IncomingCtx) =
+  ctx.respLen = 0
+  ctx.bufLen = 0
+
+proc sendMessage(ctx: IncomingCtx, body: string) =
+  let ol = ctx.respLen
+  while unlikely ctx.respLen + body.len > ctx.resp.len:
+    ctx.resp.setLen(ctx.resp.len + ctx.resp.len)
+  copyMem(addr ctx.resp[ol], unsafeAddr body[0], body.len)
+  ctx.respLen += body.len
+
+proc readMessage*(ctx: IncomingCtx, st: StreamTransport): Future[int] {.async.} =
+  let rcvLimit =
+    block:
+      if unlikely(ctx.buf.len - ctx.bufLen == 0):
+        ctx.buf.setLen(ctx.buf.len + ctx.buf.len)
+      ctx.buf.len - ctx.bufLen
+
+  let rcv = await st.readOnce(addr ctx.buf[ctx.bufLen], rcvLimit)
+  ctx.bufLen += rcv
+  return rcv
+
+proc makeResp(serverTime: string): string =
+  result = fmt("HTTP/1.1 200 OK\r\L" &
+               "Date: {serverTime}\r\l" &
+               "Server: asyncdispatch2\r\L" &
+               "Content-Type: text/plain\r\L" &
+               "Content-Length: 13\r\L\r\L" &
+               "Hello, World!")
+
+proc handleIncoming(srv: ServerCtx, ctx: IncomingCtx, st: StreamTransport) {.async.} =
+  try:
+    while true:
+      let rcv = await ctx.readMessage(st)
+      if rcv == 0:
+        st.close()
+        srv.freeCtx(ctx)
+        return
+
+      var pos = 0
+      while (ctx.bufLen - pos) > 3:
+        if ctx.buf[pos] == '\r':
+          if ctx.buf[pos+1] == '\L' and
+            ctx.buf[pos+2] == '\r' and
+            ctx.buf[pos+3] == '\L':
+            ctx.sendMessage(makeResp(srv.serverTime))
+        inc pos
+
+      let wr = await st.write(ctx.resp[0].addr, ctx.respLen)
+      assert wr == ctx.respLen
+      ctx.resetBuffer()
+  except:
+    st.close()
+    srv.freeCtx(ctx)
+
+proc handleConnection(ss: StreamServer, st: StreamTransport) {.async.} =
+  var srv = getUserData[ServerCtx](ss)
+  var ctx = srv.getIncomingCtx(1024, 1024)
+  ctx.resetBuffer()
+  asyncCheck handleIncoming(srv, ctx, st)
+
+proc handleBreak(udata: pointer) =
+  var cdata = cast[ptr CompletionData](udata)
+  var svr = cast[StreamServer](cdata.udata)
+  echo "\nCTRL+C pressed, stopping server..."
+  svr.stop()
+  svr.close()
+
+proc updateTime(arg: pointer = nil) {.gcsafe.} =
+  var svr = cast[ServerCtx](arg)
+  svr.updateServerTime()
+
+proc incomingConnection(onAddress: string) =
+  let
+    ta = initTAddress(onAddress)
+    ctx = newServerCtx(1024, 1024, 128)
+    svr = createStreamServer(ta, handleConnection, {ReuseAddr, ReusePort}, backlog = 128, udata = cast[pointer](ctx))
+  when not defined(windows):
+    discard addSignal(SIGINT, handleBreak, udata = cast[pointer](svr))
+
+  ctx.updateServerTime()
+  addTimer(1000, updateTime, udata = cast[pointer](ctx))
+
+  svr.start()
+  echo "Server started at ", ta
+  waitFor svr.join()
+
+proc runServer() {.thread.} =
+  incomingConnection("0.0.0.0:8080")
+
+proc runSingleThread() =
+  spawn runServer()
+  threadpool.sync()
+
+proc runMultipleThreads() =
+  for _ in 0 ..< countProcessors():
+    spawn runServer()
+  threadpool.sync()
+
+proc main() =
+  if os.getEnv("USE_THREADS") == "1":
+    echo "use threads"
+    runMultipleThreads()
+  else:
+    echo "no threads"
+    runSingleThread()
+
+main()

tests/benchmark/asyncnet/build.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+if [ ! -f server ]; then
+  echo "building asyncnet..."
+  nim c -d:release --verbosity:0 --hints:off --threads:on server.nim
+fi

tests/benchmark/asyncnet/plaintext.dockerfile

@@ -0,0 +1,5 @@
+FROM statusteam/nim-base
+WORKDIR /server
+COPY server.nim server.nim
+RUN nim c -d:release --threads:on server.nim
+CMD ["./server"]