Merge branch 'adn' into main

Cargo.toml

@@ -34,9 +34,7 @@ members = [
   # tools
   "tools/phoenix_cargo",
 ]
-exclude = [
-  "experimental/mrpc",
-]
+exclude = ["experimental/mrpc"]
 
 [workspace.dependencies]
 phoenix-api = { path = "src/phoenix-api" }

eval/policy/delay/attach.toml

@@ -0,0 +1,12 @@
+addon_engine = "DelayEngine"
+tx_channels_replacements = [
+    ["MrpcEngine", "DelayEngine", 0, 0],
+    ["DelayEngine", "TcpRpcAdapterEngine", 0, 0],
+]
+rx_channels_replacements = []
+group = ["MrpcEngine", "TcpRpcAdapterEngine"]
+op = "attach"
+config_string = '''
+delay_probability = 0.2
+delay_ms = 100
+'''

eval/policy/delay/collect.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+from typing import List
+import glob
+import sys
+
+OD = "/tmp/mrpc-eval"
+if len(sys.argv) >= 2:
+    OD = sys.argv[1]
+
+
+def convert_msg_size(s: str) -> int:
+    if s.endswith('gb'):
+        return int(s[:-2]) * 1024 * 1024 * 1024
+    if s.endswith('mb'):
+        return int(s[:-2]) * 1024 * 1024
+    if s.endswith('kb'):
+        return int(s[:-2]) * 1024
+    if s.endswith('b'):
+        return int(s[:-1])
+
+    raise ValueError(f"unknown input: {s}")
+
+
+def get_rate(path: str) -> List[float]:
+    rates = []
+    with open(path, 'r') as fin:
+        for line in fin:
+            words = line.strip().split(' ')
+            if words[-3] == 'rps,':
+                rate = float(words[-4])
+                rates.append(rate)
+    return rates[1:]
+
+
+def load_result(sol_before, sol_after, f: str):
+    # print(f)
+    rates = get_rate(f)
+    before = rates[5:25]
+    after = rates[-25:-5]
+    for r in before:
+        print(f'{round(r/1000,2)},{sol_before},w/o Fault')
+    for r in after:
+        print(f'{round(r/1000,2)},{sol_after},w/ Fault')
+
+
+for f in glob.glob(OD+"/policy/delay/rpc_bench_tput_32b/rpc_bench_client_danyang-04.stdout"):
+    load_result('mRPC', 'Native mRPC', f)

eval/policy/delay/config.toml

@@ -0,0 +1,9 @@
+workdir = "~/nfs/Developing/livingshade/phoenix/experimental/mrpc"
+
+[env]
+RUST_BACKTRACE = "1"
+RUST_LOG_STYLE = "never"
+CARGO_TERM_COLOR = "never"
+PHOENIX_LOG = "info"
+PROTOC = "/usr/bin/protoc"
+PHOENIX_PREFIX = "/tmp/phoenix"

eval/policy/delay/detach.toml

@@ -0,0 +1,4 @@
+addon_engine = "DelayEngine"
+tx_channels_replacements = [["MrpcEngine", "TcpRpcAdapterEngine", 0, 0]]
+rx_channels_replacements = [["TcpRpcAdapterEngine", "MrpcEngine", 0, 0]]
+op = "detach"

eval/policy/delay/rpc_bench_tput_32b.toml

@@ -0,0 +1,15 @@
+name = "policy/delay/rpc_bench_tput_32b"
+description = "Run rpc_bench benchmark"
+group = "delay"
+timeout_secs = 70
+
+[[worker]]
+host = "danyang-06"
+bin = "rpc_bench_server"
+args = "--port 5002 -l info --transport tcp"
+
+[[worker]]
+host = "danyang-04"
+bin = "rpc_bench_client"
+args = "--transport tcp -c rdma0.danyang-06 --concurrency 128 --req-size 32 --duration 65 -i 1 --port 5002 -l error"
+dependencies = [0]

