Instead of a multiprocessing queue use sockets via asyncore

For a local process based executor usage currently to ensure
that task emitted notifications are proxied we use the multi
processing library and use its queue concept. This sadly creates
a proxy process that gets associated, and this proxy process
handles the queue and messages sent to and from it. Instead of
doing this we can instead just create a temporary local socket
using a random socket and have tasks (which are running in
different processes) use that to communicate back any emitted
notifications instead (and we can use the asyncore module to handle
the emitted notifications since it handles the lower level socket
reading, polling and dispatching).

To ensure that the socket created is somewhat secure we use a
similar process as the multi-processing library uses where we
sign all messages with a hmac that uses a one time key that only
the main process and the child process know about (and reject
any messages that do not validate using this key).

Change-Id: Iff9180054bf14495e5667af00ae2fafbdbc23791

2 files changed, 102 insertions, 2 deletions

diff --git a/taskflow/tests/unit/action_engine/test_creation.py b/taskflow/tests/unit/action_engine/test_creation.py
index a239099..1568dfe 100644
--- a/taskflow/tests/unit/action_engine/test_creation.py
+++ b/taskflow/tests/unit/action_engine/test_creation.py
@@ -19,6 +19,7 @@ import testtools
 
 from taskflow.engines.action_engine import engine
 from taskflow.engines.action_engine import executor
+from taskflow.engines.action_engine import process_executor
 from taskflow.patterns import linear_flow as lf
 from taskflow.persistence import backends
 from taskflow import test
@@ -47,7 +48,7 @@ class ParallelCreationTest(test.TestCase):
         for s in ['process', 'processes']:
             eng = self._create_engine(executor=s)
             self.assertIsInstance(eng._task_executor,
-                                  executor.ParallelProcessTaskExecutor)
+                                  process_executor.ParallelProcessTaskExecutor)
 
     def test_thread_executor_creation(self):
         with futurist.ThreadPoolExecutor(1) as e:
@@ -59,7 +60,7 @@ class ParallelCreationTest(test.TestCase):
         with futurist.ProcessPoolExecutor(1) as e:
             eng = self._create_engine(executor=e)
             self.assertIsInstance(eng._task_executor,
-                                  executor.ParallelProcessTaskExecutor)
+                                  process_executor.ParallelProcessTaskExecutor)
 
     @testtools.skipIf(not eu.EVENTLET_AVAILABLE, 'eventlet is not available')
     def test_green_executor_creation(self):
diff --git a/taskflow/tests/unit/action_engine/test_process_executor.py b/taskflow/tests/unit/action_engine/test_process_executor.py
new file mode 100644
index 0000000..2bca18f
--- /dev/null
+++ b/taskflow/tests/unit/action_engine/test_process_executor.py
@@ -0,0 +1,99 @@
+# -*- coding: utf-8 -*-
+
+#    Copyright (C) 2015 Yahoo! Inc. All Rights Reserved.
+#
+#    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 asyncore
+import errno
+import socket
+import threading
+
+from taskflow.engines.action_engine import process_executor as pu
+from taskflow import task
+from taskflow import test
+from taskflow.test import mock
+from taskflow.tests import utils as test_utils
+
+
+class ProcessExecutorHelpersTest(test.TestCase):
+    def test_reader(self):
+        capture_buf = []
+
+        def do_capture(identity, message_capture_func):
+            capture_buf.append(message_capture_func())
+
+        r = pu.Reader(b"secret", do_capture)
+        for data in pu._encode_message(b"secret", ['hi'], b'me'):
+            self.assertEqual(len(data), r.bytes_needed)
+            r.feed(data)
+
+        self.assertEqual(1, len(capture_buf))
+        self.assertEqual(['hi'], capture_buf[0])
+
+    def test_bad_hmac_reader(self):
+        r = pu.Reader(b"secret-2", lambda ident, capture_func: capture_func())
+        in_data = b"".join(pu._encode_message(b"secret", ['hi'], b'me'))
+        self.assertRaises(pu.BadHmacValueError, r.feed, in_data)
+
+    @mock.patch("socket.socket")
+    def test_no_connect_channel(self, mock_socket_factory):
+        mock_sock = mock.MagicMock()
+        mock_socket_factory.return_value = mock_sock
+        mock_sock.connect.side_effect = socket.error(errno.ECONNREFUSED,
+                                                     'broken')
+        c = pu.Channel(2222, b"me", b"secret")
+        self.assertRaises(socket.error, c.send, "hi")
+        self.assertTrue(c.dead)
+        self.assertTrue(mock_sock.close.called)
+
+    def test_send_and_dispatch(self):
+        details_capture = []
+
+        t = test_utils.DummyTask("rcver")
+        t.notifier.register(
+            task.EVENT_UPDATE_PROGRESS,
+            lambda _event_type, details: details_capture.append(details))
+
+        d = pu.Dispatcher({}, b'secret', b'server-josh')
+        d.setup()
+        d.targets[b'child-josh'] = t
+
+        s = threading.Thread(target=asyncore.loop, kwargs={'map': d.map})
+        s.start()
+        self.addCleanup(s.join)
+
+        c = pu.Channel(d.port, b'child-josh', b'secret')
+        self.addCleanup(c.close)
+
+        send_what = [
+            {'progress': 0.1},
+            {'progress': 0.2},
+            {'progress': 0.3},
+            {'progress': 0.4},
+            {'progress': 0.5},
+            {'progress': 0.6},
+            {'progress': 0.7},
+            {'progress': 0.8},
+            {'progress': 0.9},
+        ]
+        e_s = pu.EventSender(c)
+        for details in send_what:
+            e_s(task.EVENT_UPDATE_PROGRESS, details)
+
+        # This forces the thread to shutdown (since the asyncore loop
+        # will exit when no more sockets exist to process...)
+        d.close()
+
+        self.assertEqual(len(send_what), len(details_capture))
+        self.assertEqual(send_what, details_capture)