TaskVineExecutor: add new features #2809

tphung3 · 2023-07-10T22:04:11Z

Description

This PR brings new updates to the TaskVineExecutor:

Refactor the constructor of TaskVineExecutor. This helps de-clutter the current executor constructor and separates (almost all) configurations of different components (Executor interface that talks with DataFlowKernel, Manager process that manages tasks, and Factory process that manages workers)
Add a new option to use TaskVine factory as an alternative to Parsl provider. The benefits of the TaskVine factory are carried over to Parsl, with abilities to handle variable number of workers, dynamically size the number of workers according to the number of outstanding/pending tasks, ask for workers with an exact amount of resources (e.g., 8 cores 16 GBs memory worker), etc. Other features can be ported/added to upon request, but the features in this PR should serve well as the core/main use. The default factory will spawn a local process like the local provider, but much faster (reason: factory is run as a different process, while the provider shares CPU with the executor upon startup). The default worker provider is still Parsl provider for now.
Add a new keyword exec_mode to apps and introduce a new execution mode to apps. Now the executor supports regular execution mode, which uses the executor's serialization methods (i.e., exec_parsl_function.py and its friends), and python execution mode, which uses TaskVine manager's serialization methods (in short, works in the same principle as exec_parsl_function.py but uses cloudpickle and PythonTask, which are more native to TaskVine). The serverless feature (function service) was intended to be included in here but there seems to be a bug within TaskVine serverless tasks so it's delayed for now. Apps that use TaskVineExecutor will have a signature of something a long this line:

@python_app
def f(x, call_specs={'cores':1,'memory':4000,'gpus':4,'exec_mode': 'serverless'}):
    ...

@python_app
def g(x, call_specs={'cores':4, 'exec_mode': 'python'}):
    ...

@bash_app
def h(x, call_specs={'exec_mode': 'regular'}):
    ...

Add support for automatic conda environment packaging. This helps users with the ease of transition from local application execution to remote execution given that they use conda package manager to manage software dependencies. This works by packaging a local conda environment (the one that users are using), replicating this environment to workers, and running apps within this environment. From the users' perspective, one line of configuration supplying the conda environment name or path is enough for everything to work out.
A big chunk of changes to many little issues that arise when developing this PR. If needed I'll list them here, but there's really a lot and I want to publish these changes first for discussions.

Fixes # (issue)
N/A

Type of change

