Use harness to manage kamel termination (#56)

Author: tardieu

rayvens/core/harness.py

@@ -0,0 +1,30 @@
+#
+# Copyright IBM Corporation 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import os
+import psutil
+import signal
+import subprocess
+import sys
+
+# start child process
+process = subprocess.Popen(sys.argv[1:], start_new_session=True)
+
+# wait for parent process
+psutil.wait_procs([psutil.Process().parent()])
+
+# kill child process
+os.killpg(os.getpgid(process.pid), signal.SIGKILL)

rayvens/core/impl.py

@@ -14,7 +14,6 @@
 # limitations under the License.
 #
 
-import atexit
 import os
 import ray
 import signal
@@ -26,20 +25,12 @@
 import random
 from rayvens.core.validation import Validation
 import rayvens.core.catalog as catalog
-
-integrations = []
-
-
-def killall():
-    global integrations
-    for integration in integrations:
-        integration.cancel()
+import sys
 
 
 # instantiate camel actor manager and setup exit hook
 def start(prefix, mode):
     camel = Camel(mode)
-    atexit.register(camel.killall)
     return camel
 
 
@@ -92,10 +83,6 @@ def await_start(self, integration_name):
     def await_start_all(self, stream):
         return True
 
-    def killall(self):
-        for stream in self.streams:
-            stream._exec.remote(killall)
-
 
 # the low-level implementation-specific stuff
 
@@ -126,15 +113,14 @@ def __init__(self, name, spec):
         with open(filename, 'w') as f:
             yaml.dump(spec, f)
         self.port = random_port()
+        harness = os.path.join(os.path.dirname(__file__), 'harness.py')
         queue = os.path.join(os.path.dirname(__file__), 'Queue.java')
         command = [
-            'kamel', 'local', 'run', queue, '--property',
-            f'quarkus.http.port={self.port}', filename
+            sys.executable, harness, 'kamel', 'local', 'run', queue,
+            '--property', f'quarkus.http.port={self.port}', filename
         ]
         process = subprocess.Popen(command, start_new_session=True)
         self.pid = process.pid
-        global integrations
-        integrations.append(self)
 
     def send_to(self, stream):
         def append():
@@ -158,6 +144,6 @@ def recv_from(self, stream):
 
     def cancel(self):
         try:
-            os.killpg(os.getpgid(self.pid), signal.SIGTERM)
+            os.kill(self.pid, signal.SIGTERM)
         except ProcessLookupError:
             pass