[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jan-janssen merged 14 commits into from
Jun 3, 2026
Merged

[Bug] Ignore duplicate entries in cache #997

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
92cff67
[Bug] Ignore duplicate entries in cache
jan-janssen Jun 3, 2026
4e16887
fix tests
jan-janssen Jun 3, 2026
9d4c0ad
Add test
jan-janssen Jun 3, 2026
8aa864b
fix error
jan-janssen Jun 3, 2026
501f340
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jun 3, 2026
7fdbe1c
Add duplicate to file based executors
jan-janssen Jun 3, 2026
5d057e8
fix test
jan-janssen Jun 3, 2026
188d480
another fix
jan-janssen Jun 3, 2026
243ffc8
Define duplicate_dict as optional
jan-janssen Jun 3, 2026
ae6e978
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jun 3, 2026
eb1b5e8
fixes
jan-janssen Jun 3, 2026
72f690c
fix coverage
jan-janssen Jun 3, 2026
fe9c8bf
Format black
pyiron-runner Jun 3, 2026
d1701ef
Add docstring
jan-janssen Jun 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions src/executorlib/standalone/hdf.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
with h5py.File(file_name_abs, "a") as fname:
for data_key, data_value in data_dict.items():
if data_key in group_dict:
fname.create_dataset(
name="/" + group_dict[data_key],
data=np.void(cloudpickle.dumps(data_value)),
)
with contextlib.suppress(ValueError):
fname.create_dataset(
name="/" + group_dict[data_key],
data=np.void(cloudpickle.dumps(data_value)),
)


def load(file_name: str) -> dict:
Expand Down
53 changes: 50 additions & 3 deletions src/executorlib/task_scheduler/file/shared.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
process_dict: dict = {}
cache_dir_dict: dict = {}
file_name_dict: dict = {}
duplicate_dict: dict = {}
while True:
task_dict = None
with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
cancel_futures=task_dict.get("cancel_futures", False),
memory_dict=memory_dict,
process_dict=process_dict,
duplicate_dict=duplicate_dict,
cache_dir_dict=cache_dir_dict,
terminate_function=terminate_function,
pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
process_dict[task_key] = queue_id
memory_dict[task_key] = task_dict["future"]
cache_dir_dict[task_key] = cache_directory
elif memory_dict[task_key] != task_dict["future"]:
if task_key not in duplicate_dict:
duplicate_dict[task_key] = []
duplicate_dict[task_key].append(task_dict["future"])
future_queue.task_done()
else:
memory_dict = _refresh_memory_dict(
memory_dict=memory_dict,
cache_dir_dict=cache_dir_dict,
process_dict=process_dict,
duplicate_dict=duplicate_dict,
terminate_function=terminate_function,
pysqa_config_directory=pysqa_config_directory,
backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
task_key: str, future_obj: Future, cache_directory: str
task_key: str,
future_obj: Future,
cache_directory: str,
duplicate_dict: Optional[dict] = None,
) -> Future:
"""
Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
task_key (str): The key of the task.
future_obj (Future): The future object associated with the task.
cache_directory (str): The directory where the HDF5 files are stored.

duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
Returns:
Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
if not os.path.exists(file_name):
return future_obj
exec_flag, no_error_flag, result = get_output(file_name=file_name)
_update_future(
future_obj=future_obj,
exec_flag=exec_flag,
no_error_flag=no_error_flag,
result=result,
)
if duplicate_dict is not None and task_key in duplicate_dict:
for duplicate_future in duplicate_dict[task_key]:
_update_future(
future_obj=duplicate_future,
exec_flag=exec_flag,
no_error_flag=no_error_flag,
result=result,
)
del duplicate_dict[task_key]
return future_obj


def _update_future(
future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
"""
Update the future object with the result of the task execution.

Args:
future_obj (Future): The future object to be updated.
exec_flag (bool): Flag indicating whether the task has been executed.
no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
result (Any): The result of the task execution.
"""
if exec_flag and no_error_flag:
future_obj.set_result(result)
elif exec_flag:
future_obj.set_exception(result)
return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
memory_dict: dict,
cache_dir_dict: dict,
process_dict: dict,
duplicate_dict: Optional[dict] = None,
terminate_function: Optional[Callable] = None,
pysqa_config_directory: Optional[str] = None,
backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
memory_dict (dict): dictionary with task keys and future objects
cache_dir_dict (dict): dictionary with task keys and cache directories
process_dict (dict): dictionary with task keys and process reference.
duplicate_dict (dict): dictionary with task keys and duplicate future objects.
terminate_function (callable): The function to terminate the tasks.
pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
task_key=key,
future_obj=value,
cache_directory=cache_dir_dict[key],
duplicate_dict=duplicate_dict,
)
for key, value in memory_dict.items()
if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
memory_dict: dict,
process_dict: dict,
cache_dir_dict: dict,
duplicate_dict: Optional[dict] = None,
terminate_function: Optional[Callable] = None,
pysqa_config_directory: Optional[str] = None,
backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
cancel_futures (bool): Whether to cancel futures that have not yet started.
memory_dict (dict): Mapping of task keys to their Future objects.
process_dict (dict): Mapping of task keys to process handles or queue IDs.
duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
terminate_function (Callable, optional): Function used to terminate running processes.
pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
memory_dict=memory_dict,
cache_dir_dict=cache_dir_dict,
process_dict=process_dict,
duplicate_dict=duplicate_dict,
terminate_function=terminate_function,
pysqa_config_directory=pysqa_config_directory,
backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
memory_dict=memory_dict,
cache_dir_dict=cache_dir_dict,
process_dict=process_dict,
duplicate_dict=duplicate_dict,
terminate_function=terminate_function,
pysqa_config_directory=pysqa_config_directory,
backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
memory_dict=memory_dict,
cache_dir_dict=cache_dir_dict,
process_dict=process_dict,
duplicate_dict=duplicate_dict,
terminate_function=terminate_function,
pysqa_config_directory=pysqa_config_directory,
backend=backend,
Expand Down
12 changes: 12 additions & 0 deletions tests/unit/executor/test_api.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
self.assertEqual(len(nodes), 4)
self.assertEqual(len(edges), 4)

def test_duplicate_futures(self):
with TestClusterExecutor(cache_directory="cache_dir") as exe:
cloudpickle_register(ind=1)
future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
self.assertFalse(future_1.done())
self.assertFalse(future_2.done())
self.assertEqual(future_1.result(), 3)
self.assertEqual(future_2.result(), 3)
self.assertEqual(len(os.listdir("cache_dir")), 1)

def test_shutdown_wait_false_cancel_futures_false(self):
exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
cancel_futures=True,
memory_dict=memory_dict,
process_dict={},
duplicate_dict=None,
cache_dir_dict={"a": "cache_dir"},
terminate_function=None,
pysqa_config_directory=None,
Expand Down
14 changes: 14 additions & 0 deletions tests/unit/executor/test_single_cache.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
raise ValueError(a)


def sum_with_wait(a, b):
sleep((a + b) / 10)
return a + b


class AddClass:
def __call__(self, a, b):
return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
)

def test_cache_duplicate_function(self):
cache_directory = os.path.abspath("cache_duplicate")
with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
f1 = exe.submit(sum_with_wait, 1, 1)
f2 = exe.submit(sum_with_wait, 1, 1)
self.assertEqual(f1.result(), f2.result())

def test_cache_error(self):
cache_directory = os.path.abspath("cache_error")
with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
def tearDown(self):
shutil.rmtree("executorlib_cache", ignore_errors=True)
shutil.rmtree("cache_error", ignore_errors=True)
shutil.rmtree("cache_duplicate", ignore_errors=True)
12 changes: 6 additions & 6 deletions tests/unit/task_scheduler/file/test_backend.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
backend_execute_task_in_file(file_name=file_name)
future_obj = Future()
_check_task_output(
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
)
self.assertTrue(future_obj.done())
self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
backend_execute_task_in_file(file_name=file_name_1)
f1 = Future()
_check_task_output(
task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
)
task_key_2, data_dict = serialize_funct(
fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
backend_execute_task_in_file(file_name=file_name_2)
f2 = Future()
_check_task_output(
task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
)
fs1 = FutureSelector(future=f1, selector="a")
fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
backend_execute_task_in_file(file_name=file_name)
future_obj = Future()
_check_task_output(
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
)
self.assertTrue(future_obj.done())
self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
backend_execute_task_in_file(file_name=file_name)
future_obj = Future()
_check_task_output(
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
)
self.assertTrue(future_obj.done())
self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
backend_execute_task_in_file(file_name=file_name)
future_obj = Future()
_check_task_output(
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
)
self.assertTrue(future_obj.done())
with self.assertRaises(ValueError):
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all 
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix");
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms");
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links");
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default");
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers");
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode");
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
[Bug] Ignore duplicate entries in cache#997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

jan-janssen
merged 14 commits into

from

Jun 3, 2026
 Merged


[Bug] Ignore duplicate entries in cache
#997
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


14 commits


Select commit
Hold shift + click to select a range
92cff67

[Bug] Ignore duplicate entries in cache


jan-janssen Jun 3, 2026
4e16887

fix tests


jan-janssen Jun 3, 2026
9d4c0ad

Add test


jan-janssen Jun 3, 2026
8aa864b

fix error


jan-janssen Jun 3, 2026
501f340

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
7fdbe1c

Add duplicate to file based executors


jan-janssen Jun 3, 2026
5d057e8

fix test


jan-janssen Jun 3, 2026
188d480

another fix


jan-janssen Jun 3, 2026
243ffc8

Define duplicate_dict as optional


jan-janssen Jun 3, 2026
ae6e978

[pre-commit.ci] auto fixes from pre-commit.com hooks


pre-commit-ci[bot] Jun 3, 2026
eb1b5e8

fixes


jan-janssen Jun 3, 2026
72f690c

fix coverage


jan-janssen Jun 3, 2026
fe9c8bf

Format black


pyiron-runner Jun 3, 2026
d1701ef

Add docstring


jan-janssen Jun 3, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


10 changes: 6 additions & 4 deletions
src/executorlib/standalone/hdf.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
import contextlib
import os
from concurrent.futures import Future
from time import sleep
Expand DownExpand Up@@ -34,10 +35,11 @@ def dump(file_name: Optional[str], data_dict: dict) -> None:
 with h5py.File(file_name_abs, "a") as fname:
 for data_key, data_value in data_dict.items():
 if data_key in group_dict:
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )
 with contextlib.suppress(ValueError):
 fname.create_dataset(
 name="/" + group_dict[data_key],
 data=np.void(cloudpickle.dumps(data_value)),
 )


def load(file_name: str) -> dict:
Expand Down

53 changes: 50 additions & 3 deletions
src/executorlib/task_scheduler/file/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -91,6 +91,7 @@ def execute_tasks_h5(
 process_dict: dict = {}
 cache_dir_dict: dict = {}
 file_name_dict: dict = {}
 duplicate_dict: dict = {}
 while True:
 task_dict = None
 with contextlib.suppress(queue.Empty):
Expand All@@ -101,6 +102,7 @@ def execute_tasks_h5(
 cancel_futures=task_dict.get("cancel_futures", False),
 memory_dict=memory_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 cache_dir_dict=cache_dir_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
Expand DownExpand Up@@ -175,12 +177,17 @@ def execute_tasks_h5(
 process_dict[task_key] = queue_id
 memory_dict[task_key] = task_dict["future"]
 cache_dir_dict[task_key] = cache_directory
 elif memory_dict[task_key] != task_dict["future"]:
 if task_key not in duplicate_dict:
 duplicate_dict[task_key] = []
 duplicate_dict[task_key].append(task_dict["future"])
 future_queue.task_done()
 else:
 memory_dict = _refresh_memory_dict(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -189,7 +196,10 @@ def execute_tasks_h5(


def _check_task_output(
 task_key: str, future_obj: Future, cache_directory: str
 task_key: str,
 future_obj: Future,
 cache_directory: str,
 duplicate_dict: Optional[dict] = None,
) -> Future:
 """
 Check the output of a task and set the result of the future object if available.
Expand All@@ -198,7 +208,7 @@ def _check_task_output(
 task_key (str): The key of the task.
 future_obj (Future): The future object associated with the task.
 cache_directory (str): The directory where the HDF5 files are stored.

 duplicate_dict (dict): The dictionary mapping task keys to their associated duplicate future objects.
 Returns:
 Future: The updated future object.

Expand All@@ -207,11 +217,40 @@ def _check_task_output(
 if not os.path.exists(file_name):
 return future_obj
 exec_flag, no_error_flag, result = get_output(file_name=file_name)
 _update_future(
 future_obj=future_obj,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 if duplicate_dict is not None and task_key in duplicate_dict:
 for duplicate_future in duplicate_dict[task_key]:
 _update_future(
 future_obj=duplicate_future,
 exec_flag=exec_flag,
 no_error_flag=no_error_flag,
 result=result,
 )
 del duplicate_dict[task_key]
 return future_obj


def _update_future(
 future_obj: Future, exec_flag: bool, no_error_flag: bool, result: Any
) -> None:
 """
 Update the future object with the result of the task execution.

 Args:
 future_obj (Future): The future object to be updated.
 exec_flag (bool): Flag indicating whether the task has been executed.
 no_error_flag (bool): Flag indicating whether the task execution resulted in an error.
 result (Any): The result of the task execution.
 """
 if exec_flag and no_error_flag:
 future_obj.set_result(result)
 elif exec_flag:
 future_obj.set_exception(result)
 return future_obj


def _convert_args_and_kwargs(
Expand DownExpand Up@@ -281,6 +320,7 @@ def _refresh_memory_dict(
 memory_dict: dict,
 cache_dir_dict: dict,
 process_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -293,6 +333,7 @@ def _refresh_memory_dict(
 memory_dict (dict): dictionary with task keys and future objects
 cache_dir_dict (dict): dictionary with task keys and cache directories
 process_dict (dict): dictionary with task keys and process reference.
 duplicate_dict (dict): dictionary with task keys and duplicate future objects.
 terminate_function (callable): The function to terminate the tasks.
 pysqa_config_directory (str): path to the pysqa config directory (only for pysqa based backend).
 backend (str): name of the backend used to spawn tasks.
Expand All@@ -315,6 +356,7 @@ def _refresh_memory_dict(
 task_key=key,
 future_obj=value,
 cache_directory=cache_dir_dict[key],
 duplicate_dict=duplicate_dict,
 )
 for key, value in memory_dict.items()
 if not value.done()
Expand DownExpand Up@@ -400,6 +442,7 @@ def _shutdown_executor(
 memory_dict: dict,
 process_dict: dict,
 cache_dir_dict: dict,
 duplicate_dict: Optional[dict] = None,
 terminate_function: Optional[Callable] = None,
 pysqa_config_directory: Optional[str] = None,
 backend: Optional[str] = None,
Expand All@@ -421,6 +464,7 @@ def _shutdown_executor(
 cancel_futures (bool): Whether to cancel futures that have not yet started.
 memory_dict (dict): Mapping of task keys to their Future objects.
 process_dict (dict): Mapping of task keys to process handles or queue IDs.
 duplicate_dict (dict): Mapping of task keys to lists of duplicate Future objects.
 cache_dir_dict (dict): Mapping of task keys to the cache directory for each task.
 terminate_function (Callable, optional): Function used to terminate running processes.
 pysqa_config_directory (str, optional): Path to the pysqa config directory.
Expand All@@ -433,6 +477,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -447,6 +492,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand All@@ -465,6 +511,7 @@ def _shutdown_executor(
 memory_dict=memory_dict,
 cache_dir_dict=cache_dir_dict,
 process_dict=process_dict,
 duplicate_dict=duplicate_dict,
 terminate_function=terminate_function,
 pysqa_config_directory=pysqa_config_directory,
 backend=backend,
Expand Down

12 changes: 12 additions & 0 deletions
tests/unit/executor/test_api.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -156,6 +156,17 @@ def test_executor_dependency_plot(self):
 self.assertEqual(len(nodes), 4)
 self.assertEqual(len(edges), 4)

 def test_duplicate_futures(self):
 with TestClusterExecutor(cache_directory="cache_dir") as exe:
 cloudpickle_register(ind=1)
 future_1 = exe.submit(add_with_sleep, 1, parameter_2=2)
 future_2 = exe.submit(add_with_sleep, 1, parameter_2=2)
 self.assertFalse(future_1.done())
 self.assertFalse(future_2.done())
 self.assertEqual(future_1.result(), 3)
 self.assertEqual(future_2.result(), 3)
 self.assertEqual(len(os.listdir("cache_dir")), 1)

 def test_shutdown_wait_false_cancel_futures_false(self):
 exe = TestClusterExecutor(cache_directory="shutdown_1_dir")
 cloudpickle_register(ind=1)
Expand DownExpand Up@@ -202,6 +213,7 @@ def test_shutdown_executor_function(self):
 cancel_futures=True,
 memory_dict=memory_dict,
 process_dict={},
 duplicate_dict=None,
 cache_dir_dict={"a": "cache_dir"},
 terminate_function=None,
 pysqa_config_directory=None,
Expand Down

14 changes: 14 additions & 0 deletions
tests/unit/executor/test_single_cache.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
import os
import shutil
import unittest
from time import sleep

from executorlib import SingleNodeExecutor, get_cache_data
from executorlib.standalone.serialize import cloudpickle_register
Expand All@@ -17,6 +18,11 @@ def get_error(a):
 raise ValueError(a)


def sum_with_wait(a, b):
 sleep((a + b) / 10)
 return a + b


class AddClass:
 def __call__(self, a, b):
 return a+b
Expand DownExpand Up@@ -69,6 +75,13 @@ def test_cache_key(self):
 sum([sum(c["input_args"][0]) for c in cache_lst]), sum(result_lst)
 )

 def test_cache_duplicate_function(self):
 cache_directory = os.path.abspath("cache_duplicate")
 with SingleNodeExecutor(hostname_localhost=True, cache_directory=cache_directory) as exe:
 f1 = exe.submit(sum_with_wait, 1, 1)
 f2 = exe.submit(sum_with_wait, 1, 1)
 self.assertEqual(f1.result(), f2.result())

 def test_cache_error(self):
 cache_directory = os.path.abspath("cache_error")
 with SingleNodeExecutor(cache_directory=cache_directory) as exe:
Expand All@@ -93,3 +106,4 @@ def test_cache_error_file(self):
 def tearDown(self):
 shutil.rmtree("executorlib_cache", ignore_errors=True)
 shutil.rmtree("cache_error", ignore_errors=True)
 shutil.rmtree("cache_duplicate", ignore_errors=True)

12 changes: 6 additions & 6 deletions
tests/unit/task_scheduler/file/test_backend.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -50,7 +50,7 @@ def test_execute_function_mixed(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -101,7 +101,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_1)
 f1 = Future()
 _check_task_output(
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory
 task_key=task_key_1, future_obj=f1, cache_directory=cache_directory,
 )
 task_key_2, data_dict = serialize_funct(
 fn=return_list,
Expand All@@ -113,7 +113,7 @@ def test_execute_function_mixed_selector_convert(self):
 backend_execute_task_in_file(file_name=file_name_2)
 f2 = Future()
 _check_task_output(
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory
 task_key=task_key_2, future_obj=f2, cache_directory=cache_directory,
 )
 fs1 = FutureSelector(future=f1, selector="a")
 fs2 = FutureSelector(future=f2, selector=1)
Expand DownExpand Up@@ -143,7 +143,7 @@ def test_execute_function_args(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand All@@ -170,7 +170,7 @@ def test_execute_function_kwargs(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 self.assertEqual(future_obj.result(), 3)
Expand DownExpand Up@@ -198,7 +198,7 @@ def test_execute_function_error(self):
 backend_execute_task_in_file(file_name=file_name)
 future_obj = Future()
 _check_task_output(
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory
 task_key=task_key, future_obj=future_obj, cache_directory=cache_directory,
 )
 self.assertTrue(future_obj.done())
 with self.assertRaises(ValueError):
Expand Down
Loading