[Bug] file-based execution violates futures contract

[liamhuber]

I'm still working on getting stuff integrated with pyiron_workflow, and ran into a case where the SlurmClusterExecutor was hanging indefinitely. I was able to reproduce one of my problems locally using LocalFileExecutor:

import os
import time
from concurrent.futures import ThreadPoolExecutor

from executorlib import SingleNodeExecutor
from pyiron_workflow.executors.wrapped_executorlib import LocalFileExecutor

def foo(x):
    time.sleep(x)
    return x

for executor_class, kwargs in (
    (ThreadPoolExecutor, {}),
    (SingleNodeExecutor, {"cache_directory": "cache_sne"}),
    (LocalFileExecutor, {"cache_directory": "cache_lfe"}),
):
    print(executor_class.__name__)
    with executor_class(**kwargs) as executor:
        future = executor.submit(foo, 1)
        print(future)
    print(future.result())
    print(future, "\n")

Gives

ThreadPoolExecutor
<Future at 0x108cc0860 state=running>
1
<Future at 0x108cc0860 state=finished returned int> 

SingleNodeExecutor
<Future at 0x108c71bb0 state=pending>
1
<Future at 0x108c71bb0 state=finished returned int> 

LocalFileExecutor
<Future at 0x10fecc440 state=pending>

And then hangs indefinitely. I can see that cache_lfe/..._i.h5 gets written, it's just not running. This can be avoided if the relevant function is fast enough, or adding a long enough sleep to the with body clause, or (obviously) by moving the future.result() into the with clause. I believe what's happening is that the file-based approach takes some extra time somewhere and gets in a race with TaskSchedulerBase.shutdown such that the clauses checking on self._future_queue and self._process are not behaving appropriately. There is something I'm not understanding deeply enough though, because it works fine for the SingleNodeExecutor even when it's using the filesystem cache.

[jan-janssen]

To me this sounds like a LocalFileExecutor specific bug and not so much an issue in the BaseExecutor.

[liamhuber]

As I said, this is also a problem for the user-facing production SlurmClusterExecutor:

import os
import time
from concurrent.futures import ThreadPoolExecutor

from executorlib import SlurmClusterExecutor

def foo(x):
    time.sleep(x)
    return x

with SlurmClusterExecutor(cache_directory="cache_sce", resource_dict={"partition": "s.cmfe"}) as executor:
    future = executor.submit(foo, 1)
    print(future)
print(future.result())
print(future, "\n")

hangs just as well, although there I do ultimately wind up with a *_o.h5 file there instead of getting stuck at the input, although the output is never registered by the future.

I.e. it appears to be specific to file-based tools so far, but it does impact a real production executor.

[jan-janssen]

I am sorry, I meant file-based. I agree there is a real issue, I just think it should be fixed in the file-based executor class rather than the base class.

[liamhuber]

I forgot to say, #710 does resolve the issue for the slurm example as well. At a minimum, I hope that helps to track down where the cause of problem is.

[jan-janssen]

@liamhuber I proposed a solution in #712 . The part that is unclear to me is the behavior we expect for long running jobs. Let's assume I want to submit jobs and gather their results the next day. This is no longer possible with these changes.

[liamhuber]

Let's assume I want to submit jobs and gather their results the next day. This is no longer possible with these changes.

This is certainly not working on main right now, but is indeed working fine with these changes. Condsider:

import os
import time
from concurrent.futures import ThreadPoolExecutor

from executorlib import SlurmClusterExecutor

def foo(x):
    time.sleep(x)
    return x

with SlurmClusterExecutor(cache_directory="cache_sce", resource_dict={"partition": "s.cmfe"}) as executor:
    future = executor.submit(foo, 1)
    print(future)
print(future.result())
print(future, "\n")

On main, it hangs after the first print statement. On my file system I can watch cache_sce/foo_i.h5 get written and later replaced by ...o.h5. Even then, if I kill the kernel and restart, it still hangs. So I currently have no way to ever recover this result.

With #710 (and presumably #712 given the good behaviour of my local tests), it is true that with these changes the cell will also "hang" waiting for the long-running job to complete. However, if the job finishes, we correctly get the result; if I come back and re-run it later, I correctly recover the hash; if I wait to see the foo_i.h5 file then quickly stop the kernel, watch on my file system for it to become ...o.h5, and re-run the job, I nicely recover the output. So yes, you'll need to interrupt the kernel if you want to use it this way. Maybe there is a "best of both worlds" situation, but for now I think needing to stop the kernel is a small price to pay in convenience to get it to actually work correctly.

[jan-janssen]

On the one hand I set a terminate_function to kill the jobs when then executor is killed and on the other hand when I think about the jobs waiting in the queue, I might want to submit them, close the notebook and come back later when the submitted jobs are completed. I have to do some more tests tomorrow and maybe even integrate the SLURM interface in the tests to validate this behavior.

[liamhuber]

On the one hand I set a terminate_function to kill the jobs when then executor is killed

Oh, yikes. Yes, that would destroy this. I guess my little test with this 1s sleeping node was just faster than all the signals could process and I got lucky when I restarted my notebook and it worked.

when I think about the jobs waiting in the queue, I might want to submit them, close the notebook and come back later when the submitted jobs are completed.

💯 this is exactly the sort of scenario I want to support and envision being normal when people are using the SLURM (or flux) executors.

I have to do some more tests tomorrow and maybe even integrate the SLURM interface in the tests to validate this behavior.

That sounds good. Since one already needs to keep the executor alive to avoid the terminate_function call, would it be possible to get #712 merged and a new version out? This would let me incorporate easy access to SLURM via executorlib in pyiron_workflow under the condition that the graph process stays active the whole time. That's an annoying limitation, but I guess this already enough to streamline the SLURM interface for people like Han who anyhow submit the entire graph to the queue as a simple python script and call executors from inside there.