Merge pull request #15 from josehu07/reed-solomon

TCP write retrying bug fix & adaptive bench client

.gitignore

@@ -9,3 +9,5 @@
 /tla+/**/states/
 /tla+/**/*.out
 /tla+/**/*.old
+
+/results/

Cargo.toml

@@ -12,7 +12,7 @@ authors = ["Guanzhou Jose Hu <huguanzhou123@gmail.com>"]
 async-trait = "0.1"
 fixedbitset = "0.4"
 flashmap = "0.1"
-bytes = "1.4"
+bytes = { version = "1.4", features = ["serde"] }
 futures = "0.3"
 tokio = { version = "1.29", features = ["full"] }
 rand = "0.8"
@@ -21,3 +21,4 @@ rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }
 toml = { version = "0.7", features = ["parse"] }
 log = "0.4"
+reed-solomon-erasure = { version = "6.0", features = ["simd-accel"] }

README.md

@@ -37,6 +37,12 @@ These design choices make protocol implementation in Summerset surprisingly stra
 
 ## Build
 
+Install the [Rust toolchain](https://rustup.rs/) if haven't. For \*nix:
+
+```bash
+curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
+```
+
 Build everything in debug or release (`-r`) mode:
 
 ```bash
@@ -111,7 +117,6 @@ Complete cluster management and benchmarking scripts are available in another re
   - [ ] separate commit vs. exec responses?
   - [ ] membership discovery & view changes
 - [ ] implementation of Raft
-- [ ] implementation of Crossword prototype
 - [x] client-side utilities
   - [x] REPL-style client
   - [x] random benchmarking client

scripts/local_bench.tmp.py

@@ -0,0 +1,119 @@
+import subprocess
+import time
+import statistics
+
+
+def do_cargo_build():
+    print("Building everything...")
+    cmd = ["cargo", "build", "--workspace", "-r"]
+    proc = subprocess.Popen(cmd)
+    proc.wait()
+
+
+def run_process(cmd):
+    # print("Run:", " ".join(cmd))
+    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return proc
+
+
+def kill_all_matching(name):
+    # print("Kill all:", name)
+    assert name.count(" ") == 0
+    cmd = ["pkill", "-9", "-f", name]
+    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    proc.wait()
+
+
+def launch_cluster(protocol, num_replicas):
+    cmd = [
+        "python3",
+        "./scripts/local_cluster.py",
+        "-p",
+        protocol,
+        "-n",
+        str(num_replicas),
+        "-r",
+    ]
+    return run_process(cmd)
+
+
+def run_bench_client(protocol, value_size, put_ratio, length_s):
+    cmd = [
+        "python3",
+        "./scripts/local_client.py",
+        "-p",
+        protocol,
+        "-r",
+        "bench",
+        "-v",
+        str(value_size),
+        "-w",
+        str(put_ratio),
+        "-l",
+        str(length_s),
+    ]
+    return run_process(cmd)
+
+
+def parse_output(output):
+    lines = [l.strip() for l in output.split("\n") if l.count("|") == 3]
+    assert len(lines) >= 4
+    assert lines[0].startswith("Elapsed")
+    lines = lines[1:]
+
+    warmup, tail = len(lines) // 3, len(lines) // 10
+    lines = lines[warmup:-tail]
+
+    tpts, lats = [], []
+    for line in lines:
+        segs = line.split()
+        tpt = float(segs[2])  # reqs/s
+        lat = float(segs[4]) / 1000.0  # ms
+        tpts.append(tpt)
+        lats.append(lat)
+
+    median_tpt = tpts[len(tpts) // 2]
+    median_lat = lats[len(lats) // 2]
+    print(f"  med  tpt {median_tpt:9.2f} reqs/s  lat {median_lat:9.2f} ms")
+
+    avg_tpt = sum(tpts) / len(tpts)
+    std_tpt = statistics.stdev(tpts)
+    avg_lat = sum(lats) / len(lats)
+    std_lat = statistics.stdev(lats)
+    print(f"  avg  tpt {avg_tpt:9.2f} reqs/s  lat {avg_lat:9.2f} ms")
+    print(f"  std  tpt {std_tpt:9.2f}         lat {std_lat:9.2f}")
+
+
+def bench_round(protocol, num_replicas, value_size, put_ratio, length_s):
+    print(
+        f"{protocol:<10s}  n={num_replicas:1d}  v={value_size:<9d}  w%={put_ratio:<3d}  {length_s:3d}s"
+    )
+    kill_all_matching("summerset_client")
+    kill_all_matching("summerset_server")
+    kill_all_matching("summerset_manager")
+
+    proc_cluster = launch_cluster(protocol, num_replicas)
+    time.sleep(15)
+
+    proc_client = run_bench_client(protocol, value_size, put_ratio, length_s)
+    out, err = proc_client.communicate()
+
+    proc_cluster.terminate()
+    proc_cluster.wait()
+
+    if proc_client.returncode != 0:
+        print(err.decode())
+    else:
+        parse_output(out.decode())
+
+
+if __name__ == "__main__":
+    do_cargo_build()
+
+    for num_replicas in (3, 7):
+        for value_size in (1024, 65536, 4194304):
+            for protocol in ("MultiPaxos", "RSPaxos"):
+                bench_round(protocol, num_replicas, value_size, 100, 60)
+
+    bench_round("MultiPaxos", 7, 4194304, 10, 60)
+    bench_round("RSPaxos", 7, 4194304, 10, 60)

scripts/local_client.py

@@ -25,6 +25,7 @@ def run_process(cmd):
     "RepNothing": "",
     "SimplePush": "",
     "MultiPaxos": "",
+    "RSPaxos": "",
 }
 
 
@@ -68,6 +69,10 @@ def compose_client_cmd(protocol, manager, config, utility, params, release):
     if len(params) > 0:
         cmd += ["--params", params]
 
+    # if in benchmarking mode, lower the client's CPU scheduling priority
+    if utility == "bench":
+        cmd = ["nice", "-n", "15"] + cmd
+
     return cmd