wren runs its model on one BLAS thread
threadpoolctl scopes the limit to wren.act, so other numpy calls in the engine keep their threads.
This commit is contained in:
@@ -14,6 +14,7 @@ from collections.abc import Iterable
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from threadpoolctl import threadpool_limits
|
||||
|
||||
from fluksio.flow.messages import DType, MessageSpec
|
||||
from fluksio.flow.nodes.base import Node
|
||||
@@ -306,7 +307,11 @@ class WrenNode(Node):
|
||||
wren = _wren()
|
||||
specs = {spec.port: spec for spec in self.output_ports}
|
||||
house = self._house(wren, specs, presence, outdoor, dark)
|
||||
calls, confidence = wren.act(prompt, house)
|
||||
# One BLAS thread answers in about half the time for a model this
|
||||
# small. Scoped to the call, so numpy elsewhere in the process keeps
|
||||
# its threads.
|
||||
with threadpool_limits(1):
|
||||
calls, confidence = wren.act(prompt, house)
|
||||
confidence = round(float(confidence), 3)
|
||||
write = self.cfg.apply and confidence >= self.cfg.min_confidence
|
||||
writes: dict[str, Any] = {}
|
||||
|
||||
@@ -58,6 +58,10 @@ dependencies = [
|
||||
# The `mlp` node, which is registered at boot. Kept out of the extras
|
||||
# because a data-science environment has it already.
|
||||
"numpy>=2.2.6",
|
||||
# The `wren` node runs its model on one BLAS thread, which answers in about
|
||||
# half the time. Scoped to the call: OPENBLAS_NUM_THREADS would bind every
|
||||
# other numpy call in the engine too, and only counts before numpy loads.
|
||||
"threadpoolctl>=3.5",
|
||||
# The `inject` node's cron. Imported where it is used, behind a message
|
||||
# that cannot fire while this line is here.
|
||||
"croniter>=1.3.0",
|
||||
@@ -164,6 +168,11 @@ ignore_missing_imports = true
|
||||
module = ["wren"]
|
||||
ignore_missing_imports = true
|
||||
|
||||
# What scopes the `wren` node's BLAS threads to its call; it ships no types.
|
||||
[[tool.mypy.overrides]]
|
||||
module = ["threadpoolctl"]
|
||||
ignore_missing_imports = true
|
||||
|
||||
|
||||
[tool.ruff]
|
||||
target-version = "py312"
|
||||
|
||||
@@ -143,3 +143,25 @@ def test_nothing_is_written_unless_applied_and_sure(fake_wren, params, confidenc
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def test_the_model_runs_on_one_blas_thread(fake_wren, monkeypatch):
|
||||
"""About twice as fast for a model this small; numpy is loaded by `mlp`."""
|
||||
from threadpoolctl import threadpool_info
|
||||
|
||||
threads: list[int] = []
|
||||
act = fake_wren.act
|
||||
|
||||
def counted(command: str, house: Any) -> tuple[list[Any], float]:
|
||||
threads.extend(
|
||||
pool["num_threads"]
|
||||
for pool in threadpool_info()
|
||||
if pool["user_api"] == "blas"
|
||||
)
|
||||
return act(command, house)
|
||||
|
||||
monkeypatch.setattr(fake_wren, "act", counted)
|
||||
_node(("kitchen_lamp", DType.BOOL)).execute({"house.command": "lights on"})
|
||||
|
||||
assert threads
|
||||
assert set(threads) == {1}
|
||||
|
||||
Reference in New Issue
Block a user