Loops - Concurrent execution

[Sriram2631]

I am trying to make a For node fire off run signals to all the child classes at once to make them run concurrently.

Probably naive but from what I gleaned, the executor is assigned when a child node/node instance in the sequence is reached. I tried to see if setting the executor apriori would allow for running the child nodes/node instances concurrently.

Would this be the right approach? I see that doing so, particularly when the body_node is a macro, eventually causes errors, but I wanted to confirm the approach before debugging.

I tested it out by simply adding body_node_executor = ( ProcessPoolExecutor, (), {"max_workers": 4}), but I am not sure if this is the way to go.

[liamhuber]

body_node_executor is indeed the right way to go, and will allow body node instances to launch (nearly) simultaneously. Having a macro isn't a fundamental problem. There are four subtleties:

1. ProcessPoolExecutor requires all the nodes to be somewhere importable, so defining them in your notebook won't work
2. Using the executor-instructions creates a separate executor for each body node instance, while assigning an instance gives them all the same executor
3. For-loop macros are dynamic, so if the for-loop runs over a macro, you cannot specify an executor for a child of the macro explicitly; all the children of the macro will be executing in the context of the macro's executor
4. Nesting executors (usually) doesn't work

Let's take a simple setup:

import time

import pyiron_workflow as pwf

@pwf.as_function_node("start", "end", "dt")
def Snore(t_init, t_sleep):
    t_start = time.time()
    time.sleep(t_sleep)
    t_finish = time.time()
    return t_start - t_init, t_finish - t_init, t_finish - t_start

@pwf.as_macro_node
def SnoreMacro(self, t_init, t_sleep):
    self.snoring = Snore(t_init, t_sleep)
    start = self.snoring.outputs.start
    end = self.snoring.outputs.end
    dt = self.snoring.outputs.dt
    return start, end, dt

This will let us see the absolute start and end times for the operation, and the dt as a quick sanity check. For a ProcessPoolExecutor we'll need to put it somewhere importable. For this example, I just stuck a new file in the pyiron_workflow.nodes submodule real quick because I know it will already be available in my path. If you want to put everything in your notebook, you can run all the rest of this with a ThreadPoolExecutor instead; the exact time values will differ, but the principles will be identical.

We can use this to build a simple workflow:

import time
from concurrent import futures

import pyiron_workflow as pwf
from pyiron_workflow.nodes.my_importable_nodes import SnoreMacro

wf = pwf.Workflow("concurrent_starts")
wf.my_loop = pwf.for_node(
    body_node_class=SnoreMacro,
    iter_on="t_sleep",
    t_init=time.time(),
    t_sleep=[1, 2, 3, 4],
    output_as_dataframe=False,
)

If we run this with an executor instance,

with futures.ProcessPoolExecutor(max_workers=2) as executor:
    wf.my_loop.body_node_executor = executor
    out = wf.run()

we get the following:

{'my_loop__t_sleep': [1, 2, 3, 4],
 'my_loop__start': [5.302109956741333,
  4.263456106185913,
  1.2538471221923828,
  1.2926840782165527],
 'my_loop__end': [6.303383827209473,
  6.268503904342651,
  4.259019136428833,
  5.297932147979736],
 'my_loop__dt': [1.0012738704681396,
  2.0050477981567383,
  3.00517201423645,
  4.005248069763184]}

Here, we can see that (the last) two nodes started running simultaneously, and then as the workers became available (3 and 4 seconds later), the remaining jobs started.

If instead we use executor instructions,

wf.my_loop.body_node_executor = (futures.ProcessPoolExecutor, (), {"max_workers": 2})
out=wf.run()

we get

{'my_loop__t_sleep': [1, 2, 3, 4],
 'my_loop__start': [0.8295421600341797,
  0.8294551372528076,
  0.8294589519500732,
  0.8294720649719238],
 'my_loop__end': [1.8346118927001953,
  2.8345022201538086,
  3.8345260620117188,
  4.834505081176758],
 'my_loop__dt': [1.0050697326660156,
  2.005047082901001,
  3.0050671100616455,
  4.005033016204834]}

Now all four nodes start at once. This is because each node is getting is own ProcessPoolExecutor instance, and each of those nodes (uselessly, in this case) has access to two workers!

