Example with wait callback?

[jlaura]

The state language docs define the Wait State spec. AWS step functions seem to extend this spec with the ability to register a callback. Is there an example where a globus flow has a wait and registers a callback?

Use Case:
The flow submits a job to a slurm scheduler for either immediate or future execution, pending resource availability. The Wait State needs to wait for an indeterminate amount of time before executing the next step in the flow (likely a choice step to parse job success)

Work Around:
A times wait state that calls a compute endpoint that polls for the current job status, sends that status to a parser, and then either returns to the wait state or continues to the next step in the flow. This is not ideal as a large percentage of the logging output becomes wait-choice-repeat logging.

[sirosen]

Hi @jlaura, sorry it took so long for us to reply! Last week was the annual Globus-held conference, GlobusWorld, so our team was really tied up with that.

Something we recently took care of is fixing the Globus Flows documentation to not say that we implement the AWS State Language. It hass said that for a long time, but it's always been inaccurate -- we don't support all of the features of the State Language and Wait with a callback is one of them.

---

How are you planning to submit the job to Slurm from within your flow? Depending on the method of submission, there may be other ways to address this solution that interface more cleanly with Flows and don't require looping in the flow itself no clogging up the event log.

The use of Flows to poll on your Slurm job assumes that it's not more appropriate for you to submit your jobs via Globus Compute, which opens up more natural workflows for blocking and waiting on job completion.
Have you considered using the Slurm backend for Globus Compute, which allows a Compute Endpoint to natively "talk Slurm"? I'm not an expert on this myself, but if you reach out to us at support@globus.org, we can get someone more experienced with advanced Globus Compute capabilities to help out.

The workaround you mention is pretty similar to what I would have recommended for an initial solution. You can make the logging less noisy by making your compute function internally poll or wait to see if your Slurm job completes -- a compute function can take as much as 30 days to complete without disrupting a flow, but I would recommend a smaller period (e.g., one week).
We can try to help you brainstorm other solutions, but my first two thoughts are polling within Flows -- but with more of the work done in the polling function -- or plugging Compute into your scheduler a little bit more directly.

[jlaura]

@sirosen I fear I am competing for longer delay time! Thank you for the answer, which I very much appreciate. I have been doing some more exploratory work here and have a follow-up question.

I have a multi-user compute endpoint that is working. I see in the example configurations that the walltime / hold blocks that teams are using range from 5 to 60 minutes. That makes total sense when the compute endpoint is being used as a serverless endpoint for Python function invocation (e.g., a lambda).

Given your reply above about trying Globus Compute, how would a long-running Slurm job work with a provider where the walltime is less than the total execution time? I am trying to understand the walltime specification inside the configuration, and the 30-day (we have a 3-day QoS, so well within the recommended week) max runtime to not disrupt the flow.

I can 100% take this over to the support channel as well if you prefer. Please let me know. I am with USGS.

[khk-globus]

Hi @jlaura ; I'm on the Compute team. Let's see if among the 5+ of us watching this ticket we can get you on your way!

I have a multi-user compute endpoint that is working. I see in the example configurations that the walltime / hold blocks that teams are using range from 5 to 60 minutes. That makes total sense when the compute endpoint is being used as a serverless endpoint for Python function invocation (e.g., a lambda).

Bear in mind that these are examples. While the structure and salient details for proving they were working configurations for those sites is sound (e.g., provider: type: ..., address: ifname: ...), we intentionally curbed other details to play nice while folks get a feel for what a configuration is and how it changes the endpoint behavior. walltime is one of the latter. We'd rather not have a poor default unintentionally burn a week of node hours. Put differently, use those configurations as a starting point for your task needs. If your job needs 3 days to run, then by all means, increase the walltime to something appropriate for you.

Given your reply above about trying Globus Compute, how would a long-running Slurm job work with a provider where the walltime is less than the total execution time? I am trying to understand the walltime specification inside the configuration, and the 30-day (we have a 3-day QoS, so well within the recommended week) max runtime to not disrupt the flow.

First, some clarity on terminology:

• job - a unit of acquisition from the site scheduler. This might be 1 node for 5m, or 1,000 nodes for 2 days, et cetera. This is distinct from a task (see next bullet). When the endpoint receives tasks from the Compute web-services, it uses the configuration to determine the shape of job to request of the scheduler, including number of nodes and for how long.
• task - the unit of work that Globus Compute uses. This would amount to a single Executor.submit() call.

@ryanchard may correct my precision here, but the walltime is a request to the site scheduler (e.g., Slurm), to put a hard limit on how long the job is allowed to run. If a job will need 60+ minutes to run, but the configuration has walltime: 00:05:00, then the job will be cut off 5m after starting — in that case, "I hope you implemented [some variant of] checkpointing," else those 5m would be lost. More likely you'd want to increase the walltime to at least higher than the total expected job time.

