Merge pull request #76 from knix-microfunctions/release/0.8.6

Release/0.8.6

Author: Istemi Ekin Akkus

DataLayerService/src/main/java/org/microfunctions/data_layer/RiakAccess.java

@@ -578,6 +578,12 @@ public AbstractMap.SimpleEntry<String, ByteBuffer> selectRow (String keyspace, S
 			}
 
 			ByteBuffer value = ByteBuffer.wrap(object.getValue().unsafeGetValue());
+			
+			if (value.hasArray() && value.array().length == 1 && value.array()[0] == 0)
+			{
+				value = ByteBuffer.wrap(new byte[] {});
+			}
+			
 			return new AbstractMap.SimpleEntry<String, ByteBuffer>(key, value);
 		} catch (Exception e) {
 			LOGGER.error("selectRow() failed.  Keyspace: " + keyspace + "  Table: " + table, e);

FunctionWorker/.gitignore

@@ -1,3 +1,5 @@
-data_layer/
-local_queue/
+/python/data_layer/
+/python/local_queue/
+/doc/
 __init__.py
+__pycache__

FunctionWorker/Makefile

@@ -19,20 +19,23 @@ include ../build_env.mk
 UID:=`id -u`
 GID:=`id -g`
 
-default: python_api_doc
+default: doc/MicroFunctionsAPI.html
 
-thrift_data_layer: ../build_env.mk \
+PYTHON_DATALAYER=$(shell find python/data_layer -type f -size +1c) python/data_layer
+$(PYTHON_DATALAYER): ../build_env.mk \
 	../DataLayerService/thrift/DataLayerMessage.thrift \
 	../DataLayerService/thrift/DataLayerService.thrift
 	docker run --user $(UID):$(GID) --rm -v $(CURDIR)/..:/root -w /root $(THRIFT_IMAGE_NAME):$(THRIFT_VERSION) bash -c '\
 		thrift --gen py -out FunctionWorker/python/ DataLayerService/thrift/DataLayerMessage.thrift; \
 		thrift --gen py -out FunctionWorker/python/ DataLayerService/thrift/DataLayerService.thrift'
 
-python_api_doc: thrift_data_layer
+doc/MicroFunctionsAPI.html: $(PYTHON_DATALAYER)
 	docker run --user $(UID):$(GID) --rm -it \
 		--name mfnapi_doc_gen \
 		-v $(CURDIR):/opt/mfnapi \
 		-w /opt/mfnapi \
+		-e http_proxy \
+		-e https_proxy \
 		python:3.6 \
 		/bin/bash -c '\
 		pip3 install pdoc3 --target /opt/mfnapi/tmppip/pdoc3; \

FunctionWorker/python/DataLayerClient.py

@@ -566,6 +566,25 @@ def createTriggerableTable(self, tableName):
                 raise
         return status
 
+    def listKeys(self, start, count, tableName=None):
+        listkeys_response = []
+        table = self.tablename if tableName is None else tableName
+
+        for retry in range(MAX_RETRIES):
+            try:
+                listkeys_response = self.datalayer.selectKeys(self.keyspace, table, start, count, self.locality)
+                if listkeys_response == None or type(listkeys_response) != type([]):
+                    listkeys_response = []
+                break
+            except TTransport.TTransportException as exc:
+                print("[DataLayerClient] Reconnecting because of failed selectKeys: " + str(exc))
+                self.connect()
+            except Exception as exc:
+                print("[DataLayerClient] failed selectKeys: " + str(exc))
+                raise
+
+        return listkeys_response
+
     def shutdown(self):
         try:
             self.transport.close()

FunctionWorker/python/MicroFunctionsAPI.py

@@ -253,7 +253,7 @@ def send_to_running_function_in_session_with_alias(self, alias, message, send_no
         else:
             self._logger.warning("Cannot send a session update message in a workflow with no session functions.")
 
-    def get_session_update_messages(self, count=1):
+    def get_session_update_messages(self, count=1, block=False):
         '''
         Retrieve the list of update messages sent to a session function instance.
         The list contains messages that were sent and delivered since the last time the session function instance has retrieved it.
@@ -263,9 +263,11 @@ def get_session_update_messages(self, count=1):
 
         Args:
             count (int): the number of messages to retrieve; default: 1
+            block (boolean): whether it should block until 'count' number of messages have been received.
 
         Returns:
             List of messages that were sent to the session function instance.
+            If in non-blocking mode, the number of returned messages will be less than or equal to 'count'.
 
         Warns:
             When the calling function is not a session function.
@@ -278,7 +280,7 @@ def get_session_update_messages(self, count=1):
         messages = []
         if self._is_session_function:
             #self._logger.debug("[MicroFunctionsAPI] getting session update messages...")
-            messages = self._session_utils.get_session_update_messages_with_local_queue(count)
+            messages = self._session_utils.get_session_update_messages_with_local_queue(count=count, block=block)
         else:
             self._logger.warning("Cannot get session update messages in a non-session function: " + self._function_state_name)
 

FunctionWorker/python/MicroFunctionsLogWriter.py

@@ -27,10 +27,7 @@ def __init__(self, logger, level=logging.DEBUG):
 
     def write(self, msg):
         msg = msg.rstrip()
-        lines = msg.splitlines()
-        for line in lines:
-            line = line.rstrip()
-            self._logger.log(self._level, line)
+        self._logger.log(self._level, msg)
 
     def flush(self):
         pass

FunctionWorker/python/SessionHelperThread.py

@@ -16,6 +16,7 @@
 
 import json
 import time
+import queue
 from collections import deque
 
 from DataLayerClient import DataLayerClient
@@ -27,7 +28,7 @@
 
 class SessionHelperThread(threading.Thread):
 
-    def __init__(self, helper_params, logger, pubutils, sessutils, queue, datalayer):
+    def __init__(self, helper_params, logger, pubutils, sessutils, queueservice, datalayer):
 
         self._logger = logger
 
@@ -37,7 +38,7 @@ def __init__(self, helper_params, logger, pubutils, sessutils, queue, datalayer)
 
         self._session_utils = sessutils
 
-        self._queue = queue
+        self._queue_service = queueservice
         self._datalayer = datalayer
 
         self._sandboxid = helper_params["sandboxid"]
@@ -49,7 +50,8 @@ def __init__(self, helper_params, logger, pubutils, sessutils, queue, datalayer)
         # need a separate backup data layer client from the publication utils; otherwise, we run into concurrent modification
         # problems from Thrift
         # locality = -1 means that the writes happen to the local data layer first and then asynchronously to the global data layer
-        self._backup_data_layer_client = DataLayerClient(locality=-1, for_mfn=True, sid=self._sandboxid, connect=self._datalayer)
+        # will only initialize if heartbeats are enabled
+        self._backup_data_layer_client = None
 
         # set up heartbeat parameters
         self._heartbeat_enabled = False
@@ -74,12 +76,10 @@ def __init__(self, helper_params, logger, pubutils, sessutils, queue, datalayer)
         # we can send regular heartbeats
         self._local_poll_timeout = py3utils.ensure_long(10000)
 
-        # use a deque to keep the list of messages
-        # updating the list and retrieving the list would be done by two threads
-        # this should be safe without lock because of the global interpreter lock in python
-        self._message_queue = deque()
+        # use a queue to keep the incoming update messages for blocking and/or blocking get_update_messages() requests
+        self._message_queue = queue.Queue()
 
-        self._local_queue_client = LocalQueueClient(connect=self._queue)
+        self._local_queue_client = LocalQueueClient(connect=self._queue_service)
 
         self._special_messages = {}
         self._special_messages["--stop"] = True
@@ -102,10 +102,15 @@ def _init_heartbeat_parameters(self, heartbeat_params):
 
         if self._heartbeat_method == "function":
             if "heartbeat_function" in heartbeat_params:
+                # enable function related heartbeat
                 self._heartbeat_function = heartbeat_params["heartbeat_function"]
                 #self._logger.debug("[SessionHelperThread] New heartbeat function: " + str(self._heartbeat_function))
+                if self._backup_data_layer_client is None:
+                    self._backup_data_layer_client = DataLayerClient(locality=-1, for_mfn=True, sid=self._sandboxid, connect=self._datalayer)
                 if self._local_queue_client_heartbeat is None:
-                    self._local_queue_client_heartbeat = LocalQueueClient(connect=self._queue)
+                    self._local_queue_client_heartbeat = LocalQueueClient(connect=self._queue_service)
+
+                # disable data layer related heartbeat
                 if self._data_layer_client_heartbeat is not None:
                     self._data_layer_client_heartbeat.delete(self._heartbeat_data_layer_key)
                     self._heartbeat_data_layer_key = None
@@ -118,13 +123,22 @@ def _init_heartbeat_parameters(self, heartbeat_params):
             # OR keep a new map for heartbeats of the session functions
             # so that the checker can retrieve the keys and their values (e.g., timestamps)
             # if a session function misses a heartbeat, the checker function reports to policy handler
+
+            # enable data layer related heartbeat
             self._heartbeat_data_layer_key = "heartbeat_" + self._session_id + "_" + self._session_function_id
             if self._data_layer_client_heartbeat is None:
                 self._data_layer_client_heartbeat = DataLayerClient(locality=1, for_mfn=True, sid=self._sandboxid, connect=self._datalayer)
+
+            # disable function related heartbeat
             if self._local_queue_client_heartbeat is not None:
                 self._local_queue_client_heartbeat.shutdown()
                 self._local_queue_client_heartbeat = None
                 self._heartbeat_function = None
+            if self._backup_data_layer_client is not None:
+                self._backup_data_layer_client.shutdown()
+                self._backup_data_layer_client = None
+
+
         else:
             raise MicroFunctionsSessionAPIException("Unsupported heartbeat method for session function.")
 
@@ -223,19 +237,19 @@ def _process_message(self, lqm):
         # _XXX_: we are encoding/decoding the delivered message; should not actually execute this code
         # it is here for not envisioned corner case (i.e., let the user code deal with it)
         if not is_json:
-            self._queue_message(msg)
+            self._store_message(msg)
             self._publication_utils.set_metadata(metadata)
         else:
             # the message is json encoded, but it doesn't guarantee that it is a special message
             if "action" in msg and msg["action"] in self._special_messages:
                 self._handle_special_message(msg)
             else:
-                self._queue_message(msg)
+                self._store_message(msg)
                 self._publication_utils.set_metadata(metadata)
 
 
-    def _queue_message(self, msg):
-        self._message_queue.append(msg)
+    def _store_message(self, msg):
+        self._message_queue.put(msg)
 
     def _handle_special_message(self, msg):
         action = msg["action"]
@@ -247,17 +261,16 @@ def _handle_special_message(self, msg):
         elif action == "--update-heartbeat":
             self._init_heartbeat_parameters(msg["heartbeat_parameters"])
 
-    def get_messages(self, count=1):
+    def get_messages(self, count=1, block=False):
         messages = []
 
-        # this check ensures that we never try to pop() from an empty deque
-        num_messages = len(self._message_queue)
-        if num_messages < count:
-            count = num_messages
-
         for i in range(count):
-            msg = self._message_queue.popleft()
-            messages.append(msg)
+            try:
+                msg = self._message_queue.get(block=block)
+                messages.append(msg)
+                self._message_queue.task_done()
+            except Exception as exc:
+                pass
 
         #self._logger.debug("returning messages: " + str(messages))
         return messages
@@ -317,19 +330,20 @@ def _cleanup(self):
             self._local_queue_client_heartbeat.shutdown()
             self._local_queue_client_heartbeat = None
 
+        if self._backup_data_layer_client is not None:
+            self._backup_data_layer_client.shutdown()
+            self._backup_data_layer_client = None
+
         self._local_queue_client.shutdown()
         self._local_queue_client = None
 
-        self._backup_data_layer_client.shutdown()
-        self._backup_data_layer_client = None
-
     def shutdown(self):
         self._is_running = False
         # remove/unregister the topic here
         # if done in _cleanup(), it may be the case that the parent process exits
         # and we never get to execute that part in this thread
         # need a new lqc, because if using the actual self._local_queue_client,
         # it will corrupt the protocol stack, if it is still waiting on new messages
-        lqc = LocalQueueClient(connect=self._queue)
+        lqc = LocalQueueClient(connect=self._queue_service)
+        lqc.removeTopic(self._local_topic_communication)
         lqc.shutdown()
-

FunctionWorker/python/SessionUtils.py

@@ -476,9 +476,9 @@ def send_to_running_function_in_session_with_alias(self, session_function_alias,
 
         self.send_to_running_function_in_session(rgid, message, send_now)
 
-    def get_session_update_messages_with_local_queue(self, count=1):
+    def get_session_update_messages_with_local_queue(self, count=1, block=False):
         if self._session_function_id is not None:
-            messages = self._helper_thread.get_messages(count)
+            messages = self._helper_thread.get_messages(count=count, block=block)
             return messages
         return None