Use standard thread library#580

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 18 commits into from
Feb 14, 2025
Merged

Use standard thread library #580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
1ce72a5
Use standard thread library
jan-janssen Feb 14, 2025
42dc067
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Feb 14, 2025
c021ba3
Update shared.py
jan-janssen Feb 14, 2025
81e0bbc
Update executor.py
jan-janssen Feb 14, 2025
7353ef5
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Feb 14, 2025
01bd741
clean up
jan-janssen Feb 14, 2025
a5fc401
fixes
jan-janssen Feb 14, 2025
d30c7a2
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Feb 14, 2025
846ec5b
ignore type warnings
jan-janssen Feb 14, 2025
9ed7562
Merge remote-tracking branch 'origin/standard_thread' into standard_t…
jan-janssen Feb 14, 2025
3203223
more test fixes
jan-janssen Feb 14, 2025
3e463da
another fix
jan-janssen Feb 14, 2025
1eec23c
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Feb 14, 2025
1af9f30
more fixes
jan-janssen Feb 14, 2025
21dd5cc
Merge remote-tracking branch 'origin/standard_thread' into standard_t…
jan-janssen Feb 14, 2025
1b0badf
try to fix old version
jan-janssen Feb 14, 2025
2da722e
Fix python 3.10 issue
jan-janssen Feb 14, 2025
d4afda7
Merge branch 'main' into standard_thread
jan-janssen Feb 14, 2025
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
12 changes: 6 additions & 6 deletions executorlib/base/executor.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
cloudpickle_register(ind=3)
self._max_cores = max_cores
self._future_queue: Optional[queue.Queue] = queue.Queue()
self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
self._process: Optional[Union[Thread, list[Thread]]] = None

@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent

🧩 Analysis chain

Verify exception handling behavior with standard Thread.

The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.

🏁 Script executed:

#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py

Length of output: 11463

🏁 Script executed:

#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"

Length of output: 477

Attention: Inconsistent Thread Type References May Affect Exception Propagation

The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.

  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.


@property
def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
Optional[dict]: Information about the executor.
"""
if self._process is not None and isinstance(self._process, list):
meta_data_dict = self._process[0].get_kwargs().copy()
meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
if "future_queue" in meta_data_dict:
del meta_data_dict["future_queue"]
meta_data_dict["max_workers"] = len(self._process)
return meta_data_dict
elif self._process is not None:
meta_data_dict = self._process.get_kwargs().copy()
meta_data_dict = self._process._kwargs.copy() # type: ignore
if "future_queue" in meta_data_dict:
del meta_data_dict["future_queue"]
return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
cancel_items_in_queue(que=self._future_queue)
if self._process is not None and self._future_queue is not None:
self._future_queue.put({"shutdown": True, "wait": wait})
if wait and isinstance(self._process, RaisingThread):
if wait and isinstance(self._process, Thread):
self._process.join()
self._future_queue.join()
self._process = None
self._future_queue = None

def _set_process(self, process: RaisingThread):
def _set_process(self, process: Thread):
"""
Set the process for the executor.

