Refactored simulation client. Added keyed subscription substitute. Refactored app entry point. Added args. Added requirements.txt.

Author: adrian

requirements.txt

@@ -0,0 +1,11 @@
+ray==0.8.0
+sanic==19.9.0
+numpy==1.17.3
+tensorflow==2.0.0
+torch==1.3.1
+gym==0.15.4
+pulsar-client==2.4.0
+protobuf==3.11.0
+apache-bookkeeper-client==4.10.0
+
+

src/backend/messaging/__init__.py

@@ -0,0 +1,63 @@
+import pulsar
+from pulsar.schema import *
+from schema.simulation import SimulationRequestV1, CycleV1
+import logging
+import uuid
+import ray
+
+@ray.remote
+class SimulationClient:
+ client = None
+ consumer = None
+ def __init__(self, request):
+ self.client = pulsar.Client('pulsar://localhost:6650')
+ self.logger = logging.getLogger(__name__)
+ self.consumer = self.client.subscribe("simulation-result", subscription_name="simulator-%s" % request.simId, schema=JsonSchema(CycleV1))
+
+ def __del__(self):
+ if self.client:
+ self.consumer.close()
+ self.client.close()
+ if ray.is_initialized():
+ ray.shutdown()
+
+ def request(self, request):
+ producer = self.client.create_producer('simulation-request', schema=JsonSchema(SimulationRequestV1))
+ producer.send(request)
+
+ def listen(self, request):
+ msg = self.consumer.receive()
+ cycle = msg.value()
+ try:
+ # todo use a topic for each or key based subscription would be better
+ # python client not caught up to java client yet
+ if msg.partition_key() == request.simId:
+ self.consumer.acknowledge(msg)
+ return cycle
+ except Exception as e:
+ print(e)
+ self.consumer.negative_acknowledge(msg)
+ return None
+
+ @staticmethod
+ def requestToRequest(r):
+ j = json.loads(r)
+ return SimulationRequestV1(**j)
+
+ @staticmethod
+ async def run(request, ws):
+ if not ray.is_initialized():
+ ray.init()
+ while True:
+ msg = await ws.recv()
+ msg = SimulationClient.requestToRequest(msg)
+ sim = SimulationClient.remote(msg)
+ sim.request.remote(msg)
+ for i in range(msg.cycles):
+ result = ray.get(sim.listen.remote(msg))
+ if result:
+ await ws.send(json.dumps(result.__dict__))
+ else:
+ i=i-1
+ await ws.close()
+ del sim

src/backend/messaging/client.py

@@ -3,63 +3,20 @@
 from sanic.websocket import WebSocketProtocol
 from os.path import dirname, abspath, join
 import json
-import pulsar
-from pulsar.schema import *
-# todo make library
+
 from schema.simulation import SimulationRequestV1, FoodV1, AgentV1, CycleV1, ToDTO
-import logging
-import uuid
+from backend.messaging.client import SimulationClient
+import ray
 
 _CURDIR = dirname(abspath(__file__))
 
 app = Sanic()
 app.static('/', join(_CURDIR, './dist'))
 
-class SimulationClient:
- def __init__(self):
- self.client = pulsar.Client('pulsar://localhost:6650')
- self.logger = logging.getLogger(__name__)
-
- def __del__(self):
- self.client.close()
-
- def request(self, request):
- producer = self.client.create_producer('simulation-request', schema=JsonSchema(SimulationRequestV1))
- producer.send(request)
- return request.simId
-
- async def listen(self, request, ws):
- consumer = self.client.subscribe("simulation-result", subscription_name="simulator", schema=JsonSchema(CycleV1))
- for _ in range(request.cycles):
- msg = consumer.receive()
- cycle = msg.value()
- try:
- # todo use a topic for each or key based subscription would be better
- # if msg.partition_key() == request.simId | 1 == 1:
- consumer.acknowledge(msg)
- # Acknowledge successful processing of the message
- await ws.send(json.dumps(cycle.__dict__))
- #else:
- # consumer.negative_acknowledge(msg)
- except Exception as e:
- # Message failed to be processed
- print(e)
- consumer.negative_acknowledge(msg)
- consumer.close()
-
-def requestToRequest(r):
- j = json.loads(r)
- return SimulationRequestV1(**j)
-
-sim = SimulationClient()
-
 @app.websocket('/environment')
 async def environment(request, ws):
- while True:
- msg = await ws.recv()
- msg = requestToRequest(msg)
- id = sim.request(msg)
- await sim.listen(msg, ws)
+ await SimulationClient.run(request, ws)
+ del sim
 
 def start():
  app.run(host="0.0.0.0", port=3000, protocol=WebSocketProtocol, debug=True)

src/backend/server.py

@@ -2,7 +2,21 @@
 import sys
 from simulator.messaging.messaging import MessageClient 
 from schema.simulation import SimulationRequestV1, FoodV1, AgentV1, ToDTO
+from backend.server import start
+import argparse
+
+def getArgs():
+ parser = argparse.ArgumentParser(description='Start frontend or workers. Add --frontend for server.')
+ parser.add_argument('--frontend',
+ default=False, 
+ action="store_true",
+ help='start socket server and front end app')
+ return parser.parse_args()
 
 if __name__ == "__main__":
- client = MessageClient()
- client.listen()
+ args = getArgs()
+ if not args.frontend:
+ client = MessageClient()
+ client.listen()
+ else:
+ start()

src/main.py

@@ -32,10 +32,8 @@ def runSimulation(self, request : SimulationRequestV1):
  producer = self.client.create_producer('simulation-result', schema=JsonSchema(CycleV1))
  for i in range(request.cycles):
  cycle = sim.step(i)
- print(cycle)
  cyclev1 = ToDTO.toCycle(cycle)
- print("SENGING")
- producer.send(cyclev1)
+ producer.send(cyclev1, partition_key=request.simId, sequence_id=i)
 
  # subscribe to a simulation request topic
  def listen(self):