Status of tasks in hyperqueue

[pmu2022]

I have a general question. In our Python framework that manages and executes dynamic task graphs, I would like to utilize hyperqueue for submission of tasks. Is it possible to receive some callback whether a job I submitted has been completed, so that potential new jobs can be started?

[spirali]

There is wait_for_jobs in Client that blocks until the job is not finished.

E.g.:

job_id = client.submit(job)
client.wait_for_jobs([job_id])

[Kobzol]

It is currently not possible to wait for the completion of individual/specific tasks within a job though. We could add it if you have a use-case :)

[lahwaacz]

If waiting is only for the purpose of starting new jobs/tasks, it can be already achieved with dependencies.

[Kobzol]

Right, of course, you can specify dependencies with the usual semantics. It's not currently possible in the Python API to have an actual callback to react to a task being completed in the "driver" Python script.

[danielstone-opengosim]

Hello,

Sorry to add to an old post here but I think I do have a use case for this. Consider the following prototype code that I have working:

  with LocalCluster() as cluster:
    cluster.start_worker(WorkerConfig(cores=allowed_cores))
    client = cluster.client()

    submitted_list = []
    submitted_progs = []
    for cmd in cmds:
      job = Job()
      prog = job.program(cmd,resources=ResourceRequest(cpus=run_cores))
      submitted_progs.append(prog)
      submitted = client.submit(job)
      submitted_list.append(submitted)

    # Stop and wait for all jobs
    client.wait_for_jobs(submitted_list)

But what my application really needs would be something like:

  with LocalCluster() as cluster:
    cluster.start_worker(WorkerConfig(cores=allowed_cores))
    client = cluster.client()

    submitted_list = []
    submitted_progs = []
    for cmd in cmds:
      job = Job()
      prog = job.program(cmd,resources=ResourceRequest(cpus=run_cores))
      submitted_progs.append(prog)
      submitted = client.submit(job)
      submitted_list.append(submitted)

    # periodically check each job has finished
    all_done = False
    N_done = 0
    while not all_done:
      perform_other_work()                     # do something else in between periodically checking the jobs
      time.sleep(5)
      for job in submitted_list:
         status = get_status(job)              # basically like a future
         if status == finished:                # this is pretty much pseudocode now
           print_custom_run_finished_msg(job)  # tell the user the job is finished, in my own format
           N_done +=1
           if N_done == len(submitted):
             all_done = True                      

You can see that by polling individual jobs for their status (maybe I should be using tasks for this?) I could do two essential things:

• perform other necessary work while the jobs are processing
• do my own logic around whether a job has completed, not least is a custom output to the user

I don't think this kind of freedom is available through task dependencies, but I could be wrong.

I think my imaginary get_status(job) function is something like a wrapper around what would be, in the cli,

hq job info [job-id]

right?

What do you thinks the odds are of something like this functionality being added soon or at all? I could potentially help with contributions.

Many thanks for any feedback.

[danielstone-opengosim]

That could work in the short term. The clunkiest part of this is that it requires the HQ binary to be present as well as the pip install of the python api - at least I think so. I do see hyperqueue.abi3.so in the pip install but it's a bit opaque how the python is able to interact with it, or if I could do arbitrary calls from it.

[Kobzol]

Not easily, it's a shared library, not a binary that you could directly call.

[danielstone-opengosim]

I'm not sure this plan works then. I've tried using a separate HQ binary to check jobs as my prototype is running but there are various issues ("Access token found but HQ server peb3:40903 is unreachable.") Probably not surprising. We've got two instances of the software (one from the binary, from the python through the library) interacting with the same server, which I assume wasn't what it was designed for.

How difficult is the process of adding in something like a job_info() functionality to the python api? Is there any technical documentation that I could look at?

[Kobzol]

As long as the hq binary version is the same as the pyhq version, using it together shouldn't be an issue. Though you would have to pass the correct --server-dir to the hq binary, the same one used by the cluster started by pyhq.

How difficult is the process of adding in something like a job_info() functionality to the python api? Is there any technical documentation that I could look at?

The entrypoint for functions exposed from Rust to Python is here. Most such functions then generate some HQ message that is sent to the server; you could inspect what message is sent by the corresponding HQ CLI command, and send the same message here in the Python wrapper.

[spirali]

Another option (besides busy waiting) is to read journal events (through hq journal stream/replay). I think that we do not have any Python API for this yet. btw: HQ server may just send you type of events that you are interested in; it does not have a CLI options, but it is quite easy to add it.