Even when you are assigning a .executor instead of a .body_node_executor, this is always the case: assigning by instruction means the assignee is getting its own executor instance. Suppose instead of this silly SnoreMacro, I had some set of 8 VaspMacro unit cell calculation that I wanted to parallelize on my 4-core laptop. Then if I use .body_node_executor = ProcessPoolExecutor(max_workers=2), I would nicely run the calculations in pairs, each with two cores for a mpirun -n 2 command somewhere inside the macro. On the other hand, if I used .body_node_executor = (ProcessPoolExecutor, (), {"max_workers": 2}), my internal mpirun -n 2 command would still be fine, but all eight jobs would get launched at once! I'd be overloading my cores and maybe risking memory overruns. I.e. setting with instructions vs setting with an instance has slightly different behaviours.

I think that answers your questions completely, but I just want to follow up on points (3) and (4) a bit more.

Regarding (3): The way for-loops work, we actually re-instantiate the child graph at run-time, so this is the maximal control you'll ever get for assigning executors to children -- we can't go inside SnoreMacro and assign an executor to snoring in particular. You can do this for static macros though, as shown in the test suite

Regarding (4): this is the idea of nesting executors inside each other. I don't have extensive testing on this, and there are times when it works ok and times when it blows up in your face. AFAIK there are nice nesting options in executorlib, and you might have luck naively nesting the wrapped executorlib executors we provide. For the built-in python executors some limitations will be more fundamental. If you wind up really needing this sort of power for a particular scientific workflow, let me know and we can work with Jan to figure something out.

In the meantime, an example of the fragility I'm talking about is that if we

Try combinations of nested executors

import time
from concurrent import futures

import pyiron_workflow as pwf
from pyiron_workflow.nodes.my_importable_nodes import SnoreMacro

from pyiron_workflow.executors.wrapped_executorlib import NodeSingleExecutor

for wf_exe, child_exe in (
    (
        (futures.ProcessPoolExecutor, (), {"max_workers": 4}),
        (futures.ProcessPoolExecutor, (), {"max_workers": 1}),
    ),
    (
        (futures.ProcessPoolExecutor, (), {"max_workers": 4}),
        (futures.ThreadPoolExecutor, (), {"max_workers": 1}),
    ),
    (
        (futures.ThreadPoolExecutor, (), {"max_workers": 4}),
        (futures.ProcessPoolExecutor, (), {"max_workers": 1})
    ),
    (
        (futures.ThreadPoolExecutor, (), {"max_workers": 4}),
        (futures.ThreadPoolExecutor, (), {"max_workers": 1}),
    ),
    (
        (NodeSingleExecutor, (), {}),
        (NodeSingleExecutor, (), {}),
    ),
    (
        (futures.ProcessPoolExecutor, (), {}),
        (NodeSingleExecutor, (), {}),
    ),
    (
        (NodeSingleExecutor, (), {}),
        (futures.ProcessPoolExecutor, (), {}),
    ),
    (
        (futures.ThreadPoolExecutor, (), {}),
        (NodeSingleExecutor, (), {}),
    ),
    (
        (NodeSingleExecutor, (), {}),
        (futures.ThreadPoolExecutor, (), {}),
    ),
):
    wf = pwf.Workflow("concurrent_starts")
    wf.my_loop = pwf.for_node(
        body_node_class=SnoreMacro,
        iter_on="t_sleep",
        t_init=time.time(),
        t_sleep=[1, 2, 3, 4],
        output_as_dataframe=False,
    )
    wf.executor = wf_exe
    wf.my_loop.body_node_executor = child_exe
    try:
        out=wf.run()
        out.result()
        wf.outputs.to_value_dict()
        print(wf_exe[0].__name__, child_exe[0].__name__, "Works")
    except Exception as e:
        print(wf_exe[0].__name__, child_exe[0].__name__, "Raises", type(e).__name__)

It only works without further debugging when at least one of the (workflow executor, body node executor) executors is a simple thread-pool executor:

ProcessPoolExecutor ProcessPoolExecutor Raises BrokenProcessPool
ProcessPoolExecutor ThreadPoolExecutor Works
ThreadPoolExecutor ProcessPoolExecutor Works
ThreadPoolExecutor ThreadPoolExecutor Works
NodeSingleExecutor NodeSingleExecutor Raises KeyError
ProcessPoolExecutor NodeSingleExecutor Raises BrokenProcessPool
NodeSingleExecutor ProcessPoolExecutor Raises KeyError
ThreadPoolExecutor NodeSingleExecutor Works
NodeSingleExecutor ThreadPoolExecutor Works

