fixes to update head

Author: tphung3

parsl/executors/taskvine/executor.py

@@ -323,6 +323,7 @@ def _path_in_task(self, executor_task_id, *path_components):
         return os.path.join(self._function_data_dir.name, task_dir, *path_components)
 
     def submit(self, func, resource_specification, *args, **kwargs):
+        import cloudpickle
         """Processes the Parsl app by its arguments and submits the function
         information to the task queue, to be executed using the TaskVine
         system. The args and kwargs are processed for input and output files to
@@ -459,8 +460,15 @@ def submit(self, func, resource_specification, *args, **kwargs):
                     function_context_args = resource_specification.get('function_context_args', [])
                     function_context_kwargs = resource_specification.get('function_context_kwargs', {})
                     function_context_file = os.path.join(self._function_data_dir.name, func.__name__, 'function_context')
-                    self._serialize_object_to_file(function_context_file, [function_context, function_context_args, function_context_kwargs])
+
+                    # DEBUG
+                    with open('/tmp/tmp.context.executor.function.module.file', 'w') as f:
+                        f.write(str(function_context.__module__))
+
+                    self._cloudpickle_serialize_object_to_file(function_context_file, [function_context, function_context_args, function_context_kwargs])
                     self._map_func_names_to_func_details[func.__name__].update({'function_context_file': function_context_file})
+                    # DEBUG
+                    shutil.copyfile(function_context_file, '/tmp/tmp.context.file')
                 else:
                     function_context_file = self._map_func_names_to_func_details[func.__name__]['function_context_file']
 
@@ -472,6 +480,21 @@ def submit(self, func, resource_specification, *args, **kwargs):
             self._serialize_object_to_file(function_file, func)
             if exec_mode == 'serverless':
                 self._map_func_names_to_func_details[func.__name__]['is_serialized'] = True
+
+        # DEBUG
+        if exec_mode == 'serverless':
+            import copy
+            kwargs = copy.deepcopy(kwargs)
+            logger.info(f'ThanhDBG before trimming kwargs: {kwargs}')
+            # pop function context stuff, that should not propagate with the invocation function
+            if 'function_context' in kwargs['parsl_resource_specification']:
+                del kwargs['parsl_resource_specification']['function_context']
+            if 'function_context_args' in kwargs['parsl_resource_specification']:
+                del kwargs['parsl_resource_specification']['function_context_args']
+            if 'function_context_kwargs' in kwargs['parsl_resource_specification']:
+                del kwargs['parsl_resource_specification']['function_context_kwargs']
+            logger.info(f'ThanhDBG after trimming kwargs: {kwargs}')
+
         args_dict = {'args': args, 'kwargs': kwargs}
         self._serialize_object_to_file(argument_file, args_dict)
 
@@ -556,6 +579,12 @@ def _serialize_object_to_file(self, path, obj):
             while written < len(serialized_obj):
                 written += f_out.write(serialized_obj[written:])
 
+    def _cloudpickle_serialize_object_to_file(self, path, obj):
+        """Takes any object and serializes it to the file path."""
+        import cloudpickle
+        with open(path, 'wb') as f:
+            cloudpickle.dump(obj, f)
+
     def _construct_map_file(self, map_file, input_files, output_files):
         """ Map local filepath of parsl files to the filenames at the execution worker.
         If using a shared filesystem, the filepath is mapped to its absolute filename.

parsl/executors/taskvine/manager.py

@@ -57,6 +57,9 @@ def _set_manager_attributes(m, config):
         for k, v in config.tune_parameters.items():
             m.tune(k, v)
 
+    # DEBUG
+    m.tune("watch-library-logfiles", 1)
+
 
 def _prepare_environment_serverless(manager_config, env_cache_dir, poncho_create_script):
     # Return path to a packaged poncho environment
@@ -276,8 +279,22 @@ def _taskvine_submit_wait(ready_task_queue=None,
                     # Deserialize the function context to add it to the library if available
                     # This cost is paid only once per function/app.
                     function_context_list = None
+
                     if task.function_context_file:
-                        function_context_list = _deserialize_object_from_file(task.function_context_file)
+                        #function_context_list = _deserialize_object_from_file(task.function_context_file)
+                        import cloudpickle
+                        with open(task.function_context_file, 'rb') as f:
+                            function_context_list = cloudpickle.load(f)
+
+                    # DEBUG
+                    with open('/tmp/tmp.context.manager.print.context.file.file', 'w') as f:
+                        f.write(str(task.function_context_file))
+                    with open('/tmp/tmp.context.manager.file', 'w') as f:
+                        f.write(str(function_context_list))
+                    logger.info(f'ThanhDBG {function_context_list}')
+                    with open('/tmp/tmp.context.manager.function.module.file', 'w') as f:
+                        f.write(str(function_context_list[0].__module__))
+                    #logger.info('ThanhDBG' + function_context_list[0].__module__)
 
                     # Don't automatically add environment so manager can declare and cache the vine file associated with the environment file
                     add_env = False
@@ -290,6 +307,7 @@ def _taskvine_submit_wait(ready_task_queue=None,
                                                                      hoisting_modules=[parsl.serialize, run_parsl_function],
                                                                      exec_mode='direct',
                                                                      library_context_info=function_context_list)
+                    
 
                     # Configure the library if provided
                     if manager_config.library_config:
@@ -324,6 +342,11 @@ def _taskvine_submit_wait(ready_task_queue=None,
                     args = all_args['args']
                     kwargs = all_args['kwargs']
 
+                    # DEBUG
+                    with open('/tmp/manager.check.args.kwargs.function.call.file', 'w') as f:
+                        f.write(str(args))
+                        f.write(str(kwargs))
+
                     t = FunctionCall(libs_installed[task.func_name], task.func_name, *args, **kwargs)
                 except Exception as e:
                     logger.error("Unable to create executor task (mode:serverless): {}".format(e))