[Python] Add support for multimachine operators.

python/erdos/__init__.py

@@ -18,6 +18,7 @@
 import erdos.utils
 
 _num_py_operators = 0
+node_to_address = dict()
 
 # Set the top-level logger for ERDOS logging.
 # Users can change the logging level to the required level by calling setLevel
@@ -83,6 +84,11 @@ def connect(
     node_id = _num_py_operators
     logger.debug("Connecting operator #{num} ({name}) to the graph.".format(
         num=node_id, name=config.name))
+    if config._address is not None:
+        global node_to_address
+        if len(node_to_address) == 0:
+            node_to_address[0] = config._address  # Configure main node first
+        node_to_address[node_id] = config._address
 
     py_read_streams = []
     op_read_streams = list(
@@ -120,6 +126,8 @@ def reset():
     logger.info("Resetting the default graph.")
     global _num_py_operators
     _num_py_operators = 0
+    global node_to_address
+    node_to_address.clear()
     _internal.reset()
 
 
@@ -192,24 +200,49 @@ def run_async(graph_filename: Optional[str] = None,
         A :py:class:`.NodeHandle` that allows the driver to interface with the
         dataflow graph.
     """
-    data_addresses = [
-        "127.0.0.1:{port}".format(port=start_port + i)
-        for i in range(_num_py_operators + 1)
-    ]
-    control_addresses = [
-        "127.0.0.1:{port}".format(port=start_port + len(data_addresses) + i)
-        for i in range(_num_py_operators + 1)
-    ]
+    global node_to_address
+    multimachine = len(node_to_address) > 0
+
+    if not multimachine:
+        data_addresses = [
+            "127.0.0.1:{port}".format(port=start_port + i)
+            for i in range(_num_py_operators + 1)
+        ]
+        control_addresses = [
+            "127.0.0.1:{port}".format(port=start_port + len(data_addresses) +
+                                      i) for i in range(_num_py_operators + 1)
+        ]
+    else:
+        data_addresses = [
+            node_to_address[i] + ":{port}".format(port=start_port + offset)
+            for offset, i in enumerate(range(_num_py_operators + 1))
+        ]
+        control_addresses = [
+            node_to_address[i] + ":{port}".format(
+                port=start_port + len(data_addresses) + offset)
+            for offset, i in enumerate(range(_num_py_operators + 1))
+        ]
     logger.debug(
         "Running the dataflow graph on addresses: {}".format(data_addresses))
 
     def runner(node_id, data_addresses, control_addresses):
         _internal.run(node_id, data_addresses, control_addresses)
 
-    processes = [
-        mp.Process(target=runner, args=(i, data_addresses, control_addresses))
-        for i in range(1, _num_py_operators + 1)
-    ]
+    if not multimachine:
+        processes = [
+            mp.Process(target=runner,
+                       args=(i, data_addresses, control_addresses))
+            for i in range(1, _num_py_operators + 1)
+        ]
+    else:
+        import socket
+        local_address = socket.gethostbyname(socket.gethostname())
+        processes = [
+            mp.Process(target=runner,
+                       args=(i, data_addresses, control_addresses))
+            for i in range(1, _num_py_operators + 1)
+            if local_address == node_to_address[i]
+        ]
 
     # Needed to shut down child processes
     def sigint_handler(sig, frame):

python/erdos/operator.py

@@ -147,12 +147,14 @@ def __init__(self,
                  flow_watermarks: bool = True,
                  log_file_name: str = None,
                  csv_log_file_name: str = None,
-                 profile_file_name: str = None):
+                 profile_file_name: str = None,
+                 address: str = None):
         self._name = name
         self._flow_watermarks = flow_watermarks
         self._log_file_name = log_file_name
         self._csv_log_file_name = csv_log_file_name
         self._profile_file_name = profile_file_name
+        self._address = address
 
     @property
     def name(self):

python/examples/multimachine_pipeline.py

@@ -0,0 +1,70 @@
+"""Every second, sends the message count to 2 receivers.
+One receiver processes messages using a callback,
+one uses the blocking read() call.
+"""
+
+import erdos
+import time
+
+
+class SendOp(erdos.Operator):
+    def __init__(self, write_stream):
+        self.write_stream = write_stream
+
+    @staticmethod
+    def connect():
+        return [erdos.WriteStream()]
+
+    def run(self):
+        count = 0
+        while True:
+            msg = erdos.Message(erdos.Timestamp(coordinates=[count]), count)
+            print("SendOp: sending {msg}".format(msg=msg))
+            self.write_stream.send(msg)
+
+            count += 1
+            time.sleep(1)
+
+
+class CallbackOp(erdos.Operator):
+    def __init__(self, read_stream):
+        print("initializing  op")
+        read_stream.add_callback(CallbackOp.callback)
+
+    @staticmethod
+    def callback(msg):
+        print("CallbackOp: received {msg}".format(msg=msg))
+
+    @staticmethod
+    def connect(read_streams):
+        return []
+
+
+class PullOp(erdos.Operator):
+    def __init__(self, read_stream):
+        self.read_stream = read_stream
+
+    @staticmethod
+    def connect(read_streams):
+        return []
+
+    def run(self):
+        while True:
+            data = self.read_stream.read()
+            print("PullOp: received {data}".format(data=data))
+
+
+def main():
+    """Creates and runs the dataflow graph."""
+    (count_stream, ) = erdos.connect(
+        SendOp, erdos.OperatorConfig(address="172.17.0.9"), [])
+    erdos.connect(CallbackOp, erdos.OperatorConfig(address="172.17.0.10"),
+                  [count_stream])
+    erdos.connect(PullOp, erdos.OperatorConfig(address="172.17.0.11"),
+                  [count_stream])
+
+    erdos.run()
+
+
+if __name__ == "__main__":
+    main()