While the job is allocated (reminder: roughly N nodes for M time), the endpoint will dole out tasks it receives to the allocated nodes. If individual tasks take less than the 5m example from above, then nominally "there's the checkpointing." But any tasks that are currently in progress on any of the allocated nodes in that job will be lost when the 5m cutoff happens.

how would a long-running Slurm job work with a provider where the walltime is less than the total execution time?

If the thrust of this question is "What happens to in-progress tasks?" then the last sentence above is the answer: they will be lost. Compute does not currently implement task checkpointing or restarting in this scenario.

[jlaura]

@khk-globus Thanks for the info. I work managing the HPC resources, and we are looking at the multi-user endpoint. I see that all of the example configurations are less than one hour of wall time. In your experience would it be advisable to increase walltime to something larger than 1 hour (our current max outside a specific QoS is 36 hours) or to use the callback approach described above where a compute function inside a flow polls the scheduler for job status? It seems that the former approach integrates better with the flow paradigm and results in less verbose logging at the "cost" of potentially having very long running idle jobs that are not executing any tasks.

The answer that you provide about in-process tasks is exactly what we needed to know. Thank you.

Can I piggy back on three more questions? Or would you rather see them via email? I'll post them below, in case you want to see them here. Happy to email.

Additional Questions

We have a multi-user endpoint (MUE) that is working to launch jobs via the SlurmProvider.

1. The MUE configuration has variables the {{ ACCOUNT }} and {{ COMMAND }}. How does one pass those in at runtime from the flow, e.g., from a user defined input.json. We are unsure what the syntax should be.
2. How can a flow support parallel execution? Say the inputs are an array of files to be processed. Must this be
   $n$-independent flow executions or is there an equivalent to the AWS "Map" executor? We do not see a clean way to use {"inputs":["a", "b", "c"]} in the input.json file without hard coding multiple tasks.

[ryanchard]

Hello, adding a little more background here. One thing that wasn't mentioned was the max_idletime option in the endpoint config. This is the maximum duration in seconds that workers are allowed to idle before they are marked for termination. I believe the default is 2 minutes, so even if you have a 12 hour walltime and you submit just one task that only takes 5 minutes, the node will be released after roughly 7 minutes. The downside to setting a large default walltime is that it will take longer to go through the queue. I'd suggest making walltime a configurable option like your {{ ACCOUNT }} field.

To answer these questions:

1. These fields are specified under the user_endpoint_config field in the task input. There is an example here: https://globus-compute.readthedocs.io/en/latest/actionprovider.html#v3-action-input-schema
2. I don't believe the Map executor is supported in Flows. However, the task payload for a Compute action is a list of tasks. Assuming your endpoint is configured to support parallel execution (e.g., multiple workers per node, or scaling to multiple nodes), those tasks will run concurrently. It is also possible to dynamically set that list of tasks either by doing some payload manipulation with expressions in the flow itself, or by having one function generate the payload for a subsequent function. Does that answer your question?

[derek-globus]

Hey Jay, Flows developer here.

I'd basically +1 everything Ryan said.
Flow runs will keep 2 to 3 log records for each state execution, even if that execution takes 10+ days. So a single state approach will keep your flow run log count down and offload the polling to us.

Similarly, we don't support any parallelism constructs (e.g., Map) within flows itself. To achieve multi-invocation fanout, use expressions to format your user's input into multiple tasks within the single compute state (computev3).

---

In case it's helpful, here's an example flow definition which maps a list of user-input files onto a variable number of compute tasks:

• Definition: flow-definition
• Input Schema:

  {
    "required": ["file_list"],
    "properties": {
       "file_list": {
          "type": "array",
          "items": {
             "type": "string"
          }
       }
    }
  }

User invocation looks like:

[jlaura]

This discussion has been exceptionally helpful. Thank you! We have a working multi-user endpoint using the SlurmProvider. Job tracking, logging, and queuing are working wonderfully. Many thanks!

The jinja2 templates are working and jobs are executing with variable walltimes and with exclusive turned off by default (as we allow shared nodes on this particular system). I have copied the user_config_template_yaml.lj2 below in case it is of interest.

A few questions related to conda environments. In the examples and doc strings, I see something as simple as conda activate ENV. On our system, this is being silently ignored and am I not sure if this is a system configuration issue or related to how the workers are being provisioned.

Two questions:

1. To get a conda env to activate for code execution in the Slurm provisioned worker, I am having to include the following worker_init. Is this expected? module load miniforge/latest;source /PATH/miniforge3/etc/profile.d/conda.sh; conda activate gdal3.9.0. This is is loading the module, sourcing the conda.sh, and then activating the environment. This seems like I am missing something based on how simple the examples are.
2. What are the environment requirements for an activated env? A matching python version, globus-compute, globus-compute-endpoint, dill, parsl, dill, cloudpickle? Or can the sourced environment (in the worker_init) differ from the globus-compute-endpoint environment. I am seeing parsl version mismatch errors and now serialization errors.

As always, many thanks! We are close to being able to deploy this for our users which is always quite exciting!

display_name: HovenweepCompute
engine:
    type: GlobusComputeEngine
    max_workers_per_node: {{ MAX_WORKERS | default(4) }}
    worker_debug: False

    address:
        type: address_by_interface
        ifname: ib0

    provider:
        type: SlurmProvider
        partition: {{ PARTITION | default('cpu') }}
        account: {{ ACCOUNT }}
        exclusive: {{ EXCLUSIVE | default('false') | lower == 'true' }}

        launcher:
            type: SrunLauncher

        # string to prepend to #SBATCH blocks in the submit
        # script to the scheduler
        # e.g., "#SBATCH --constraint=knl,quad,cache"
        # #SBATCH directives process in order. Set reasonable defaults and then
        # allow the user to override those.
        scheduler_options: {{ OPTIONS | default("") }}

        # Command to be run before starting a worker
        # e.g., "module load anaconda3; source activate gce_env"
        worker_init: {{ COMMAND | default("") }}

        nodes_per_block: 1
        init_blocks: 0
        min_blocks: 0
        max_blocks: 1
        walltime: {{ WALLTIME | default('00:30:00') }}

[khk-globus]

In the examples and doc strings, I see something as simple as conda activate ENV. On our system, this is being silently ignored and am I not sure if this is a system configuration issue or related to how the workers are being provisioned.

More below, but the short of it is that other than "more cleanly," the child processes (UEPs) are not started in a special way. See the first note on the MEP documentation page, that says:

Note

The only difference between a “normal” endpoint process and a UEP is a semantic one for discussion so as to differentiate the two different starting contexts. An endpoint is started manually by a human, while a UEP will always have a parent MEP process.

Which, as implied by my "more cleanly" statement above, perhaps isn't strictly true. Care is taken to be secure, care is taken not to leak context, and so forth.

1. To get a conda env to activate for code execution in the Slurm provisioned worker, I am having to include the following worker_init. Is this expected? module load miniforge/latest;source /PATH/miniforge3/etc/profile.d/conda.sh; conda activate gdal3.9.0. This is is loading the module, sourcing the conda.sh, and then activating the environment. This seems like I am missing something based on how simple the examples are.

I suspect that the problem is that the MEP is much, much more strict about pesky details like environment variables — an absolute requirement when designing with security in mind. Those examples were written before we had the MEP concept, and are designed around a regular user logging in (i.e., ssh) and running a single-instance endpoint. In that instance, the standard setup will typically inject environment variables during the ssh session startup, and then those are passed naively along to the subprocesses.

In contrast, the MEP passes the minimum set of environment variables by default, which includes PATH, HOME, USER, PWD, and CWD. The PATH is a default path, that includes the directory of the Python interpreter that's running the MEP. (See same link.)

So, my suspicion is that the system can't find the conda executable, or is finding the wrong one.

Two avenues that may be of interest:

• Take advantage of PAM.
• Think about using a "man in the middle" shell script. See Advanced Environment Customization for a starter example and explanation.

2. What are the environment requirements for an activated env? A matching python version, globus-compute, globus-compute-endpoint, dill, parsl,, dill, cloudpickle? Or can the sourced environment (in the worker_init) differ from the globus-compute-endpoint environment.

Mostly just that: matching "everything." The key issue that is difficult to robustly handle is that Compute serializes the requested function at registration and reconstitutes it on the endpoint at task invocation time. This is in contrast to something like AWS Lambda, where the entire code file is given to the run-site. Robustly deserializing strictly the function code in an entirely different process (and geographic location!) is fraught with low-level details, like opcodes that may not even exist on the invoking side. Consequently, the best chance of success is "match everything."

That said, the function is not deserialized in the endpoint — it's deserialized on the worker, so if you're game for adventure, have fun!

Meanwhile, note that the MEP is an entirely different process (unix PID) than the child UEP processes it starts, so while we strongly encourage that the SDK/function/UEP environments match, that doesn't necessarily apply to the MEP, which never sees tasks.

We are close to being able to deploy this for our users which is always quite exciting!

This is exciting for us too. Keep us posted, please!