I expect that the BrokenProcessPool errors are fundamentally unavoidable, but if the need arises it might be possible to debug and resolve the KeyError cases.

EDIT: I had modified Snore and SnoreMacro to include end in addition to start and dt, but had failed to propagate that change back to the example code. If you read this a few hours ago, they were inconsistent.

[Sriram2631]

Hi, thanks a lot! The examples work perfectly!

Regarding (3): Ah, I understand. But would that hinder the workflow or make it waste resources somehow? If my understanding is right, say within the SnoreMacro we include a second process (just calling the same function here). We do not need a the executors to be assigned to the functions per se, since the operations within the macro still need to proceed one by one. i.e., the executor assigned for self.snoring will also be the same one for self.snoring2, i.e., the execution has to be serial here (Sorry if I misunderstood, please correct me if I am wrong).

@pwf.as_macro_node
def SnoreMacro(self, t_init, t_sleep):
    self.snoring = Snore(t_init, t_sleep)
    start = self.snoring.outputs.start
    end = self.snoring.outputs.end
    dt = self.snoring.outputs.dt

   self.snoring2 = Snore(t_init+end, t_sleep)

    return start, end, dt

Secondly,

I tried to include the same in my workflow, using both assigning routes

wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2})

and

with futures.ThreadPoolExecutor(max_workers=2) as executor:
        wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2})

However, in both cases, the execution still seems serial. For example, here is a short form of the dataframe generated by the for loop node, with the starting time.