eval/policy/delay/start_traffic.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+trap "trap - SIGTERM && kill -- -$$" SIGINT SIGTERM SIGHUP EXIT
+
+OD=/tmp/mrpc-eval
+if [[ $# -ge 1 ]]; then
+    OD=$1
+fi
+
+WORKDIR=$(dirname $(realpath $0))
+cd $WORKDIR
+
+# concurrency = 128
+cargo rr --bin launcher -- --output-dir ${OD} --timeout=120 --benchmark ./rpc_bench_tput_32b.toml --configfile ./config.toml &
+
+sleep 30
+
+LIST_OUTPUT="${OD}"/policy/list.json
+cargo rr --bin list -- --dump "${LIST_OUTPUT}" # Need to specifiy PHOENIX_PREFIX
+cat "${LIST_OUTPUT}"
+ARG_PID=$(cat "${LIST_OUTPUT}" | jq '.[] | select(.service == "Mrpc") | .pid')
+ARG_SID=$(cat "${LIST_OUTPUT}" | jq '.[] | select(.service == "Mrpc") | .sid')
+echo $ARG_SID
+
+sleep 1
+
+cargo run --bin addonctl -- --config ./attach.toml --pid ${ARG_PID} --sid ${ARG_SID} # Need to specifiy PHOENIX_PREFIX
+
+wait

eval/policy/fault/attach.toml

@@ -0,0 +1,33 @@
+addon_engine = "FaultEngine"
+tx_channels_replacements = [
+    [
+        "MrpcEngine",
+        "FaultEngine",
+        0,
+        0,
+    ],
+    [
+        "FaultEngine",
+        "TcpRpcAdapterEngine",
+        0,
+        0,
+    ],
+]
+rx_channels_replacements = [
+    [
+        "TcpRpcAdapterEngine",
+        "FaultEngine",
+        0,
+        0,
+    ],
+    [
+        "FaultEngine",
+        "MrpcEngine",
+        0,
+        0,
+    ],
+]
+group = ["MrpcEngine", "TcpRpcAdapterEngine"]
+op = "attach"
+config_string = '''
+'''

eval/policy/fault/collect.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+from typing import List
+import glob
+import sys
+
+OD = "/tmp/mrpc-eval"
+if len(sys.argv) >= 2:
+    OD = sys.argv[1]
+
+
+def convert_msg_size(s: str) -> int:
+    if s.endswith('gb'):
+        return int(s[:-2]) * 1024 * 1024 * 1024
+    if s.endswith('mb'):
+        return int(s[:-2]) * 1024 * 1024
+    if s.endswith('kb'):
+        return int(s[:-2]) * 1024
+    if s.endswith('b'):
+        return int(s[:-1])
+
+    raise ValueError(f"unknown input: {s}")
+
+
+def get_rate(path: str) -> List[float]:
+    rates = []
+    with open(path, 'r') as fin:
+        for line in fin:
+            words = line.strip().split(' ')
+            if words[-3] == 'rps,':
+                rate = float(words[-4])
+                rates.append(rate)
+    return rates[1:]
+
+
+def load_result(sol_before, sol_after, f: str):
+    # print(f)
+    rates = get_rate(f)
+    before = rates[5:25]
+    after = rates[-25:-5]
+    for r in before:
+        print(f'{round(r/1000,2)},{sol_before},w/o Fault')
+    for r in after:
+        print(f'{round(r/1000,2)},{sol_after},w/ Fault')
+
+
+for f in glob.glob(OD+"/policy/fault/rpc_bench_tput_32b/rpc_bench_client_danyang-04.stdout"):
+    load_result('mRPC', 'ADN+mRPC', f)

eval/policy/fault/config.toml

@@ -0,0 +1,9 @@
+workdir = "~/nfs/Developing/livingshade/phoenix/experimental/mrpc"
+
+[env]
+RUST_BACKTRACE = "1"
+RUST_LOG_STYLE = "never"
+CARGO_TERM_COLOR = "never"
+PHOENIX_LOG = "info"
+PROTOC = "/usr/bin/protoc"
+PHOENIX_PREFIX = "/tmp/phoenix"

eval/policy/fault/detach.toml

@@ -0,0 +1,4 @@
+addon_engine = "FaultEngine"
+tx_channels_replacements = [["MrpcEngine", "TcpRpcAdapterEngine", 0, 0]]
+rx_channels_replacements = [["TcpRpcAdapterEngine", "MrpcEngine", 0, 0]]
+op = "detach"

eval/policy/fault/rpc_bench_tput_32b.toml

@@ -0,0 +1,15 @@
+name = "policy/fault/rpc_bench_tput_32b"
+description = "Run rpc_bench benchmark"
+group = "fault"
+timeout_secs = 70
+
+[[worker]]
+host = "danyang-06"
+bin = "rpc_bench_server"
+args = "--port 5002 -l info --transport tcp"
+
+[[worker]]
+host = "danyang-04"
+bin = "rpc_bench_client"
+args = "--transport tcp -c rdma0.danyang-06 --concurrency 128 --req-size 32 --duration 65 -i 1 --port 5002 -l error"
+dependencies = [0]

eval/policy/fault/start_traffic.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+trap "trap - SIGTERM && kill -- -$$" SIGINT SIGTERM SIGHUP EXIT
+
+OD=/tmp/mrpc-eval
+if [[ $# -ge 1 ]]; then
+    OD=$1
+fi
+
+WORKDIR=$(dirname $(realpath $0))
+cd $WORKDIR
+
+# concurrency = 128
+cargo rr --bin launcher -- --output-dir ${OD} --timeout=120 --benchmark ./rpc_bench_tput_32b.toml --configfile ./config.toml &
+
+sleep 30
+
+LIST_OUTPUT="${OD}"/policy/list.json
+cargo rr --bin list -- --dump "${LIST_OUTPUT}" # Need to specifiy PHOENIX_PREFIX
+cat "${LIST_OUTPUT}"
+ARG_PID=$(cat "${LIST_OUTPUT}" | jq '.[] | select(.service == "Mrpc") | .pid')
+ARG_SID=$(cat "${LIST_OUTPUT}" | jq '.[] | select(.service == "Mrpc") | .sid')
+echo $ARG_SID
+
+sleep 1
+
+cargo run --bin addonctl -- --config ./attach.toml --pid ${ARG_PID} --sid ${ARG_SID} # Need to specifiy PHOENIX_PREFIX
+
+wait

eval/policy/fault2/attach.toml

@@ -0,0 +1,33 @@
+addon_engine = "Fault2Engine"
+tx_channels_replacements = [
+    [
+        "MrpcEngine",
+        "Fault2Engine",
+        0,
+        0,
+    ],
+    [
+        "Fault2Engine",
+        "TcpRpcAdapterEngine",
+        0,
+        0,
+    ],
+]
+rx_channels_replacements = [
+    [
+        "TcpRpcAdapterEngine",
+        "Fault2Engine",
+        0,
+        0,
+    ],
+    [
+        "Fault2Engine",
+        "MrpcEngine",
+        0,
+        0,
+    ],
+]
+group = ["MrpcEngine", "TcpRpcAdapterEngine"]
+op = "attach"
+config_string = '''
+'''

eval/policy/fault2/collect.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python3
+from typing import List
+import glob
+import sys
+
+OD = "/tmp/mrpc-eval"
+if len(sys.argv) >= 2:
+    OD = sys.argv[1]
+
+
+def convert_msg_size(s: str) -> int:
+    if s.endswith('gb'):
+        return int(s[:-2]) * 1024 * 1024 * 1024
+    if s.endswith('mb'):
+        return int(s[:-2]) * 1024 * 1024
+    if s.endswith('kb'):
+        return int(s[:-2]) * 1024
+    if s.endswith('b'):
+        return int(s[:-1])
+
+    raise ValueError(f"unknown input: {s}")
+
+
+def get_rate(path: str) -> List[float]:
+    rates = []
+    with open(path, 'r') as fin:
+        for line in fin:
+            words = line.strip().split(' ')
+            if words[-3] == 'rps,':
+                rate = float(words[-4])
+                rates.append(rate)
+    return rates[1:]
+
+
+def load_result(sol_before, sol_after, f: str):
+    # print(f)
+    rates = get_rate(f)
+    before = rates[5:25]
+    after = rates[-25:-5]
+    for r in before:
+        print(f'{round(r/1000,2)},{sol_before},w/o Fault')
+    for r in after:
+        print(f'{round(r/1000,2)},{sol_after},w/ Fault')
+
+
+for f in glob.glob(OD+"/policy/fault2/rpc_bench_tput_32b/rpc_bench_client_danyang-04.stdout"):
+    load_result('mRPC', 'Native mRPC', f)

eval/policy/fault2/config.toml

@@ -0,0 +1,9 @@
+workdir = "~/nfs/Developing/livingshade/phoenix/experimental/mrpc"
+
+[env]
+RUST_BACKTRACE = "1"
+RUST_LOG_STYLE = "never"
+CARGO_TERM_COLOR = "never"
+PHOENIX_LOG = "info"
+PROTOC = "/usr/bin/protoc"
+PHOENIX_PREFIX = "/tmp/phoenix"

eval/policy/fault2/detach.toml

@@ -0,0 +1,4 @@
+addon_engine = "Fault2Engine"
+tx_channels_replacements = [["MrpcEngine", "TcpRpcAdapterEngine", 0, 0]]
+rx_channels_replacements = [["TcpRpcAdapterEngine", "MrpcEngine", 0, 0]]
+op = "detach"

eval/policy/fault2/rpc_bench_tput_32b.toml

@@ -0,0 +1,15 @@
+name = "policy/fault2/rpc_bench_tput_32b"
+description = "Run rpc_bench benchmark"
+group = "fault2"
+timeout_secs = 70
+
+[[worker]]
+host = "danyang-06"
+bin = "rpc_bench_server"
+args = "--port 5002 -l info --transport tcp"
+
+[[worker]]
+host = "danyang-04"
+bin = "rpc_bench_client"
+args = "--transport tcp -c rdma0.danyang-06 --concurrency 128 --req-size 32 --duration 65 -i 1 --port 5002 -l error"
+dependencies = [0]