app.py

from ray import serve
from ray.serve.drivers import DAGDriver
from ray.serve.dag import InputNode
from ray.serve.http_adapters import json_request


@serve.deployment
class A:
    def predict(self, inp):
        return inp

@serve.deployment
def combine(*inps):
    return sum(inps)


As = [A.bind() for _ in range(4)]
with InputNode() as inp:
    dag = combine.bind(*[a.predict.bind(inp) for a in As])
    dag = DAGDriver.bind(dag, http_adapter=json_request)

script.lua

wrk.method = "POST"
wrk.body   = "1"

watch.py

import psutil
import sys
import json
import time

pid = int(sys.argv[1])
proc = psutil.Process(pid)
while True:
    data = proc.memory_info()._asdict()
    data["time"] = time.time()
    print(json.dumps(data))
    time.sleep(2)

serve run app:dag
wrk http://localhost:8000 -s script.lua -d 100000
python watch.py 15396 | tee fixed.log