strain_value 	timestart 	WrappedResults
0 	-0.00010 	34.066719 	{'gamma': 0.135900465475471, 'Area': 122.92526...
1 	-0.00009 	32.283479 	{'gamma': 0.13590130615448284, 'Area': 122.926...
2 	-0.00008 	30.571820 	{'gamma': 0.13590214592499245, 'Area': 122.927...
3 	-0.00007 	29.108218 	{'gamma': 0.13590298478650775, 'Area': 122.928...
4 	-0.00006 	27.448852 	{'gamma': 0.13590382273946222, 'Area': 122.930...
5 	-0.00005 	25.814411 	{'gamma': 0.1359018328086625, 'Area': 122.9314...
6 	-0.00004 	24.172856 	{'gamma': 0.13590306982161435, 'Area': 122.932...
7 	-0.00003 	22.536442 	{'gamma': 0.13590390495110757, 'Area': 122.933...
8 	-0.00002 	20.865917 	{'gamma': 0.13590473917239487, 'Area': 122.935...
9 	-0.00001 	18.983033 	{'gamma': 0.1359055724854388, 'Area': 122.9363...
10 	0.00000 	17.090322 	{'gamma': 0.13590357806939296, 'Area': 122.937...
11 	0.00001 	15.404674 	{'gamma': 0.13590441049077748, 'Area': 122.938...
12 	0.00002 	13.629515 	{'gamma': 0.13590524200503884, 'Area': 122.940...
13 	0.00003 	11.881130 	{'gamma': 0.13590607261195334, 'Area': 122.941...
14 	0.00004 	10.298518 	{'gamma': 0.13590690231163574, 'Area': 122.942...
15 	0.00005 	8.612857 	{'gamma': 0.13590773110441576, 'Area': 122.943...
16 	0.00006 	6.947896 	{'gamma': 0.13590855899018886, 'Area': 122.944...
17 	0.00007 	5.329315 	{'gamma': 0.13590938596919538, 'Area': 122.946...
18 	0.00008 	3.541730 	{'gamma': 0.13591021204162435, 'Area': 122.947...
19 	0.00009 	1.918373 	{'gamma': 0.1359110372075515, 'Area': 122.9486...
20 	0.00010 	0.249597 	{'gamma': 0.13591186146699663, 'Area': 122.949...

You can find the codes for the workflow here...
https://github.com/pyiron/Workflows_SA/blob/07122b061dd6d246c27559979d5cc27f7c5c695e/Workflows/SurfaceStressWorkflow.py#L1019

Please let me know of your thoughts on this.

While this is not an issue when running LAMMPS calculations with smaller structures where each execution within the loop takes about a minute or two. When it comes to VASP calculations or larger structures in lammps, say, the jobs are submitted to the queue, it seems wasteful to have it for the loop to go through it one-by-one instead of submitting the 10 or 20 calculations at once to the queue - given that they are all independent of each other.

Thanks!

[liamhuber]

I.e., the executor assigned for self.snoring will also be the same one for self.snoring2, i.e., the execution has to be serial here

Yes, that's totally correct. In the case that both snoring and snoring2 have similar resource requirements (e.g. both 16 core VASP calculations), this is not inefficient; similarly it's not inefficient if one of these is blazing fast (e.g. a 16-core VASP calculation, and another node doing some python on a single core for 13ms). The inefficiency comes if you have two long-duration calculations with different requirements (e.g. a 16-core VASP calculation followed by an 8-core VASP calculation). This sort of deeply-nested resource allocation for dynamic workflows (for-, while-, if-, etc.) is interesting and needs a good solution, I simply don't have one yet.

You can find the codes for the workflow here...

It does sound strange that your attempts still yield serial execution. I'd be happy to take a look, but the repo must be private as I get a 404, even when I just try to go to the repo landing page. Can you invite me to it?

[Sriram2631]

Indeed yes! I'll try to rework the looping and my macro to avoid this.

I spoke with Jan about the repo (access/permissions), please check if you can access it now.

Currently the macro has two "large" calculations - 1.) Minimization in pbc -> 2.) Minimization with surfaces. Instead I could make two loops so that they only contain one of the minimizations within.

A quick follow up, if the VASP calculations (and their resource allocations) are taken over, so to speak, by Slurm. The functions within the macro would then need just the resources to wait for completion and parse the outputs. Does this perhaps (albeit inelegantly) circumvent this allocation inefficiency?

Thanks!

[liamhuber]

I spoke with Jan about the repo (access/permissions), please check if you can access it now.

🚀 I see it.

A quick follow up, if the VASP calculations (and their resource allocations) are taken over, so to speak, by Slurm. The functions within the macro would then need just the resources to wait for completion and parse the outputs. Does this perhaps (albeit inelegantly) circumvent this allocation inefficiency?

And now that I see it, I see that you mean you're using a regular python node to launch a SLURM job (via pyiron_base), and not that you're putting SLURM-capable executors on your nodes.

Yeah, with this route the downside is that you'll need to make sure that the node's process stays alive for the entire duration of the SLURM job. I'm not a huge fan of mixing and matching pyiron_base and pyiron_workflow, but in this particular case there's a chance it helps offset this advantage. There's at least some chance here that if you're parent workflow process dies, and you naively re-run it, that pyiron_base will be robust enough to discover its running/submitted/finished SLURM job and grab that instead of starting a fresh one for you. I wouldn't count on it, but you could check what the interaction is like.

The upside with this route is that since the node is basically just doing some launching and parsing, you're absolutely spot-on that handling the body nodes in the same process but different threads with a simple ThreadPoolExecutor is probably sufficient. You might need to keep an eye on your RAM consumption if there are intermediate steps in the parsing that get heavy, but don't worry about it unless it becomes a problem.

Currently the macro has two "large" calculations - 1.) Minimization in pbc -> 2.) Minimization with surfaces. Instead I could make two loops so that they only contain one of the minimizations within.

Yes, I see that now: get_state_periodic_cell and get_state_open_surface. In your particular setup, it sounds like no changes are necessary: You're already "cheating" by using pyiron_base to manage deeply nested resource allocation -- i.e. the simple nodes launch their own SLURM jobs. You'll need to keep a parent process alive to run the workflow, but inside that process if your graph is just a bunch of nodes waiting for the SLURM jobs to finish on different threads, that's relatively efficient.

More generally, this is exactly the sort of case where we would run into trouble: the periodic cell calculation might be expensive and slow, but probably requires fewer cores than the surface calculation. If you ran the macro directly, you could give the different children different executors (with different node allocations (and in parallel since they are independent!)). But when you put the macro inside a for-loop, we only offer the ability to set one executor for the entire macro.

I can already imagine a scenario where we expose a new dictionary argument and leverage the names of the macro's children, like

    body_node_child_executors = {
        "PeriodicCellMinimization": (ProcessPoolExecutor, ...),
        "MinimizationWithSurfaces": (...),
    }

But then what if these are macros and their children need resolution? It's turtles all the way down. I'll try and think of a more elegant solution, as I do see and understand the need.

Secondly,

I tried to include the same in my workflow, using both assigning routes

wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2})

and

with futures.ThreadPoolExecutor(max_workers=2) as executor:
        wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2})

However, in both cases, the execution still seems serial. For example, here is a short form of the dataframe generated by the for loop node, with the starting time.

Aha, I think I found it. I can see now that wf.GetSurfaceEnergiesAtDifferentStrain is a macro node, who's subgraph is a for-loop node. When you assign wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor it's not doing anything, it's just a regular old python attribute assignment and never gets looked at. Regular macro's don't have the concept of a body_node_executor because after instantiation you can see all their children and give them executors directly!

I.e. try instead wf.GetSurfaceEnergiesAtDifferentStrain.DataFrame.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2}), where I've looked at your definition for SurfaceStressComputeForLoop, and accessed the child DataFrame.

Alternatively, since this macro is anyhow just a single node, you could put it directly into the wf instead of wrapping in a macro.

Here's a toy example that demonstrates all this:

import time
from concurrent import futures

import pyiron_workflow as pwf

@pwf.as_function_node("start", "end", "dt")
def Snore(t_init, t_sleep):
    t_start = time.time()
    time.sleep(t_sleep)
    t_finish = time.time()
    return t_start - t_init, t_finish - t_init, t_finish - t_start

@pwf.as_macro_node("start")
def OwnedLoop(self, t_init, t_sleep_list):
    self.loop = pwf.for_node(
        body_node_class=Snore,
        iter_on="t_sleep",
        t_init=t_init,
        t_sleep=t_sleep_list,
        output_as_dataframe=False,
    )
    return self.loop.outputs.start

wf = pwf.Workflow("example")
wf.raw_loop = pwf.for_node(
    body_node_class=Snore,
    iter_on="t_sleep",
    t_init=time.time(),
    t_sleep=[1, 2, 3],
    output_as_dataframe=False,
)
wf.owned_loop = OwnedLoop(time.time(), [1, 2, 3])


wf.owned_loop.body_node_executor = "This is just an attribute we're setting -- it does nothing"
# plain macro's don't support `body_node_excutor`!
# This is just a regular python attribute assignment

with futures.ThreadPoolExecutor(max_workers=8) as executor:
    wf.raw_loop.body_node_executor = executor
    wf.owned_loop.loop.body_node_executor = executor
    wf.run()

print(wf.outputs.raw_loop__start.value, wf.outputs.owned_loop__start.value)
>>> [0.010105133056640625, 0.010020732879638672, 0.00992894172668457],
...  [3.0433127880096436, 3.043303966522217, 3.0432870388031006]

Where you can see that the second node still runs in serial with the first node, but each set of loop bodies is concurrent.

(for fun, you can also make them all start at the same time by slapping the same thread executor on the parent nodes too

...

with futures.ThreadPoolExecutor(max_workers=8) as executor:
    wf.raw_loop.body_node_executor = executor
    wf.owned_loop.loop.body_node_executor = executor
    wf.raw_loop.executor = executor
    wf.owned_loop.executor = executor
    wf.run()

print(wf.outputs.raw_loop__start.value, wf.outputs.owned_loop__start.value)
>>> [0.013653993606567383, 0.013370752334594727, 0.012067794799804688],
... [0.016143798828125, 0.0160369873046875, 0.015960216522216797]

)

LMK if that does it for you.

In the longer term,

• You should touch base with @ligerzero-ai about pyiron_base-free atomistic calculations
• Consider switching from pickle to the "h5bag" storage back end to get workflows saved in HDF5 with versioning metadata
• Consider submitting nodes directly to SLURM with the wrapped-executorlib NodeSlurmExecutor

[Sriram2631]

Thanks!

You should touch base with @ligerzero-ai about pyiron_base-free atomistic calculations

Indeed yes, definitely! Will do! Sorry, I see that I haven't updated this to the repo, will do so in a bit. I have moved the lammps part to the lammps_function(), presently, I am testing out the vasp_function(). I retained a version with pyiron_base primarily for retrieval of completed jobs.

Aha, I think I found it. I can see now that wf.GetSurfaceEnergiesAtDifferentStrain is a macro node, who's subgraph is a for-loop node. When you assign wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor it's not doing anything ...

Short answer:
Indeed yes, I had moved the loop into a macro while testing assigning executors. I tried doing it as

wf.GetSurfaceEnergiesAtDifferentStrain = pwf.Workflow.create.for_node(body_node_class=SurfaceEnergyForStrainValue,
                                                                          iter_on="strain_value",
                                                                          IdataClass = wf.FullyRelax,
                                                                          strain_value= wf.FullyRelax.Strainlist,
                                                                          output_as_dataframe=True
                                                                         )
wf.GetSurfaceEnergiesAtDifferentStrain.body_node_executor = (futures.ThreadPoolExecutor, (), {"max_workers": 2})

which resulted in the following error,

pyiron_workflow.nodes.composite.FailedChildError: /Surface_stress/GetSurfaceEnergiesAtDifferentStrain/DataFrame/body_20 encountered error in child: {'/Surface_stress/GetSurfaceEnergiesAtDifferentStrain/DataFrame/body_20/PeriodicCellMinimization.accumulate_and_run': ValueError('signal only works in main thread of the main interpreter')}

My original question and follow-up should've essentially been about this error, sorry for the run-around.

Please let me know of your thoughts on this.

• Consider switching from pickle to the "h5bag" storage back end to get workflows saved in HDF5 with versioning metadata
• Consider submitting nodes directly to SLURM with the wrapped-executorlib NodeSlurmExecutor

Definitely interesting, will do!

(Probably needless) context:
While trying to debug, the above error - to no avail, I went back to an earlier version where I wrapped it as a separate macro, and amidst the back-and-forth, the unnecessary wrapping as macro remained. Hence the confusion, my apologies.

@pwf.as_macro_node
def SurfaceStressComputeForLoop(self, I:DataInputs.dataclass):
    from concurrent import futures
    
    self.DataFrame = pwf.Workflow.create.for_node(
        body_node_class=SurfaceEnergyForStrainValue,
        iter_on="strain_value",
        IdataClass = I,
        strain_value= I.Strainlist,
    )
    self.DataFrame.body_node_executor=futures.ThreadPoolExecutor(max_workers=2)
    ...
    ...
    return self.DataFrame

[liamhuber]

...which resulted in the following error,

pyiron_workflow.nodes.composite.FailedChildError: /Surface_stress/GetSurfaceEnergiesAtDifferentStrain/DataFrame/body_20 encountered error in child: {'/Surface_stress/GetSurfaceEnergiesAtDifferentStrain/DataFrame/body_20/PeriodicCellMinimization.accumulate_and_run': ValueError('signal only works in main thread of the main interpreter')}

My original question and follow-up should've essentially been about this error, sorry for the run-around.

Please let me know of your thoughts on this.

Despite the fact that pyiron_workflow also uses the term signal, this is not directly related to pyiron_workflow. This is a more generic error. (Cf., e.g., this SO thread.)

The for-loop bodies often (always?) start executing in reverse order, so if you are running this loop over data that is 20 items long, that means this error is being hit right away. Walking the paths in the error message, the only opaque part is that body_20 is a SurfaceStressComputeForLoop instance, then we get nicely directed that the problem is coming from the PeriodicCellMinimization child of that macro -- i.e. a get_state_periodic_cell instance. All I can say from what's given here is that something in that function does not appreciate being run in a separate thread when executed by the node.

I can't actually run the code myself as I don't have the right environment. I tried naively to reproduce it by installing a couple of the dependencies I saw in your code in a fresh conda environment, but the solver gave me some archaic version of pyiron_atomistics. For me to dig further, I'd need your full conda list. However, by way of the next paragraph I don't think this is the right path for us to take.

Ultimately, I suspect you have hit an error that is not the fault of pyiron_workflow at all. Rather, something in the body of get_state_periodic_cell doesn't play nicely with the standard library's ThreadPoolExecutor. Figuring out what that is, or patching it in pyiron_atomstics is outside the scope of what I can do efficiently. However, I can provide some advice on how to track down what, exactly, is causing the problem...

get_state_periodic_cell is just a plain as_function_node, and there's no explicit references to pyiron_workflow inside of it. That means you can just copy-paste the function into a jupyter notebook cell without the as_function_node decorator and run it. My prediction is that this will work fine, but that wrapping the whole thing in a thread pool executor will get you the same "signal..." error you see above.

You can run it in a thread pool following this pattern:

from concurrent import futures

# @pwf.as_function_node  # Nope! Just a regular function
def some_function(x, y):
    return x + y

with futures.ThreadPoolExecutor() as exe:
    f = exe.submit(some_function, 2, 3)
    output = f.result()

print(output)

If my suspicion is correct and this does give you the same error, then either the stack trace will already show you where the root cause is, or you can start breaking the function apart so that only bits of it are submitted to exe and see which line is causing the problem. This won't actually solve the issue, but it will let you figure out where it's coming from.