log backups instead of storing them in the data layer; fixes #85

Author: Istemi Ekin Akkus

FunctionWorker/python/PublicationUtils.py

@@ -453,14 +453,7 @@ def _publish_output(self, key, trigger, lqcpub, timestamp_map=None):
     # and the backup will be overwritten
     # if one or more nexts were generated when publishing
     # at the end of execution, they will have been appended to our list
-    # in memory and we will store the backup once for the entire list
-    def _store_trigger_backups(self, dlc, input_backup_map, current_function_instance_id, store_next_backup_list=False):
-        if self._execution_info_map_name is not None:
-            for input_backup_key in input_backup_map:
-                dlc.putMapEntry(self._execution_info_map_name, input_backup_key, input_backup_map[input_backup_key])
-            if store_next_backup_list:
-                dlc.putMapEntry(self._execution_info_map_name, "next_" + current_function_instance_id, json.dumps(self._next_backup_list))
-
+    # in memory and we will log the backup once for the entire list
     def _log_trigger_backups(self, input_backup_map, current_function_instance_id, store_next_backup_list=False):
         if self._execution_info_map_name is not None:
             for input_backup_key in input_backup_map:
@@ -469,6 +462,7 @@ def _log_trigger_backups(self, input_backup_map, current_function_instance_id, s
                 self._logger.info("[__mfn_backup] [%s] [%s] %s", self._execution_info_map_name, "next_" + current_function_instance_id, json.dumps(self._next_backup_list))
 
     def _send_message_to_recovery_manager(self, key, message_type, topic, func_exec_id, has_error, error_type, lqcpub):
+        # TODO
         return
         message_rec = {}
         message_rec["messageType"] = message_type
@@ -484,8 +478,8 @@ def _send_message_to_recovery_manager(self, key, message_type, topic, func_exec_
         # message via global publisher to pub manager's queue for backups
         self._send_local_queue_message(lqcpub, self._pub_topic_global, key, outputstr)
 
-    # need to store backups of inputs and send message to recovery manager
-    def send_to_function_now(self, key, trigger, lqcpub=None, dlc=None, backup_to_log=False):
+    # need to log backups of inputs and send message to recovery manager
+    def send_to_function_now(self, key, trigger, lqcpub=None):
         trigger["value"] = self.encode_output(trigger["value"])
 
         # get a local queue client
@@ -513,12 +507,7 @@ def send_to_function_now(self, key, trigger, lqcpub=None, dlc=None, backup_to_lo
                 self._next_backup_list.append(next_function_instance_id)
                 any_next = True
 
-            if backup_to_log:
-                self._log_trigger_backups(input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
-            else:
-                if dlc is None:
-                    dlc = self.get_backup_data_layer_client()
-                self._store_trigger_backups(dlc, input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
+            self._log_trigger_backups(input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
 
             for next_func_exec_id in starting_next:
                 next_func_topic = starting_next[next_func_exec_id]
@@ -654,8 +643,7 @@ def publish_output_direct(self, key, value_output, has_error, error_type, timest
                 if self._should_checkpoint:
                     timestamp_map["t_start_backtrigger"] = time.time() * 1000.0
                     # backups for next of successfully completed function execution instances
-                    self._store_trigger_backups(dlc, input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
-                    #self._log_trigger_backups(input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
+                    self._log_trigger_backups(input_backup_map, current_function_instance_id, store_next_backup_list=any_next)
 
                     for next_func_exec_id in starting_next:
                         next_func_topic = starting_next[next_func_exec_id]

FunctionWorker/python/SessionHelperThread.py

@@ -46,13 +46,6 @@ def __init__(self, helper_params, logger, pubutils, sessutils, queueservice, dat
         self._session_function_id = helper_params["session_function_id"]
         self._session_id = helper_params["session_id"]
 
-        # initialize only needed
-        # 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
-        # will only initialize if heartbeats are enabled
-        self._backup_data_layer_client = None
-
         # set up heartbeat parameters
         self._heartbeat_enabled = False
         self._heartbeat_method = None
@@ -105,8 +98,6 @@ def _init_heartbeat_parameters(self, 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_service)
 
@@ -134,9 +125,6 @@ def _init_heartbeat_parameters(self, heartbeat_params):
                 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:
@@ -310,7 +298,7 @@ def _send_heartbeat_to_function(self, hb_message):
         trigger_hb = {}
         trigger_hb["next"] = self._heartbeat_function
         trigger_hb["value"] = hb_message
-        self._publication_utils.send_to_function_now("-1l", trigger_hb, self._local_queue_client_heartbeat, self._backup_data_layer_client)
+        self._publication_utils.send_to_function_now("-1l", trigger_hb, self._local_queue_client_heartbeat)
 
     def _send_heartbeat_to_data_layer(self, hb_message):
         self._data_layer_client_heartbeat.put(self._heartbeat_data_layer_key, json.dumps(hb_message))
@@ -328,10 +316,6 @@ 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
-
         # remove/unregister the topic
         self._local_queue_client.removeTopic(self._local_topic_communication)