Expand Down
4 changes: 2 additions & 2 deletions executorlib/cache/executor.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
check_max_workers_and_cores,
check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
cache_directory_path = os.path.abspath(cache_directory)
os.makedirs(cache_directory_path, exist_ok=True)
self._set_process(
RaisingThread(
Thread(
target=execute_tasks_h5,
kwargs={
"future_queue": self._future_queue,
Expand Down
6 changes: 4 additions & 2 deletions executorlib/cache/shared.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
]
else:
if len(future_wait_key_lst) > 0:
raise ValueError(
"Future objects are not supported as input if disable_dependencies=True."
task_dict["future"].set_exception(
ValueError(
"Future objects are not supported as input if disable_dependencies=True."
)
)
task_dependent_lst = []
process_dict[task_key] = execute_function(
Expand Down
4 changes: 2 additions & 2 deletions executorlib/interactive/executor.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
generate_nodes_and_edges,
generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
) -> None:
super().__init__(max_cores=max_cores)
self._set_process(
RaisingThread(
Thread(
target=execute_tasks_with_dependencies,
kwargs={
# Executor Arguments
Expand Down
36 changes: 18 additions & 18 deletions executorlib/interactive/shared.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
self._process = None
self._future_queue = None

def _set_process(self, process: list[RaisingThread]): # type: ignore
def _set_process(self, process: list[Thread]): # type: ignore
"""
Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
executor_kwargs["queue_join_on_shutdown"] = False
self._set_process(
process=[
RaisingThread(
Thread(
target=execute_parallel_tasks,
kwargs=executor_kwargs,
)
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
executor_kwargs["max_cores"] = max_cores
executor_kwargs["max_workers"] = max_workers
self._set_process(
RaisingThread(
Thread(
target=execute_separate_tasks,
kwargs=executor_kwargs,
)
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
):
future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
exception_lst = _get_exception_lst(future_lst=future_lst)
if len(exception_lst) > 0:
task_dict["future"].set_exception(exception_lst[0])
elif len(future_lst) == 0 or ready_flag:
# No future objects are used in the input or all future objects are already done
task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
args=task_dict["args"], kwargs=task_dict["kwargs"]
)
executor_queue.put(task_dict)
else: # Otherwise add the function to the wait list
task_dict["future_lst"] = future_lst
wait_lst.append(task_dict)
if not _get_exception(future_obj=task_dict["future"]):
if len(exception_lst) > 0:
task_dict["future"].set_exception(exception_lst[0])
elif len(future_lst) == 0 or ready_flag:
# No future objects are used in the input or all future objects are already done
task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
args=task_dict["args"], kwargs=task_dict["kwargs"]
)
executor_queue.put(task_dict)
else: # Otherwise add the function to the wait list
task_dict["future_lst"] = future_lst
wait_lst.append(task_dict)
future_queue.task_done()
elif len(wait_lst) > 0:
number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
"init_function": None,
}
)
process = RaisingThread(
process = Thread(
target=execute_parallel_tasks,
kwargs=task_kwargs,
)
Expand All@@ -610,14 +611,13 @@ def _execute_task(
future_queue (Queue): Queue for receiving new tasks.
"""
f = task_dict.pop("future")
if f.set_running_or_notify_cancel():
if not f.done() and f.set_running_or_notify_cancel():
try:
f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
except Exception as thread_exception:
interface.shutdown(wait=True)
future_queue.task_done()
f.set_exception(exception=thread_exception)
raise thread_exception
else:
future_queue.task_done()

Expand Down
2 changes: 0 additions & 2 deletions executorlib/standalone/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
"SocketInterface",
Expand All@@ -16,6 +15,5 @@
"interface_send",
"interface_shutdown",
"interface_receive",
"RaisingThread",
"MpiExecSpawner",
]
42 changes: 0 additions & 42 deletions executorlib/standalone/thread.py
View file
Open in desktop

This file was deleted.

11 changes: 6 additions & 5 deletions tests/test_cache_executor_serial.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
execute_in_subprocess,
terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
with FileExecutor(
execute_function=execute_in_subprocess, disable_dependencies=True
) as exe:
exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
fs.result()

def test_executor_working_directory(self):
cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
)
cache_dir = os.path.abspath("cache")
os.makedirs(cache_dir, exist_ok=True)
process = RaisingThread(
process = Thread(
target=execute_tasks_h5,
kwargs={
"future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
)
cache_dir = os.path.abspath("cache")
os.makedirs(cache_dir, exist_ok=True)
process = RaisingThread(
process = Thread(
target=execute_tasks_h5,
kwargs={
"future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
)
cache_dir = os.path.abspath("cache")
os.makedirs(cache_dir, exist_ok=True)
process = RaisingThread(
process = Thread(
target=execute_tasks_h5,
kwargs={
"future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
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
Use standard thread library#580

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 18 commits into

from

Feb 14, 2025
 Merged


Use standard thread library
#580
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


18 commits


Select commit
Hold shift + click to select a range
1ce72a5

Use standard thread library


jan-janssen Feb 14, 2025
42dc067

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


pre-commit-ci[bot] Feb 14, 2025
c021ba3

Update shared.py


jan-janssen Feb 14, 2025
81e0bbc

Update executor.py


jan-janssen Feb 14, 2025
7353ef5

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


pre-commit-ci[bot] Feb 14, 2025
01bd741

clean up


jan-janssen Feb 14, 2025
a5fc401

fixes


jan-janssen Feb 14, 2025
d30c7a2

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


pre-commit-ci[bot] Feb 14, 2025
846ec5b

ignore type warnings


jan-janssen Feb 14, 2025
9ed7562

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
3203223

more test fixes


jan-janssen Feb 14, 2025
3e463da

another fix


jan-janssen Feb 14, 2025
1eec23c

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


pre-commit-ci[bot] Feb 14, 2025
1af9f30

more fixes


jan-janssen Feb 14, 2025
21dd5cc

Merge remote-tracking branch 'origin/standard_thread' into standard_t…


jan-janssen Feb 14, 2025
1b0badf

try to fix old version


jan-janssen Feb 14, 2025
2da722e

Fix python 3.10 issue


jan-janssen Feb 14, 2025
d4afda7

Merge branch 'main' into standard_thread


jan-janssen Feb 14, 2025

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


12 changes: 6 additions & 6 deletions
executorlib/base/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -6,12 +6,12 @@
from concurrent.futures import (
 Future,
)
from threading import Thread
from typing import Callable, Optional, Union

from executorlib.standalone.inputcheck import check_resource_dict
from executorlib.standalone.queue import cancel_items_in_queue
from executorlib.standalone.serialize import cloudpickle_register
from executorlib.standalone.thread import RaisingThread


class ExecutorBase(FutureExecutor):
Expand All@@ -29,7 +29,7 @@ def __init__(self, max_cores: Optional[int] = None):
 cloudpickle_register(ind=3)
 self._max_cores = max_cores
 self._future_queue: Optional[queue.Queue] = queue.Queue()
 self._process: Optional[Union[RaisingThread, list[RaisingThread]]] = None
 self._process: Optional[Union[Thread, list[Thread]]] = None
@coderabbitaicoderabbitaiBotFeb 14, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent
🧩 Analysis chain
Verify exception handling behavior with standard Thread.
The switch from RaisingThread to standard Thread might affect how exceptions are propagated from worker threads to the main thread. The standard Thread class doesn't automatically propagate exceptions to the main thread.
🏁 Script executed:
#!/bin/bash# Search for exception handling patterns in the codebase
rg -A 5 "try:|except:|finally:" --type py
Length of output: 11463
🏁 Script executed:
#!/bin/bash# Search for any remaining references to RaisingThread in the repository
rg "RaisingThread"
Length of output: 477
Attention: Inconsistent Thread Type References May Affect Exception Propagation
The update in executorlib/base/executor.py correctly changes the _process annotation to use the standard Thread. However, several parts of the codebase still reference RaisingThread (e.g., docstrings and method signatures in executorlib/base/executor.py and executorlib/interactive/shared.py). This inconsistency could lead to confusion regarding exception propagation—since the standard Thread does not automatically propagate exceptions to the main thread as RaisingThread might have.
  • Action Items:
    • Ensure that exception handling with the standard Thread is explicitly managed if propagating errors is required.
    • Update all references (docstrings, type annotations, and method signatures) from RaisingThread to Thread (or alternatively, reintroduce RaisingThread functionality where needed) for consistency.

 @property
 def info(self) -> Optional[dict]:
Expand All@@ -40,13 +40,13 @@ def info(self) -> Optional[dict]:
 Optional[dict]: Information about the executor.
 """
 if self._process is not None and isinstance(self._process, list):
 meta_data_dict = self._process[0].get_kwargs().copy()
 meta_data_dict = self._process[0]._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 meta_data_dict["max_workers"] = len(self._process)
 return meta_data_dict
 elif self._process is not None:
 meta_data_dict = self._process.get_kwargs().copy()
 meta_data_dict = self._process._kwargs.copy() # type: ignore
 if "future_queue" in meta_data_dict:
 del meta_data_dict["future_queue"]
 return meta_data_dict
Expand DownExpand Up@@ -138,13 +138,13 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 cancel_items_in_queue(que=self._future_queue)
 if self._process is not None and self._future_queue is not None:
 self._future_queue.put({"shutdown": True, "wait": wait})
 if wait and isinstance(self._process, RaisingThread):
 if wait and isinstance(self._process, Thread):
 self._process.join()
 self._future_queue.join()
 self._process = None
 self._future_queue = None

 def _set_process(self, process: RaisingThread):
 def _set_process(self, process: Thread):
 """
 Set the process for the executor.

Expand Down

4 changes: 2 additions & 2 deletions
executorlib/cache/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
import os
from threading import Thread
from typing import Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -15,7 +16,6 @@
 check_max_workers_and_cores,
 check_nested_flux_executor,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.queue_spawner import execute_with_pysqa
Expand DownExpand Up@@ -64,7 +64,7 @@ def __init__(
 cache_directory_path = os.path.abspath(cache_directory)
 os.makedirs(cache_directory_path, exist_ok=True)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": self._future_queue,
Expand Down

6 changes: 4 additions & 2 deletions
executorlib/cache/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,8 +115,10 @@ def execute_tasks_h5(
 ]
 else:
 if len(future_wait_key_lst) > 0:
 raise ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 task_dict["future"].set_exception(
 ValueError(
 "Future objects are not supported as input if disable_dependencies=True."
 )
 )
 task_dependent_lst = []
 process_dict[task_key] = execute_function(
Expand Down

4 changes: 2 additions & 2 deletions
executorlib/interactive/executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
from concurrent.futures import Future
from threading import Thread
from typing import Any, Callable, Optional

from executorlib.base.executor import ExecutorBase
Expand All@@ -8,7 +9,6 @@
 generate_nodes_and_edges,
 generate_task_hash,
)
from executorlib.standalone.thread import RaisingThread


class ExecutorWithDependencies(ExecutorBase):
Expand DownExpand Up@@ -41,7 +41,7 @@ def __init__(
 ) -> None:
 super().__init__(max_cores=max_cores)
 self._set_process(
RaisingThread(
Thread(
 target=execute_tasks_with_dependencies,
 kwargs={
 # Executor Arguments
Expand Down

36 changes: 18 additions & 18 deletions
executorlib/interactive/shared.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -5,6 +5,7 @@
import time
from asyncio.exceptions import CancelledError
from concurrent.futures import Future, TimeoutError
from threading import Thread
from time import sleep
from typing import Any, Callable, Optional, Union

Expand All@@ -20,7 +21,6 @@
)
from executorlib.standalone.interactive.spawner import BaseSpawner, MpiExecSpawner
from executorlib.standalone.serialize import serialize_funct_h5
from executorlib.standalone.thread import RaisingThread


class ExecutorBroker(ExecutorBase):
Expand DownExpand Up@@ -89,7 +89,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
 self._process = None
 self._future_queue = None

 def _set_process(self, process: list[RaisingThread]): # type: ignore
 def _set_process(self, process: list[Thread]): # type: ignore
 """
 Set the process for the executor.

Expand DownExpand Up@@ -149,7 +149,7 @@ def __init__(
 executor_kwargs["queue_join_on_shutdown"] = False
 self._set_process(
 process=[
RaisingThread(
Thread(
 target=execute_parallel_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -205,7 +205,7 @@ def __init__(
 executor_kwargs["max_cores"] = max_cores
 executor_kwargs["max_workers"] = max_workers
 self._set_process(
RaisingThread(
Thread(
 target=execute_separate_tasks,
 kwargs=executor_kwargs,
 )
Expand DownExpand Up@@ -363,17 +363,18 @@ def execute_tasks_with_dependencies(
 ):
 future_lst, ready_flag = _get_future_objects_from_input(task_dict=task_dict)
 exception_lst = _get_exception_lst(future_lst=future_lst)
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 if not _get_exception(future_obj=task_dict["future"]):
 if len(exception_lst) > 0:
 task_dict["future"].set_exception(exception_lst[0])
 elif len(future_lst) == 0 or ready_flag:
 # No future objects are used in the input or all future objects are already done
 task_dict["args"], task_dict["kwargs"] = _update_futures_in_input(
 args=task_dict["args"], kwargs=task_dict["kwargs"]
 )
 executor_queue.put(task_dict)
 else: # Otherwise add the function to the wait list
 task_dict["future_lst"] = future_lst
 wait_lst.append(task_dict)
 future_queue.task_done()
 elif len(wait_lst) > 0:
 number_waiting = len(wait_lst)
Expand DownExpand Up@@ -589,7 +590,7 @@ def _submit_function_to_separate_process(
 "init_function": None,
 }
 )
 process = RaisingThread(
 process = Thread(
 target=execute_parallel_tasks,
 kwargs=task_kwargs,
 )
Expand All@@ -610,14 +611,13 @@ def _execute_task(
 future_queue (Queue): Queue for receiving new tasks.
 """
 f = task_dict.pop("future")
 if f.set_running_or_notify_cancel():
 if not f.done() and f.set_running_or_notify_cancel():
 try:
 f.set_result(interface.send_and_receive_dict(input_dict=task_dict))
 except Exception as thread_exception:
 interface.shutdown(wait=True)
 future_queue.task_done()
 f.set_exception(exception=thread_exception)
 raise thread_exception
 else:
 future_queue.task_done()

Expand Down

2 changes: 0 additions & 2 deletions
executorlib/standalone/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,6 @@
 interface_shutdown,
)
from executorlib.standalone.interactive.spawner import MpiExecSpawner
from executorlib.standalone.thread import RaisingThread

__all__ = [
 "SocketInterface",
Expand All@@ -16,6 +15,5 @@
 "interface_send",
 "interface_shutdown",
 "interface_receive",
 "RaisingThread",
 "MpiExecSpawner",
]

42 changes: 0 additions & 42 deletions
executorlib/standalone/thread.py

View file


Open in desktop


This file was deleted.


11 changes: 6 additions & 5 deletions
tests/test_cache_executor_serial.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -3,12 +3,12 @@
from queue import Queue
import shutil
import unittest
from threading import Thread

from executorlib.cache.subprocess_spawner import (
 execute_in_subprocess,
 terminate_subprocess,
)
from executorlib.standalone.thread import RaisingThread

try:
 from executorlib.cache.executor import FileExecutor, create_file_executor
Expand DownExpand Up@@ -57,7 +57,8 @@ def test_executor_dependence_error(self):
 with FileExecutor(
 execute_function=execute_in_subprocess, disable_dependencies=True
 ) as exe:
 exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs = exe.submit(my_funct, 1, b=exe.submit(my_funct, 1, b=2))
 fs.result()

 def test_executor_working_directory(self):
 cwd = os.path.join(os.path.dirname(__file__), "executables")
Expand All@@ -81,7 +82,7 @@ def test_executor_function(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -122,7 +123,7 @@ def test_executor_function_dependence_kwargs(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand DownExpand Up@@ -163,7 +164,7 @@ def test_executor_function_dependence_args(self):
 )
 cache_dir = os.path.abspath("cache")
 os.makedirs(cache_dir, exist_ok=True)
 process = RaisingThread(
 process = Thread(
 target=execute_tasks_h5,
 kwargs={
 "future_queue": q,
Expand Down
Loading