feat(ml): improve model loading (#31742)

* use recent mimalloc

* jit caching

* bump onnxruntime-gpu

* explicit memory release

* refactor in-memory cache

* configure hf org and revision

* refactor model load

* move orjson response

* disable direct submission for i915

* flush denormals

* update tests and lock

* disable suboptimal fusions

* respect resize_mode

* increase threads for clip text

machine-learning/immich_ml/__main__.py

@@ -13,6 +13,11 @@ else:
 
 module_dir = Path(__file__).parent
 
+# on i915, Intel's runtime submits every allocation in the process with each run it gives OpenVINO's GPU plugin
+if any(driver.resolve().name == "i915" for driver in Path("/sys/class/drm").glob("renderD*/device/driver")):
+    os.environ.setdefault("NEOReadDebugKeys", "1")
+    os.environ.setdefault("EnableDirectSubmission", "0")
+
 
 def is_ipv6(host: str) -> bool:
     try:

machine-learning/immich_ml/allocator.py

@@ -0,0 +1,10 @@
+import ctypes
+
+_process = ctypes.CDLL(None)
+
+
+def release() -> None:
+    if hasattr(_process, "mi_collect"):
+        _process.mi_collect(True)
+    elif hasattr(_process, "malloc_trim"):
+        _process.malloc_trim(0)

machine-learning/immich_ml/config.py

@@ -13,7 +13,7 @@ from rich.logging import RichHandler
 from uvicorn import Server
 from uvicorn.workers import UvicornWorker
 
-from .schemas import ModelPrecision
+from .schemas import ModelOrganization, ModelPrecision
 
 
 class ClipSettings(BaseModel):
@@ -74,6 +74,12 @@ class Settings(BaseSettings):
     max_batch_size: MaxBatchSize | None = None
     openvino_precision: ModelPrecision = ModelPrecision.FP32
     rocm_precision: ModelPrecision = ModelPrecision.FP32
+    model_organization: ModelOrganization = ModelOrganization.APP
+    model_revision: str = "main"
+
+    @property
+    def legacy_models(self) -> bool:
+        return self.model_revision == "main"
 
     @property
     def device_id(self) -> str:

machine-learning/immich_ml/main.py

@@ -2,25 +2,26 @@ import asyncio
 import gc
 import os
 import signal
-import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from functools import partial
-from typing import Any, AsyncGenerator, Callable, Iterator
+from typing import Any, AsyncGenerator, Callable
 from zipfile import BadZipFile
 
 import orjson
 from fastapi import Depends, FastAPI, File, Form, HTTPException
-from fastapi.responses import PlainTextResponse
+from fastapi.responses import JSONResponse, PlainTextResponse
 from onnxruntime.capi.onnxruntime_pybind11_state import InvalidProtobuf, NoSuchFile
 from PIL.Image import Image
 from pydantic import ValidationError
 from starlette.formparsers import MultiPartParser
 
+from immich_ml import allocator
 from immich_ml.models import get_model_deps
 from immich_ml.models.base import InferenceModel
 from immich_ml.models.transforms import decode_pil
+from immich_ml.sessions.ort import flush_denormals
 
 from .config import PreloadModelData, log, settings
 from .models.cache import ModelCache
@@ -32,18 +33,24 @@ from .schemas import (
     ModelIdentity,
     ModelTask,
     ModelType,
-    ORJSONResponse,
     PipelineRequest,
     T,
 )
 
+
+class ORJSONResponse(JSONResponse):
+    def render(self, content: Any) -> bytes:
+        return orjson.dumps(content, option=orjson.OPT_SERIALIZE_NUMPY)
+
+
 MultiPartParser.spool_max_size = 2**26  # spools to disk if payload is 64 MiB or larger
 
-model_cache = ModelCache(revalidate=settings.model_ttl > 0)
+model_cache = ModelCache()
 thread_pool: ThreadPoolExecutor | None = None
-lock = threading.Lock()
+MODEL_FILE_ERRORS = (InvalidProtobuf, NoSuchFile)
 active_requests = 0
 last_called: float | None = None
+release: asyncio.TimerHandle | None = None
 
 
 @asynccontextmanager
@@ -57,19 +64,20 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     )
 
     try:
+        flush_denormals()  # for work that runs on the event loop's own thread
         if settings.request_threads > 0:
             # asyncio is a huge bottleneck for performance, so we use a thread pool to run blocking code
-            thread_pool = ThreadPoolExecutor(settings.request_threads) if settings.request_threads > 0 else None
+            thread_pool = ThreadPoolExecutor(settings.request_threads, initializer=flush_denormals)
             log.info(f"Initialized request thread pool with {settings.request_threads} threads.")
         if settings.model_ttl > 0 and settings.model_ttl_poll_s > 0:
             asyncio.ensure_future(idle_shutdown_task())
         if settings.preload is not None:
             await preload_models(settings.preload)
+            allocator.release()
         yield
     finally:
         log.handlers.clear()
-        for model in model_cache.cache._cache.values():
-            del model
+        model_cache.clear()
         if thread_pool is not None:
             thread_pool.shutdown()
         gc.collect()
@@ -78,10 +86,10 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
 async def preload_models(preload: PreloadModelData) -> None:
     log.info(f"Preloading models: clip:{preload.clip} facial_recognition:{preload.facial_recognition}")
 
-    async def load_models(model_string: str, model_type: ModelType, model_task: ModelTask) -> None:
+    async def load_models(model_string: str, model_type: ModelType, model_task: ModelTask, **options: Any) -> None:
         for model_name in model_string.split(","):
             model_name = model_name.strip()
-            model = await model_cache.get(model_name, model_type, model_task)
+            model = model_cache.get(model_name, model_type, model_task, **options)
             await load(model)
 
     if preload.clip.textual is not None:
@@ -119,14 +127,19 @@ async def preload_models(preload: PreloadModelData) -> None:
         )
 
 
-def update_state() -> Iterator[None]:
-    global active_requests, last_called
+async def update_state() -> AsyncGenerator[None, None]:
+    global active_requests, last_called, release
     active_requests += 1
     last_called = time.time()
+    if release is not None:
+        release.cancel()
+        release = None
     try:
         yield
     finally:
         active_requests -= 1
+        if not active_requests:
+            release = asyncio.get_running_loop().call_later(5, allocator.release)
 
 
 def get_entries(entries: str = Form()) -> InferenceEntries:
@@ -187,9 +200,7 @@ async def run_inference(payload: Image | str, entries: InferenceEntries) -> Infe
     response: InferenceResponse = {}
 
     async def _run_inference(entry: InferenceEntry) -> None:
-        model = await model_cache.get(
-            entry["name"], entry["type"], entry["task"], ttl=settings.model_ttl, **entry["options"]
-        )
+        model = model_cache.get(entry["name"], entry["type"], entry["task"], ttl=settings.model_ttl, **entry["options"])
         inputs = [payload]
         for dep in model.depends:
             try:
@@ -198,14 +209,15 @@ async def run_inference(payload: Image | str, entries: InferenceEntries) -> Infe
                 message = f"Task {entry['task']} of type {entry['type']} depends on output of {dep}"
                 raise HTTPException(400, message)
         model = await load(model)
-        output = await run(model.predict, *inputs, **entry["options"])
+        output = await attempt(model, partial(model.predict, *inputs, **entry["options"]), MODEL_FILE_ERRORS)
         outputs[model.identity] = output
         response[entry["task"]] = output
 
-    without_deps, with_deps = entries
-    await asyncio.gather(*[_run_inference(entry) for entry in without_deps])
-    if with_deps:
-        await asyncio.gather(*[_run_inference(entry) for entry in with_deps])
+    for stage in entries:  # those that depend on another's output run after it
+        results = await asyncio.gather(*[_run_inference(entry) for entry in stage], return_exceptions=True)
+        for result in results:
+            if isinstance(result, BaseException):
+                raise result
     if isinstance(payload, Image):
         response["imageHeight"], response["imageWidth"] = payload.height, payload.width
 
