Replies: 5 comments 16 replies
-
|
Question: should the function body be executed |
Beta Was this translation helpful? Give feedback.
-
|
Yes, You are right, the code is ugly and even throw a |
Beta Was this translation helpful? Give feedback.
-
|
Oh interesting -- a yield could work -- that would make it explicit what is before and what's after. Also how much value is there in the decorator -- versus putting those values in a dictionary/TypedDict/object to return as part of the yield? @work_at('./calc_energy')
@slurm
def calc_energy(num_molecules: int) -> Generator[dict, dict, float]:
# before:
# here we can code or render a script
# w.r.t to input
with open('calc_energy.sh', 'r') as f:
f.write(f"mpprun lmp -in run.in -var nmol {num_molecules}")
submit_dict = dict(cmd='source calc_energy.sh',
type='slurm',
is_block=True,
config={
'-A': 'snic2022-5-658',
'-n': 4,
'-J': 'JOB_NAME',
'-t': '07-00:00:00',
}
)
sub_result = yield submit_dict # note we could send the std out/error and status code back in here.
# after:
# extract result from `energy.txt`
result = fn(`energy.txt`)
return resultOtherwise question on the |
Beta Was this translation helpful? Give feedback.
-
Brilliant! Completely avoid top-heavy, I like it!
My immature idea is, that either all the arguments are put in the yield, or put
Yes, map 1-1 with a function and act as a contextmanager. I dont think we could make that a default requirement because we dont know which directory user want to submit the task or more generally, execute a command. I need |
Beta Was this translation helpful? Give feedback.
-
Right this is where we could by convention make it use that by default to match, unless provided otherwise. This could help force people to use a similar structure/approach.
My assumption was that for each |
Beta Was this translation helpful? Give feedback.
-
|
So sorry I replied late since I got sick those days. You are right, sometimes or smaller projects may not need to change the directory, especially when there is only one submission. Adhering to the principle that an API only does one thing, I think submit() and work_at() should be separated . Especially I will use the work_at() function in other tasks |
Beta Was this translation helpful? Give feedback.
-
|
Question: what customization would we want to enable with respect to decorator parameters? It could be possible to do the following. @work_at(source("bar_dir")) # <---
@submit('sleep 1000',
type='slurm',
is_block=True,
config=source("bar_config")) # <---
def bar(foo: str) -> str:
return fooso that we could do the following: result = dr.execute([...], inputs={"bar_config": dict(...), "bar_dir": "some/path/to/folder"}) |
Beta Was this translation helpful? Give feedback.
-
|
Sounds good! I'm not familiar with Hamilton right now, does this mean we need to implement a new driver or GraphAdapter? If not, and still user can replace |
Beta Was this translation helpful? Give feedback.
-
|
What would be some example procedural code that we'd want the Hamilton code to map to? -- just to ensure it's clear what Hamilton is replacing/helping with. TODO: provide example |
Beta Was this translation helpful? Give feedback.
-
|
Here is an example behand submit API, but decouple with Hamilton. I hope those function can be handle with Hamilton and make it integrate into a DAG: Ref code: |
Beta Was this translation helpful? Give feedback.
-
|
@Roy-Kid yep thanks -- does this map to what you're thinking in terms of how things would be executed?
|
Beta Was this translation helpful? Give feedback.
-
|
Yes, exactly! The point of discussion narrowed down to whether submit should be executed before or after the function. If we can introduce a |
Beta Was this translation helpful? Give feedback.
-
|
OK, @skrawcz and I thought this over. I think this is a nice solution. What we do:
Then, the function you have with the # module
def some_var() -> str:
...
@slurm(...)
def slurm_job_1(some_var: str) -> dict:
return _make_config(...)
@slurm(...)
@tag(slurm_execute=True)
def slurm_job_2(some_var: str) -> dict:
return _make_config(...)
def results(slurm_job_1: dict, slurm_job_2: dict) -> dict:
return ...
# 1. break into four tasks
# 2. run through each task
# 2.a if not decorated with slurm -- execute as normal
# 3.a if decorated with slurm -- pass to SlurmExecutor
# 3. SlurmExecutor should wait and pass some indicator of the results, interpreted as node output
# execution manager, this is custom but let's push this back to hamilton
class RemoteDelegatingExecutionManager(DefaultExecutionManager):
def __init__(local_executor, remote_executor, indicator_tag="should_execute_remotely"):
super(...)
self.indicator_tag = indicator_tag
def get_executor_for_task(self, task: DefaultExecutionManager) -> TaskExecutor:
is_single_node_task = len(task.nodes) == 1
if not is_single_node_task:
raise ValueError("Only single node tasks supported")
node, = task.nodes
if indicator_tag in node.tags:
return self.remote_executor
return self.local_executor
def slurm(params):
# Decorator that:
# 0. Tags with the indicator above, with fn = tag(is_slurm_fn=True)(fn)
# 1. takes a function,
# 2. uses it to evaluate a config
# 3. launches a task
# 4. Polls that task on a loop
# 5. Returns the result or erorrs out (depending on how you want to do failure management)
# Call out to your class for these?
dr = (
driver
.Builder()
.enable_dynamic_execution(...)
.with_execution_manager(
RemoteDelegatingExecutionManager(
SynchronousLocalTaskExecutor(),
MultiThreadingExecutor(),
indicator_tag="is_slurm_fn")
)
)
dr.execute(["results"]) |
Beta Was this translation helpful? Give feedback.
-
|
Here's a simple proof of concept that works: # cmdline.py
import functools
import subprocess
from hamilton.execution.executors import DefaultExecutionManager, TaskExecutor
from hamilton.execution.grouping import TaskImplementation
class CMDLineExecutionManager(DefaultExecutionManager):
def get_executor_for_task(self, task: TaskImplementation) -> TaskExecutor:
"""Simple implementation that returns the local executor for single task executions,
:param task: Task to get executor for
:return: A local task if this is a "single-node" task, a remote task otherwise
"""
is_single_node_task = len(task.nodes) == 1
if not is_single_node_task:
raise ValueError("Only single node tasks supported")
node, = task.nodes
if "cmdline" in node.tags: # hard coded for now
return self.remote_executor
return self.local_executor
def cmdline_decorator(func):
"""Decorator to run the result of a function as a command line command."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
# Get the command from the function
cmd = func(*args, **kwargs)
# Run the command and capture the output
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
# Return the output
return result.stdout
return wrapper# funcs.py --- hamilton functions
from cmdline import cmdline_decorator
from hamilton.function_modifiers import tag
@tag(cmdline="yes")
@cmdline_decorator
def echo_1(start: str) -> str:
return f'echo "1: {start}"'
@tag(cmdline="yes")
@cmdline_decorator
def echo_2(echo_1: str) -> str:
return f'echo "2: {echo_1}"'
@tag(cmdline="yes")
@cmdline_decorator
def echo_2b(echo_1: str) -> str:
return f'echo "2b: {echo_1}"'
@tag(cmdline="yes")
@cmdline_decorator
def echo_3(echo_2: str, echo_2b: str) -> str:
return f'echo "3: {echo_2 + ":::" + echo_2b}"'# -- run.py
from hamilton.execution.executors import SynchronousLocalTaskExecutor, MultiThreadingExecutor
if __name__ == '__main__':
from hamilton import driver
from cmdline import CMDLineExecutionManager
import funcs
dr = (
driver
.Builder()
.enable_dynamic_execution(allow_experimental_mode=True)
.with_execution_manager(
CMDLineExecutionManager(
SynchronousLocalTaskExecutor(),
MultiThreadingExecutor(5))
)
.with_modules(funcs)
.build()
)
print(dr.list_available_variables())
# for var in dr.list_available_variables():
# print(dr.execute([var.name], inputs={"start": "hello"}))
print(dr.execute(["echo_3"], inputs={"start": "hello"}))So the "decorator" just needs to know how to handle slurm. And creating the DAG works nicely too: The result with this design is then:
I haven't prototyped the decorator wrapping a generator like we discussed above, but I think that's also a design possibility with this design. The work_at decorator could still work I think, but the order of the decorators would matter here. |
Beta Was this translation helpful? Give feedback.
-
|
Here's the decorator supporting a generator function to sandwich pre and post processing for the command: def cmdline_decorator(func):
"""Decorator to run the result of a function as a command line command."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
if inspect.isgeneratorfunction(func):
# If the function is a generator, then we need to run it and capture the output
# in order to return it
gen = func(*args, **kwargs)
cmd = next(gen)
# Run the command and capture the output
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
try:
gen.send(result)
raise ValueError("Generator cannot have multiple yields.")
except StopIteration as e:
return e.value
else:
# Get the command from the function
cmd = func(*args, **kwargs)
# Run the command and capture the output
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
# Return the output
return result.stdout
if inspect.isgeneratorfunction(func):
# get the return type and set it as the return type of the wrapper
wrapper.__annotations__["return"] = inspect.signature(func).return_annotation[2]
return wrapperThen a function could look like this: @tag(cmdline="yes")
@cmdline_decorator
def echo_2b(echo_1: str) -> [str, CompletedProcess, str]:
# preprocess
print("preprocess")
msg = f'echo "2b: {echo_1}"'
completed_process = yield msg
# postprocess
print("postprocess")
output = completed_process.stdout + "!!!"
return output |
Beta Was this translation helpful? Give feedback.
-
|
What a nice design, it is quite general! I can immediately use those codes in my project to invoke other scientific binary. I think the submission of slurm can be based on the cmd decorator and monitoring code to do so. |
Beta Was this translation helpful? Give feedback.
-
|
Things to figure out:
Example user flow - 1 with polling:
Example user flow - 2 with fire & forget:
|
Beta Was this translation helpful? Give feedback.
-
This is a brainstorming discussion on what the shape of the API should be to support some decorators/functionality that would submit to slurm. The catalyst for this issue is @Roy-Kid !
For example, given a sketch of an API:
what do we want it to do and how do we want users to think?
Beta Was this translation helpful? Give feedback.
All reactions