New feature (non-breaking change that adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
Documentation update
Code maintentance/cleanup

benclifford · 2023-07-11T09:37:27Z

mypy doesn't pass at the moment (commit a922c7a) with mypy:

parsl/executors/taskvine/executor.py:50: error: Skipping analyzing "autopep8": module is installed, but missing library stubs or py.typed marker [import]

benclifford · 2023-07-11T09:38:37Z

requirements.txt

@@ -12,3 +12,4 @@ requests
 paramiko
 psutil>=5.5.1
 setproctitle
+autopep8


what is this for?

I found where it was used - sorry.

Maybe consider adding taskvine-specific dependencies to setup.py, in extras_require with a taskvine target: so that users would install like this:

pip install 'parsl[taskvine]'

autopep8 is no longer needed and is removed.

benclifford · 2023-07-11T09:40:13Z

parsl/app/app.py

@@ -58,8 +58,8 @@ def __init__(self, func, data_flow_kernel=None, executors='all', cache=False, ig
            self.kwargs['stderr'] = params['stderr'].default
        if 'walltime' in params:
            self.kwargs['walltime'] = params['walltime'].default
-        if 'parsl_resource_specification' in params:
-            self.kwargs['parsl_resource_specification'] = params['parsl_resource_specification'].default
+        if 'parsl_call_specs' in params:


don't change core parsl! this is a breaking change for WQ users, not just the taskvine in-development work, and would need some more serious justificiation

Change reverted. It's just that the variable name can be better so more information can flow from top down.

benclifford · 2023-07-11T09:42:54Z

parsl/executors/high_throughput/executor.py

+            logger.error("Ignoring the call specification. "
+                         "Parsl call specification is not supported in HighThroughput Executor. "
+                         "Please check WorkQueueExecutor or TaskVineExecutor if call specification is needed.")
+            raise UnsupportedFeatureError('call specification', 'HighThroughput Executor', 'WorkQueue Executor')


this exception is a bit icky: it's meant to direct you to use another executor - but it doesn't redirect you to either WQ or TaskVine in the exception object.

That's not too concerning but it's a bit ugly... I think the right thing here is that the UnsupportedFeatureError shouldn't be pointing people to WQ at all: in a modular, open expanding world, the High Throughput Executor code shouldn't be expected to know about other executors at all...

I agree. UnsupportedFeatureError is now agnostic to target executors if needs be.

benclifford · 2023-07-11T09:44:26Z

parsl/executors/taskvine/executor.py

+
+        # If TaskVine factory is used, disable the Parsl provider
+        if use_factory:
+            provider = LocalProvider(init_blocks=0, max_blocks=0)


have you tried using None here? ideally that would work here but maybe it doesn't...

Using None almost works, until the intepreter exits...

Error in atexit._run_exitfuncs: Traceback (most recent call last): File "/home/jason/work/parsl_taskvine/parsl/parsl/process_loggers.py", line 27, in wrapped r = func(*args, **kwargs) File "/home/jason/work/parsl_taskvine/parsl/parsl/dataflow/dflow.py", line 1205, in cleanup job_ids = executor.provider.resources.keys() AttributeError: 'NoneType' object has no attribute 'resources'

An easy fix is to check for existence of executor.provider and is incoporated in this PR. provider is set to None.

mypy doesn't allow assignment to None so revert back to dummy LocalProvider.

ok - I'll see if I can sort that out in a different PR...

I made PR #2825 to work on allowing this

benclifford · 2023-07-11T09:46:48Z

parsl/executors/taskvine/executor.py

+
+        # need source of function to send tasks as python or serverless tasks
+        if exec_mode == 'python' or exec_mode == 'serverless':
+            func_src = autopep8.fix_code(inspect.getsource(func))


does this formatting make the code more executable? or it is prettifying for human consumption? (i can't tell from the autopep8 docs if this is going to change how the code runs?)

In cases where apps are functions indented inside a class or another function, like this:

def f(): @python_app def g(): ...

then inspecting function g inspect.getsource(g) results in a string that's 1 level indented, like this:

@python_app def g(): ...

This prevents attempts to exec the source code into a temporary namespace as the indentation level is one level to the right. autopep8.fix_code corrects this and aligns the function with its body properly.

Another way to address this case is to write a custom method that aligns function source code (detecting type, either tabs or whitespaces, and number of indentations should be easy) but this is a bit hacky. Let me know what you think.

mostly I think this PR should avoid anything to do with changing serialization away from parsl's defaults, and focus on solid improvements to the TaskVineExecutor.

Then, work on serialization in a different branch - it might even be that some techniques like are more generally useful across all parsl executors and the code should go into the parsl serialization system, rather than the taskvine executor.

benclifford · 2023-07-11T09:49:53Z

parsl/executors/taskvine/executor.py

-        if self.source:
-            function_info = {"source code": inspect.getsource(parsl_fn),
+        if app_type == 'python':
+            function_info = {"source code": autopep8.fix_code(inspect.getsource(parsl_fn)),


I think in general, you shouldn't expect source code of a function to exist: sometimes callables passed to executors come from places other than raw source (for example, wrappers/closures in monitoring and file staging, or in some cases, interesting user generated callables)

That's also what a bash_app is: just some wrapped python code...

So I think this is generally the wrong thing to be doing by default.

I agree that function source code is not a given. This is a bit concerning as the serverless task model in TaskVine assumes otherwise.

The default behavior now is to transfer source code only if a function object has source code, otherwise we fall back to pack_apply_message.

does that mean bash_apps don't work with serverless mode?

With the current state of serverless task, yes. Serverless tasks use source code to compute a function hash, which is then used to put each function into its own staging directory.

ok, that feels like a more fundamental architectural problem which is going to to cause trouble long term with people actually using this for real stuff...

It seems that many parts of serverless tasks assume the existence of source code, not just computing a unique directory.

Yes tough luck... I'll see if there's any quick fix to this later.

ok, so for this PR I recommend getting rid of the changes in core (choosing which decorator) that were made to support this source code stuff, because from the above it looks to me like they're to support a use-case (serverless) which isn't going to work properly...

then let's have a proper discussion about serverless, maybe involving other people - perhaps by email.

(alternatively, if you prefer, rather than removing stuff from this PR, you can open a few new PRs, each with individual changes from this PR that are less controversial to review - that would also provide a path to get much of this work merged, leaving the awkward stuff to think about later)

benclifford · 2023-07-11T11:14:27Z

on my laptop, pytest parsl/tests/ --config parsl/tests/configs/taskvine_ex.py against this taskvine version (from git)

../cctools/I/bin/vine_worker version 8.0.0 DEVELOPMENT (released 2023-06-19 12:12:49 +0000)
	Built by benc@parsl-dev-3-11-1364 on 2023-06-19 12:12:49 +0000
	System: Linux parsl-dev-3-11-1364 5.19.0-43-generic #44~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Mon May 22 13:39:36 UTC 2 x86_64 GNU/Linux
	Configuration: --prefix=/home/benc/parsl/src/cctools/I

fails with this error:

_______________________________________ test_check_importlib_file_function ________________________________________

    def test_check_importlib_file_function():
        helper_path = pathlib.Path(__file__).parent / "callables_helper.py"
        spec = importlib.util.spec_from_file_location("dynamically_loaded_module", helper_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        some_aux_func = module.some_aux_func
>       app(some_aux_func).result()

...

  File "/home/benc/parsl/src/parsl/parsl/dataflow/dflow.py", line 719, in launch_task
    exec_fu = executor.submit(executable, task_record['call_specs'], *args, **kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/benc/parsl/src/parsl/parsl/executors/taskvine/executor.py", line 392, in submit
    self._serialize_function(call_specs['app_type'], function_file, func, args, kwargs)
  File "/home/benc/parsl/src/parsl/parsl/executors/taskvine/executor.py", line 486, in _serialize_function
    pickle.dump(function_info, f_out)
_pickle.PicklingError: Can't pickle <function some_aux_func at 0x7f7d53d26160>: import of module 'dynamically_loaded_module' failed

I think that's coming from the code in _serialize_function that tries to avoid using parsl's regular serialization mechanisms and do something different... that then breaks.

Maybe we can dig in a bit what isn't working with the parsl default serialization in this implementation.

If I comment out all the source based stuff in taskvine/executor.py and use pack_apply_message's parsl serialization, many pytest tests pass ok: these one's do not (some of them even hang with no error report):

test_callables.py: test_check_importlib_file_function and test_check_importlib_file_function_partial

tests/test_staging/test_staging_ftp.py: test_staging_ftp

benclifford · 2023-07-11T11:16:01Z

@tphung3 I reviewed enough to identify two major concerns:

breaking the publicly facing parsl_resource_specification API, and
stuff that concerns me about function serialization, which I have written about above and as per-line comments.

I haven't reviewed the other pieces yet and probably won't until we deal with the above two points.

benclifford · 2023-07-11T12:04:22Z

to fix that autopep8 mypy error, I think it will gets fixed if you add autopep8 into mypy.ini near the end, like these packages:

[mypy-oauth_ssh.*]
ignore_missing_imports = True

[mypy-flux.*]
ignore_missing_imports = True

[mypy-setproctitle.*]
ignore_missing_imports = True

…cutor

benclifford · 2023-07-19T12:20:00Z

ok for that race condition:

get all the process starts before the thread start, like this:

@@ -266,13 +266,15 @@ class TaskVineExecutor(BlockProviderExecutor, putils.RepresentationMixin):
 
         # Begin work
         self.submit_process.start()
-        self.collector_thread.start()
 
         # Run worker scaler either with Parsl provider or TaskVine factory
         if self.use_factory:
             self.factory_process.start()
         else:
             self.initialize_scaling()
+
+        self.collector_thread.start()

This doesn't fix the race condition, but reduces the likelihood of the hang - it doesn't happen at all on my laptop after an hour of looping, for example.

The broader story:

Python programs cannot really use 1) multiprocessing via the fork launch method and 2) threading together in one problem: multiprocess forks massively unsafe. See this python issue python/cpython#84559 and a plan to switch Python to use spawn by default.

In general, Parsl should move away from fork based multiprocessing. However there are some subtleties there which make it non-trivial, which is why other parts of parsl still use it - and why I don't think you could use spawn here either.

Here's an issue tracking this #2343

So I can't, in the presence of our other bad examples, insist on not using multiprocessing fork here... and that's why I suggest the swap of thread and multiprocessing above.

The race condition that's happens around here, when I've seen it before, is that the thread does a lot of stuff as it starts which takes locks... and then those locked locks are immedately forked into the new process, which never unlocks them.

Getting that process forking done before any thready-stuff reduces the chances of that.

benclifford · 2023-07-19T12:21:30Z

parsl/app/bash.py

@@ -150,6 +150,7 @@ def __call__(self, *args, **kwargs):
                             executors=self.executors,
                             cache=self.cache,
                             ignore_for_cache=self.ignore_for_cache,
-                             app_kwargs=invocation_kwargs)
+                             app_kwargs=invocation_kwargs,
+                             app_type='bash')


This app_type annotation is not used any more, right? (and I hope to persuade you it will never be needed...)

That's right. I'll revert that change some time today.

benclifford · 2023-07-19T12:39:40Z

parsl/dataflow/dflow.py

-                        msg = executor.create_monitoring_info(new_status)
-                        logger.debug("Sending message {} to hub from DFK".format(msg))
-                        self.monitoring.send(MessageType.BLOCK_INFO, msg)
+                    if executor.provider:


i'm going to move this into a separate PR because it isn't directly task vine related. Parsl shutdown has been pretty fragile, so I'd like to keep this as an explicit commit in our history. You don't need to do anything to make that happen.

tphung3 · 2023-07-20T19:12:12Z

If all is good I'd like to add some documentations for TaskVine. Is there a command that lints documentation? Also the PR fails at that part.

benclifford · 2023-07-21T06:13:21Z

@tphung3 usually I build the documentation on my dev machine. adding the docs extras requirement will get the dependencies (eg something like pip install -e .[docs]`) and then:

cd docs/
make html

will build the html docs and put them in docs/_build/html

and this:

make SPHINXOPTS=-W html

will run the more aggressive build that CI does, that fails harder so you can see what it got angry about.

benclifford · 2023-07-21T07:40:08Z

parsl/executors/high_throughput/executor.py

-                         "Parsl resource specification is not supported in HighThroughput Executor. "
-                         "Please check WorkQueueExecutor if resource specification is needed.")
-            raise UnsupportedFeatureError('resource specification', 'HighThroughput Executor', 'WorkQueue Executor')
+        if 'app_type' in resource_specification and len(resource_specification) > 1:


this can go back to the original test now, right?

Change reverted.

benclifford · 2023-07-21T07:45:10Z

parsl/executors/taskvine/executor.py

+
+        # guess the host name if the project name is not given
+        if not self.manager_config.project_name:
+            self.manager_config.address = socket.gethostname()


there is a module parsl.addresses which provides various methods for working out hostnames, intended for use with the address parameter of the high throughput executor. This is because we haven't found one method that works everywhere. That experience makes me think that eventually that might be needed here too (although that doesn't need to block this PR being merged - it's a note for the future)

benclifford · 2023-07-21T08:08:46Z

parsl/executors/taskvine/executor.py

-                executor_task_id = vine_id_to_executor_task_id[str(t.id)]
-                logger.debug("Completed TaskVine task {}, executor task {}".format(t.id, executor_task_id))
-                result_file = result_file_of_task_id.pop(executor_task_id)
+                logger.debug('Found a task!')


don't use ! surprise in log messages - if it's a surprise it should be a WARNING

benclifford · 2023-07-21T08:11:07Z

parsl/executors/threads.py

@@ -52,11 +52,10 @@ def submit(self, func, resource_specification, *args, **kwargs):
        here `Python docs: <https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor>`_

        """
-        if resource_specification:
+        if 'app_type' in resource_specification and len(resource_specification) > 1:


same comment about htex - this can go back to the original if condition?

benclifford · 2023-07-21T08:11:24Z

parsl/executors/workqueue/executor.py


            required_resource_types = set(['cores', 'memory', 'disk'])
-            acceptable_resource_types = set(['cores', 'memory', 'disk', 'gpus', 'priority', 'running_time_min'])
+            acceptable_fields = set(['cores', 'memory', 'disk', 'gpus', 'priority', 'running_time_min', 'app_type'])


spurious app_type

benclifford · 2023-07-21T08:12:06Z

parsl/executors/taskvine/factory_config.py

+from typing import Optional
+
+
+# Configuration of a TaskVine factory


Maybe this should be the first sentence of the docstring so that it appears as the short description of the structure in docs? (and same for TaskVineManagerConfig)

Yes, corrected.

benclifford · 2023-07-21T08:21:52Z

I realised how many different code paths there are through this code - it might be interesting (eg in a later PR) to put in some tests for various things that run just a single task rather than the full test suite, but for a wide selection of different configuration options.

tphung3 · 2023-07-21T17:24:12Z

Looks like CI is stuck at generating taskvine stubs in /docs/stubs. Is this automate-able?

benclifford · 2023-07-21T17:30:15Z

Looking at that error briefly, I think it's because it can't import ndcctools - because that isn't on the python path by default (and why there's an explicit path set for the taskvine and work queue pytests).

You might try adding an explicit python path onto the doc build in .GitHub/workflows/ci.yaml to tell it where to find task vine code... I haven't tried if that solves the problem because I'm away from my development environment

This PR is a follow-up to PR #2809 with the following changes: - Add support for serverless execution mode for tasks. Users can now declare serverless tasks like this: ``` @python_app def f(x, parsl_resource_specification={'exec_mode': 'serverless'}): ... ``` Under the hood, a library task is deployed to each worker and forks itself to execute an actual task/function call. - Add support for full conda environment replication on worker nodes for both serverless and regular tasks. Given a conda environment, TaskVine will distribute and cache it efficiently, so a shared filesystem is not required. The benefit mainly lies in the assumption that the conda environment can be huge in size and thousands of tasks asking the shared filesystem for the same environment directory at the same time is not very performant. TaskVine on the other hand keeps track of where files are at and has an implicit throttle mechanism to data distribution so things don't go haywire. - Add support for no conda environment replication, provided that environment is properly setup on worker nodes. Turning on `shared_fs` flag and add some setup code to `init_command` would turn off almost all TaskVine data-aware magic and the workflow run will mostly use the shared filesystem instead (e.g., `init_command="conda run -p path/to/conda"`). This serves as the baseline behavior for TaskVineExecutor. - Refactor `executor.py` into {`executor.py`, `manager.py`, and `factory.py`} as they are different components. - Change API: `use_factory` to `worker_launch_method`. - Some variable renaming and additional logging.

tphung3 added 16 commits July 10, 2023 13:18

add some docs

fb380c9

remove space

7409a64

add manager config setup code

6e3c60b

finished manager config

a6aa7e1

add more code

c7a4c47

manager config works on test

b8b1f40

restructure code

c8e9bc8

integrate factory to vine executor

215b22e

finish draft of pythontask

55229d2

dev stops as there seems to be a bug with serverless task

3fa4b94

add more docs

b0d2097

last commit before creating a divergent branch for serverless

36b5ad7

clean up code

b66040d

done cleaning code

5818e59

fix to pass flake8 and mypy

25baf64

final commit before push

a922c7a

tphung3 mentioned this pull request Jul 10, 2023

Vine: Complete Parsl Integration cooperative-computing-lab/cctools#3283

Closed

3 tasks

benclifford reviewed Jul 11, 2023

View reviewed changes

tphung3 added 3 commits July 11, 2023 15:33

revert to res spec and let unsupported faeture agnostic to target exe…

f8cb847

…cutor

provider set to none when factory is used

88728cb

remove autopep8

994dae3

benclifford reviewed Jul 19, 2023

View reviewed changes

benclifford mentioned this pull request Jul 19, 2023

Make cleanup work with executors without a provider #2830

Merged

tphung3 added 3 commits July 20, 2023 12:44

Merge branch 'master' into vine_add_new_features

3ef12bf

remove app type and start thread last

8475f39

pass mypy check

800b262

benclifford reviewed Jul 21, 2023

View reviewed changes

tphung3 added 5 commits July 21, 2023 09:37

revert app_type

4874cd5

use parsl address resolution

131468c

move comments to docstring

9c9dce8

fix table and add reference

980aa00

fix flake

9ddf18f

tphung3 and others added 2 commits July 21, 2023 13:53

add python path to make docs

968e55f

Merge branch 'master' into vine_add_new_features

a0c09dd

benclifford approved these changes Jul 21, 2023

View reviewed changes

benclifford merged commit 41357c6 into Parsl:master Jul 21, 2023
4 checks passed

tphung3 mentioned this pull request Aug 1, 2023

TaskVineExecutor: add serverless execution mode for tasks #2849

Merged

benclifford mentioned this pull request Oct 13, 2023

Readthedocs build has been broken for several months #2903

Closed

		from typing import Optional


		# Configuration of a TaskVine factory

TaskVineExecutor: add new features #2809

TaskVineExecutor: add new features #2809

Conversation

tphung3 commented Jul 10, 2023

Description

Type of change

benclifford commented Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benclifford Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tphung3 Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tphung3 Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benclifford Jul 11, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benclifford commented Jul 11, 2023

benclifford commented Jul 11, 2023 • edited Loading

benclifford commented Jul 11, 2023

benclifford commented Jul 19, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tphung3 commented Jul 20, 2023

benclifford commented Jul 21, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benclifford commented Jul 21, 2023

tphung3 commented Jul 21, 2023

benclifford commented Jul 21, 2023

benclifford commented Jul 11, 2023 •

edited

Loading

benclifford Jul 11, 2023 •

edited

Loading

tphung3 Jul 11, 2023 •

edited

Loading

tphung3 Jul 11, 2023 •

edited

Loading

benclifford Jul 11, 2023 •

edited

Loading

benclifford commented Jul 11, 2023 •

edited

Loading