@@ -220,42 +232,40 @@ async def run(func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
 
 
 async def load(model: InferenceModel) -> InferenceModel:
-    if model.loaded:
-        return model
+    if not model.loaded:
+        await attempt(model, model.load, (OSError, BadZipFile, *MODEL_FILE_ERRORS))
+    return model
 
-    def _load(model: InferenceModel) -> InferenceModel:
-        if model.load_attempts > 1:
+
+async def attempt(model: InferenceModel, func: Callable[[], T], corrupt: tuple[type[Exception], ...]) -> T:
+    def _attempt() -> T:
+        if not model.loaded and model.load_attempts > 1:
             raise HTTPException(500, f"Failed to load model '{model.model_name}'")
-        with lock:
-            try:
-                model.load()
-            except FileNotFoundError as e:
-                if model.model_format == ModelFormat.ONNX:
-                    raise e
-                log.warning(
-                    f"{model.model_format.upper()} is available, but model '{model.model_name}' does not support it.",
-                    exc_info=e,
-                )
-                model.model_format = ModelFormat.ONNX
-                model.load()
-        return model
+        try:
+            return func()
+        except FileNotFoundError as e:
+            if model.model_format == ModelFormat.ONNX:
+                raise e
+            log.warning(
+                f"{model.model_format.upper()} is available, but model '{model.model_name}' does not support it.",
+                exc_info=e,
+            )
+            model.unload()
+            model.model_format = ModelFormat.ONNX
+            return func()
 
     try:
-        return await run(_load, model)
-    except (OSError, InvalidProtobuf, BadZipFile, NoSuchFile):
+        return await run(_attempt)
+    except corrupt:
         log.warning(f"Failed to load {model.model_type.replace('_', ' ')} model '{model.model_name}'. Clearing cache.")
+        model.unload()
         model.clear_cache()
-        return await run(_load, model)
+        return await run(_attempt)
 
 
 async def idle_shutdown_task() -> None:
     while True:
-        if (
-            last_called is not None
-            and not active_requests
-            and not lock.locked()
-            and time.time() - last_called > settings.model_ttl
-        ):
+        if last_called is not None and not active_requests and time.time() - last_called > settings.model_ttl:
             log.info("Shutting down due to inactivity.")
             os.kill(os.getpid(), signal.SIGINT)
             break

machine-learning/immich_ml/models/__init__.py

@@ -1,5 +1,3 @@
-from typing import Any
-
 from immich_ml.models.base import InferenceModel
 from immich_ml.models.clip.textual import MClipTextualEncoder, OpenClipTextualEncoder
 from immich_ml.models.clip.visual import OpenClipVisualEncoder
@@ -40,9 +38,5 @@ def get_model_class(model_name: str, model_type: ModelType, model_task: ModelTas
             raise ValueError(f"Unknown model combination: {source}, {model_type}, {model_task}")
 
 
-def from_model_type(model_name: str, model_type: ModelType, model_task: ModelTask, **kwargs: Any) -> InferenceModel:
-    return get_model_class(model_name, model_type, model_task)(model_name, **kwargs)
-
-
 def get_model_deps(model_name: str, model_type: ModelType, model_task: ModelTask) -> list[tuple[ModelType, ModelTask]]:
     return get_model_class(model_name, model_type, model_task).depends

machine-learning/immich_ml/models/base.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 from pathlib import Path
 from shutil import rmtree
+from threading import Lock
 from typing import Any, ClassVar
 
 from huggingface_hub import snapshot_download
@@ -15,10 +16,17 @@ from ..config import clean_name, log, settings
 from ..schemas import ModelFormat, ModelIdentity, ModelSession, ModelTask, ModelType
 from ..sessions.ann import AnnSession
 
+_IGNORED_PATTERNS: dict[ModelFormat, list[str]] = {
+    ModelFormat.ONNX: ["*.armnn", "*.rknn"],
+    ModelFormat.ARMNN: ["*.rknn"],
+    ModelFormat.RKNN: ["*.armnn"],
+}
+
 
 class InferenceModel(ABC):
     depends: ClassVar[list[ModelIdentity]]
     identity: ClassVar[ModelIdentity]
+    threads: ClassVar[int] = 2  # how many threads one run takes for CPU inference
 
     def __init__(
         self,
@@ -30,6 +38,7 @@ class InferenceModel(ABC):
     ) -> None:
         self.loaded = session is not None
         self.load_attempts = 0
+        self._load_lock = Lock()
         self.model_name = clean_name(model_name)
         self.cache_dir = Path(cache_dir) if cache_dir is not None else self._cache_dir_default
         self.model_format = model_format if model_format is not None else self._model_format_default
@@ -37,50 +46,47 @@ class InferenceModel(ABC):
             self.session = session
 
     def download(self) -> None:
-        if not self.cached:
-            model_type = self.model_type.replace("-", " ")
-            log.info(f"Downloading {model_type} model '{self.model_name}' to {self.model_path}. This may take a while.")
-            self._download()
+        if self.cached:
+            return
+        model_type = self.model_type.replace("-", " ")
+        log.info(f"Downloading {model_type} model '{self.model_name}' to {self.model_dir}. This may take a while.")
+        snapshot_download(
+            f"{settings.model_organization}/{self.model_name}",
+            revision=settings.model_revision,
+            cache_dir=self.cache_dir,
+            local_dir=self.cache_dir,
+            ignore_patterns=_IGNORED_PATTERNS.get(self.model_format, []),
+        )
+        if not self.cached:  # the repository has nothing in this format, which is what sends a model back to ONNX
+            raise FileNotFoundError(f"Model file not found: {self.model_path}")
 
     def load(self) -> None:
-        if self.loaded:
-            return
-        self.load_attempts += 1
+        with self._load_lock:  # two requests may arrive for a model neither has loaded
+            if self.loaded:
+                return
+            self.load_attempts += 1
 
-        self.download()
-        attempt = f"Attempt #{self.load_attempts} to load" if self.load_attempts > 1 else "Loading"
-        log.info(f"{attempt} {self.model_type.replace('-', ' ')} model '{self.model_name}' to memory")
-        self.session = self._load()
-        self.loaded = True
+            self.download()
+            attempt = f"Attempt #{self.load_attempts} to load" if self.load_attempts > 1 else "Loading"
+            log.info(f"{attempt} {self.model_type.replace('-', ' ')} model '{self.model_name}' to memory")
+            self.session = self._load()
+            self.loaded = True
+
+    def unload(self) -> None:
+        with self._load_lock:
+            if self.loaded:
+                del self.session
+                self.loaded = False
 
     def predict(self, *inputs: Any, **model_kwargs: Any) -> Any:
         self.load()
-        if model_kwargs:
-            self.configure(**model_kwargs)
         return self._predict(*inputs, **model_kwargs)
 
     @abstractmethod
     def _predict(self, *inputs: Any, **model_kwargs: Any) -> Any: ...
 
-    def configure(self, **kwargs: Any) -> None:
-        pass
-
-    def _download(self) -> None:
-        ignored_patterns: dict[ModelFormat, list[str]] = {
-            ModelFormat.ONNX: ["*.armnn", "*.rknn"],
-            ModelFormat.ARMNN: ["*.rknn"],
-            ModelFormat.RKNN: ["*.armnn"],
-        }
-
-        snapshot_download(
-            f"immich-app/{clean_name(self.model_name)}",
-            cache_dir=self.cache_dir,
-            local_dir=self.cache_dir,
-            ignore_patterns=ignored_patterns.get(self.model_format, []),
-        )
-
     def _load(self) -> ModelSession:
-        return self._make_session(self.model_path)
+        return self._make_session()
 
     def clear_cache(self) -> None:
         if not self.cache_dir.exists():
@@ -104,20 +110,14 @@ class InferenceModel(ABC):
             self.cache_dir.unlink()
         self.cache_dir.mkdir(parents=True, exist_ok=True)
 
-    def _make_session(self, model_path: Path) -> ModelSession:
-        if not model_path.is_file():
-            raise FileNotFoundError(f"Model file not found: {model_path}")
-
-        match model_path.suffix:
-            case ".armnn":
-                session: ModelSession = AnnSession(model_path)
-            case ".onnx":
-                session = OrtSession(model_path)
-            case ".rknn":
-                session = rknn.RknnSession(model_path)
-            case _:
-                raise ValueError(f"Unsupported model file type: {model_path.suffix}")
-        return session
+    def _make_session(self) -> ModelSession:
+        match self.model_format:
+            case ModelFormat.ARMNN:
+                return AnnSession(self.model_path)
+            case ModelFormat.ONNX:
+                return OrtSession(self.model_path, threads=self.threads)
+            case ModelFormat.RKNN:
+                return rknn.RknnSession(self.model_path)
 
     def model_path_for_format(self, model_format: ModelFormat) -> Path:
         model_path_prefix = rknn.model_prefix if model_format == ModelFormat.RKNN else None

machine-learning/immich_ml/models/cache.py

@@ -1,60 +1,43 @@
+import asyncio
 from typing import Any
 
-from aiocache.backends.memory import SimpleMemoryCache
-from aiocache.lock import OptimisticLock
-from aiocache.plugins import TimingPlugin
-
-from immich_ml.models import from_model_type
+from immich_ml import allocator
+from immich_ml.models import get_model_class
 from immich_ml.models.base import InferenceModel
 
-from ..schemas import ModelTask, ModelType, has_profiling
+from ..schemas import ModelTask, ModelType
 
 
 class ModelCache:
-    """Fetches a model from an in-memory cache, instantiating it if it's missing."""
+    def __init__(self) -> None:
+        self._models: dict[tuple[Any, ...], InferenceModel] = {}
+        self._expiries: dict[tuple[Any, ...], asyncio.TimerHandle] = {}
 
-    def __init__(
-        self,
-        revalidate: bool = False,
-        timeout: int | None = None,
-        profiling: bool = False,
-    ) -> None:
-        """
-        Args:
-            revalidate: Resets TTL on cache hit. Useful to keep models in memory while active. Defaults to False.
-            timeout: Maximum allowed time for model to load. Disabled if None. Defaults to None.
-            profiling: Collects metrics for cache operations, adding slight overhead. Defaults to False.
-        """
-
-        plugins = []
-
-        if profiling:
-            plugins.append(TimingPlugin())
-
-        self.should_revalidate = revalidate
-
-        self.cache = SimpleMemoryCache(timeout=timeout, plugins=plugins, namespace=None)
-
-    async def get(
-        self, model_name: str, model_type: ModelType, model_task: ModelTask, **model_kwargs: Any
+    def get(
+        self, model_name: str, model_type: ModelType, model_task: ModelTask, ttl: int | None = None, **options: Any
     ) -> InferenceModel:
-        key = f"{model_name}{model_type}{model_task}"
+        model_cls = get_model_class(model_name, model_type, model_task)
+        key = (model_name, model_type, model_task)
+        if key not in self._models:
+            self._models[key] = model_cls(model_name, **options)
+        elif key not in self._expiries:
+            ttl = None  # a model that was preloaded stays
+        if ttl:
+            if key in self._expiries:
+                self._expiries[key].cancel()
+            self._expiries[key] = asyncio.get_running_loop().call_later(ttl, self._evict, key)
+        return self._models[key]
 
-        async with OptimisticLock(self.cache, key) as lock:
-            model: InferenceModel | None = await self.cache.get(key)
-            if model is None:
-                model = from_model_type(model_name, model_type, model_task, **model_kwargs)
-                await lock.cas(model, ttl=model_kwargs.get("ttl", None))
-            elif self.should_revalidate:
-                await self.revalidate(key, model_kwargs.get("ttl", None))
-        return model
+    def clear(self) -> None:
+        if not self._models:
+            return
+        self._models.clear()
+        for key in list(self._expiries.keys()):
+            self._expiries.pop(key).cancel()
+        allocator.release()
 
-    async def get_profiling(self) -> dict[str, float] | None:
-        if not has_profiling(self.cache):
-            return None
-
-        return self.cache.profiling
-
-    async def revalidate(self, key: str, ttl: int | None) -> None:
-        if ttl is not None and key in self.cache._handlers:
-            await self.cache.expire(key, ttl)
+    def _evict(self, key: tuple[Any, ...]) -> None:
+        if key not in self._models:
+            return
+        del self._models[key], self._expiries[key]
+        allocator.release()

machine-learning/immich_ml/models/clip/textual.py

@@ -18,6 +18,7 @@ from immich_ml.schemas import ModelSession, ModelTask, ModelType
 class BaseCLIPTextualEncoder(InferenceModel):
     depends = []
     identity = (ModelType.TEXTUAL, ModelTask.SEARCH)
+    threads = 4  # lower search latency
 
     def _predict(self, inputs: str, language: str | None = None) -> str:
         tokens = self.tokenize(inputs, language=language)

machine-learning/immich_ml/models/clip/visual.py

@@ -70,8 +70,16 @@ class OpenClipVisualEncoder(BaseCLIPVisualEncoder):
         return super()._load()
 
     def transform(self, image: Image.Image) -> dict[str, NDArray[np.float32]]:
-        image = resize_pil(image, self.size)
-        image = crop_pil(image, self.size)
+        image = self._resize(image)
         image_np = to_numpy(image)
         image_np = normalize(image_np, self.mean, self.std)
         return {"image": np.expand_dims(image_np.transpose(2, 0, 1), 0)}
+
+    def _resize(self, image: Image.Image) -> Image.Image:
+        match self.preprocess_cfg.get("resize_mode", "shortest"):  # open_clip's default for older configs
+            case "squash":
+                return image.resize((self.size, self.size), resample=self.resampling)
+            case "shortest":
+                return crop_pil(resize_pil(image, self.size, self.resampling), self.size)
+            case mode:
+                raise ValueError(f"Unsupported resize_mode {mode!r} in {self.preprocess_cfg_path}")

machine-learning/immich_ml/models/facial_recognition/recognition.py

@@ -33,10 +33,10 @@ class FaceRecognizer(InferenceModel):
         self.batch_size = max_batch_size if max_batch_size else self._batch_size_default
 
     def _load(self) -> ModelSession:
-        session = self._make_session(self.model_path)
+        session = self._make_session()
         if (not self.batch_size or self.batch_size > 1) and str(session.get_inputs()[0].shape[0]) != "batch":
             self._add_batch_axis(self.model_path)
-            session = self._make_session(self.model_path)
+            session = self._make_session()
         return session
 
     def _predict(

machine-learning/immich_ml/schemas.py

@@ -1,19 +1,12 @@
 from collections.abc import Sequence
 from enum import Enum
-from typing import Any, Literal, Protocol, TypeGuard, TypeVar
+from typing import Any, Literal, Protocol, TypeVar
 
 import numpy as np
 import numpy.typing as npt
-import orjson
-from fastapi.responses import JSONResponse
 from typing_extensions import TypedDict
 
 
-class ORJSONResponse(JSONResponse):
-    def render(self, content: Any) -> bytes:
-        return orjson.dumps(content, option=orjson.OPT_SERIALIZE_NUMPY)
-
-
 class StrEnum(str, Enum):
     __str__ = str.__str__
 
@@ -56,6 +49,11 @@ class ModelPrecision(StrEnum):
     FP32 = "FP32"
 
 
+class ModelOrganization(StrEnum):
+    APP = "immich-app"
+    TESTING = "immich-testing"
+
+
 ModelIdentity = tuple[ModelType, ModelTask]
 
 
@@ -84,10 +82,6 @@ class ModelSession(Protocol):
     def get_metadata(self) -> dict[str, str]: ...
 
 
-class HasProfiling(Protocol):
-    profiling: dict[str, float]
-
-
 class FaceDetectionOutput(TypedDict):
     boxes: npt.NDArray[np.float32]
     scores: npt.NDArray[np.float32]
@@ -124,8 +118,4 @@ InferenceEntries = tuple[list[InferenceEntry], list[InferenceEntry]]
 InferenceResponse = dict[ModelTask | Literal["imageHeight"] | Literal["imageWidth"], Any]
 
 
-def has_profiling(obj: Any) -> TypeGuard[HasProfiling]:
-    return hasattr(obj, "profiling") and isinstance(obj.profiling, dict)
-
-
 T = TypeVar("T")

machine-learning/immich_ml/sessions/ort.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import ctypes
+import platform
+import sys
 from collections.abc import Sequence
 from pathlib import Path
 from threading import Lock
@@ -53,9 +56,13 @@ class OrtSession:
         providers: list[str] | None = None,
         provider_options: list[dict[str, Any]] | None = None,
         sess_options: ort.SessionOptions | None = None,
+        threads: int = 2,
     ):
         self.model_path = Path(model_path)
+        self.threads = threads
         self.providers = providers if providers is not None else self._providers_default
+        self.disabled_optimizers = _disabled_optimizers_default(self.providers)
+        log.debug(f"Setting disabled_optimizers to {self.disabled_optimizers}")
         self.provider_options = provider_options if provider_options is not None else self._provider_options_default
         self.sess_options = sess_options if sess_options is not None else self._sess_options_default
         self.session = ort.InferenceSession(
@@ -63,6 +70,7 @@ class OrtSession:
             providers=self.providers,
             provider_options=self.provider_options,
             sess_options=self.sess_options,
+            disabled_optimizers=self.disabled_optimizers,
         )
 
     def get_inputs(self) -> Sequence[SessionNode]:
@@ -184,6 +192,8 @@ class OrtSession:
     @property
     def _sess_options_default(self) -> ort.SessionOptions:
         sess_options = ort.SessionOptions()
+        # some CPUs slow down many times over on subnormal operands, and ORT clears the flush in its threads otherwise
+        sess_options.add_session_config_entry("session.set_denormal_as_zero", "1")
         sess_options.enable_cpu_mem_arena = settings.model_arena
 
         # avoid thread contention between models
@@ -198,9 +208,32 @@ class OrtSession:
         if settings.model_intra_op_threads > 0:
             sess_options.intra_op_num_threads = settings.model_intra_op_threads
         elif settings.model_intra_op_threads == 0 and self.providers == ["CPUExecutionProvider"]:
-            sess_options.intra_op_num_threads = 2
+            sess_options.intra_op_num_threads = self.threads
 
         if sess_options.inter_op_num_threads > 1:
             sess_options.execution_mode = ort.ExecutionMode.ORT_PARALLEL
 
         return sess_options
+
+
+def _disabled_optimizers_default(providers: list[str]) -> list[str]:
+    disabled_optimizers: list[str] = []
+    if platform.machine() in ("arm64", "aarch64"):  # as macOS and Linux name the same architecture
+        disabled_optimizers.append("ConvAddActivationFusion")
+
+    # the Gemm it makes runs slower than the pair it replaces there, and on CUDA it also keeps BiasGelu from fusing
+    if "CoreMLExecutionProvider" in providers or "CUDAExecutionProvider" in providers:
+        disabled_optimizers.append("MatMulAddFusion")
+
+    return disabled_optimizers
+
+
+def flush_denormals() -> None:
+    """Reads subnormal floats as zero on the calling thread, which ORT does only on its own threads."""
+    if sys.platform != "linux" or platform.machine() != "x86_64":
+        return
+    libm = ctypes.CDLL("libm.so.6")
+    env = (ctypes.c_uint32 * 8)()  # glibc's fenv_t here: the x87 environment, then MXCSR
+    libm.fegetenv(env)
+    env[7] |= 0x8040  # denormals are zero, flush to zero
+    libm.fesetenv(env)