mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Memoize run-form history option queries on the request context (issue #22927)
Building the workflow Run form renders every step in one
GET /api/workflows/{id}/download?style=run request. Each data /
data_collection parameter independently re-issued the same paginated history
SQL (History.paginated_active_visible_datasets /
paginated_active_dataset_collections) in both get_initial_value and to_dict,
turning a single history scan into O(number of parameters) DB round-trips --
slow even on an empty history, and scaling with workflow size.
Memoize the (filter, page) -> (rows, total) results by signature. Tool.to_json
builds a fresh proxy_work_context_for_history per step, so the cache lives on
the request context's short-term cache (ProvidesHistoryContext.set_cache_value)
and proxy_work_context_for_history now shares the parent transaction's
_short_term_cache with each per-step proxy. That keeps the cache request-scoped
and shared across every step of a build without attaching state to the
SQLAlchemy history instance. Cache keys include the history id because a proxy
may target a history other than the parent's.
Benchmark (50-step workflow, empty history, 3 builds), vs unpatched 26.1:
_workflow_to_dict_run ~9100ms -> ~750ms (~13x faster)
SQL executions 2478 -> 180 (-93%)
paginated history scans 1164 -> 15
This commit is contained in:
@@ -5,6 +5,7 @@ Mock infrastructure for testing ModelManagers.
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from collections.abc import Hashable
|
||||
from typing import (
|
||||
Any,
|
||||
cast,
|
||||
@@ -335,6 +336,7 @@ class MockTrans:
|
||||
self.__user = user
|
||||
self.security = self.app.security
|
||||
self.history = history
|
||||
self._short_term_cache: dict[tuple[Hashable, ...], Any] = {}
|
||||
|
||||
self.request: Any = Bunch(
|
||||
headers={},
|
||||
@@ -349,6 +351,12 @@ class MockTrans:
|
||||
def tag_handler(self):
|
||||
return self.app.tag_handler
|
||||
|
||||
def set_cache_value(self, args: tuple[Hashable, ...], value: Any):
|
||||
self._short_term_cache[args] = value
|
||||
|
||||
def get_cache_value(self, args: tuple[Hashable, ...], default: Any = None) -> Any:
|
||||
return self._short_term_cache.get(args, default)
|
||||
|
||||
def check_csrf_token(self, payload):
|
||||
pass
|
||||
|
||||
|
||||
@@ -37,7 +37,10 @@ A method that requires a user but not a history should declare its
|
||||
# more checks against this issue.
|
||||
import abc
|
||||
import string
|
||||
from collections.abc import Callable
|
||||
from collections.abc import (
|
||||
Callable,
|
||||
Hashable,
|
||||
)
|
||||
from json import dumps
|
||||
from typing import (
|
||||
Any,
|
||||
@@ -201,12 +204,12 @@ class ProvidesUserContext(ProvidesAppContext):
|
||||
workflow_building_mode: Literal[1, True, False] = False
|
||||
galaxy_session: Optional[GalaxySession] = None
|
||||
_tag_handler: Optional[GalaxyTagHandlerSession] = None
|
||||
_short_term_cache: dict[tuple[str, ...], Any]
|
||||
_short_term_cache: dict[tuple[Hashable, ...], Any]
|
||||
|
||||
def set_cache_value(self, args: tuple[str, ...], value: Any):
|
||||
def set_cache_value(self, args: tuple[Hashable, ...], value: Any):
|
||||
self._short_term_cache[args] = value
|
||||
|
||||
def get_cache_value(self, args: tuple[str, ...], default: Any = None) -> Any:
|
||||
def get_cache_value(self, args: tuple[Hashable, ...], default: Any = None) -> Any:
|
||||
return self._short_term_cache.get(args, default)
|
||||
|
||||
@property
|
||||
|
||||
@@ -102,6 +102,7 @@ from .workflow_utils import (
|
||||
if TYPE_CHECKING:
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from galaxy.managers.context import ProvidesHistoryContext
|
||||
from galaxy.model import (
|
||||
History,
|
||||
HistoryItem,
|
||||
@@ -1874,6 +1875,11 @@ def _carried_state_label(value) -> str:
|
||||
return "not in current history"
|
||||
|
||||
|
||||
# Sentinel distinguishing a populated cache entry (always a ``(rows, total)``
|
||||
# tuple) from a cache miss.
|
||||
_PAGE_CACHE_MISS: Any = object()
|
||||
|
||||
|
||||
class BaseDataToolParameter(ToolParameter):
|
||||
multiple: bool
|
||||
|
||||
@@ -1969,6 +1975,68 @@ class BaseDataToolParameter(ToolParameter):
|
||||
self._acceptable_extensions_cache = accepted
|
||||
return accepted
|
||||
|
||||
def _paginated_visible_datasets(
|
||||
self,
|
||||
trans: "ProvidesHistoryContext",
|
||||
history: "History",
|
||||
*,
|
||||
extensions: Optional[set[str]],
|
||||
valid_states: Optional[tuple[str, ...]],
|
||||
search: Optional[str] = None,
|
||||
offset: int = 0,
|
||||
limit: int = 50,
|
||||
) -> tuple[list[HistoryDatasetAssociation], int]:
|
||||
"""``history.paginated_active_visible_datasets`` memoized on the request.
|
||||
|
||||
Building one form with many ``data`` parameters (most notably the workflow
|
||||
Run form, which renders every step in a single request) otherwise re-issues
|
||||
the same paginated SQL against the same, unchanging history once per
|
||||
parameter -- O(parameters) round-trips, slow even on an empty history
|
||||
(issue #22927). The results are cached by signature on the request context's
|
||||
short-term cache (see ``ProvidesHistoryContext.set_cache_value``); the cache
|
||||
is shared across every step's proxy work context for the request and never
|
||||
outlives it, so the history cannot change underneath it.
|
||||
"""
|
||||
key = (
|
||||
"data_param_hda_page",
|
||||
history.id,
|
||||
frozenset(extensions) if extensions is not None else None,
|
||||
tuple(valid_states) if valid_states is not None else None,
|
||||
search or None,
|
||||
offset,
|
||||
limit,
|
||||
)
|
||||
cached = trans.get_cache_value(key, _PAGE_CACHE_MISS)
|
||||
if cached is not _PAGE_CACHE_MISS:
|
||||
return cached
|
||||
result = history.paginated_active_visible_datasets(
|
||||
extensions=extensions, valid_states=valid_states, search=search, offset=offset, limit=limit
|
||||
)
|
||||
trans.set_cache_value(key, result)
|
||||
return result
|
||||
|
||||
def _paginated_dataset_collections(
|
||||
self,
|
||||
trans: "ProvidesHistoryContext",
|
||||
history: "History",
|
||||
*,
|
||||
visible_only: bool,
|
||||
search: Optional[str] = None,
|
||||
offset: int = 0,
|
||||
limit: int = 50,
|
||||
) -> tuple[list[HistoryDatasetCollectionAssociation], int]:
|
||||
"""``history.paginated_active_dataset_collections`` memoized on the request
|
||||
context's short-term cache (see :meth:`_paginated_visible_datasets`)."""
|
||||
key = ("data_param_hdca_page", history.id, bool(visible_only), search or None, offset, limit)
|
||||
cached = trans.get_cache_value(key, _PAGE_CACHE_MISS)
|
||||
if cached is not _PAGE_CACHE_MISS:
|
||||
return cached
|
||||
result = history.paginated_active_dataset_collections(
|
||||
visible_only=visible_only, search=search, offset=offset, limit=limit
|
||||
)
|
||||
trans.set_cache_value(key, result)
|
||||
return result
|
||||
|
||||
def _uses_python_options_filter(self) -> bool:
|
||||
"""True iff matching depends on per-row Python state that cannot be
|
||||
pushed to SQL (dynamic options with ``options_filter_attribute``, or a
|
||||
@@ -1993,7 +2061,9 @@ class BaseDataToolParameter(ToolParameter):
|
||||
chunk_size = MAX_OPTIONS_PAGE_SIZE
|
||||
db_offset = 0
|
||||
while True:
|
||||
rows, total = history.paginated_active_visible_datasets(
|
||||
rows, total = self._paginated_visible_datasets(
|
||||
trans,
|
||||
history,
|
||||
extensions=self._acceptable_extensions(),
|
||||
valid_states=dataset_matcher_factory.valid_input_states,
|
||||
offset=db_offset,
|
||||
@@ -2013,17 +2083,19 @@ class BaseDataToolParameter(ToolParameter):
|
||||
chunk_size = MAX_OPTIONS_PAGE_SIZE
|
||||
db_offset = 0
|
||||
while True:
|
||||
rows, total = history.paginated_active_dataset_collections(
|
||||
collection_rows, total = self._paginated_dataset_collections(
|
||||
trans,
|
||||
history,
|
||||
visible_only=True,
|
||||
offset=db_offset,
|
||||
limit=chunk_size,
|
||||
)
|
||||
if not rows:
|
||||
if not collection_rows:
|
||||
return None
|
||||
for hdca in rows:
|
||||
for hdca in collection_rows:
|
||||
if dataset_collection_matcher.hdca_match(hdca):
|
||||
return hdca
|
||||
db_offset += len(rows)
|
||||
db_offset += len(collection_rows)
|
||||
if db_offset >= total:
|
||||
return None
|
||||
|
||||
@@ -2517,7 +2589,7 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
job_input_values = util.listify(other_values.get(self.name))
|
||||
|
||||
job_input_values = self._page_hda_matches(
|
||||
builder, history, dataset_matcher, dataset_matcher_factory, job_input_values
|
||||
trans, builder, history, dataset_matcher, dataset_matcher_factory, job_input_values
|
||||
)
|
||||
unresolved = self._pin_live_hda_inputs(builder, history, dataset_matcher, job_input_values)
|
||||
self._carry_unresolved_inputs(builder, history, unresolved)
|
||||
@@ -2553,6 +2625,7 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
|
||||
def _page_hda_matches(
|
||||
self,
|
||||
trans,
|
||||
builder: DataOptionsBuilder,
|
||||
history,
|
||||
dataset_matcher,
|
||||
@@ -2569,7 +2642,9 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
valid_states = dataset_matcher_factory.valid_input_states
|
||||
|
||||
def hda_query(*, offset, limit):
|
||||
return history.paginated_active_visible_datasets(
|
||||
return self._paginated_visible_datasets(
|
||||
trans,
|
||||
history,
|
||||
extensions=acceptable_extensions,
|
||||
valid_states=valid_states,
|
||||
search=hda_search,
|
||||
@@ -2677,8 +2752,8 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
_offset, _limit, hdca_search = builder.page("hdca")
|
||||
|
||||
def hdca_query(*, offset, limit):
|
||||
return history.paginated_active_dataset_collections(
|
||||
visible_only=True, search=hdca_search, offset=offset, limit=limit
|
||||
return self._paginated_dataset_collections(
|
||||
trans, history, visible_only=True, search=hdca_search, offset=offset, limit=limit
|
||||
)
|
||||
|
||||
def hdca_filter(hdca):
|
||||
@@ -2902,8 +2977,8 @@ class DataCollectionToolParameter(BaseDataToolParameter):
|
||||
history_query = self._history_query(trans)
|
||||
|
||||
def hdca_query(*, offset, limit):
|
||||
return history.paginated_active_dataset_collections(
|
||||
visible_only=False, search=hdca_search, offset=offset, limit=limit
|
||||
return self._paginated_dataset_collections(
|
||||
trans, history, visible_only=False, search=hdca_search, offset=offset, limit=limit
|
||||
)
|
||||
|
||||
def hdca_filter(hdca):
|
||||
|
||||
@@ -7,6 +7,7 @@ import os
|
||||
import re
|
||||
import socket
|
||||
import time
|
||||
from collections.abc import Hashable
|
||||
from contextlib import ExitStack
|
||||
from http.cookies import CookieError
|
||||
from typing import (
|
||||
@@ -337,7 +338,7 @@ class GalaxyWebTransaction(base.DefaultWebTransaction, context.ProvidesHistoryCo
|
||||
self.galaxy_session = None
|
||||
self.error_message = None
|
||||
self.host = self.request.host
|
||||
self._short_term_cache: dict[tuple[str, ...], Any] = {}
|
||||
self._short_term_cache: dict[tuple[Hashable, ...], Any] = {}
|
||||
|
||||
# set any cross origin resource sharing headers if configured to do so
|
||||
self.set_cors_headers()
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import abc
|
||||
from collections.abc import Hashable
|
||||
from typing import (
|
||||
Any,
|
||||
Literal,
|
||||
@@ -39,13 +40,19 @@ class WorkRequestContext(ProvidesHistoryContext):
|
||||
workflow_building_mode=False,
|
||||
url_builder=None,
|
||||
galaxy_session: Optional["GalaxySession"] = None,
|
||||
short_term_cache: Optional[dict[tuple[Hashable, ...], Any]] = None,
|
||||
):
|
||||
self._app = app
|
||||
self.__user = user
|
||||
self.__user_current_roles: Optional[list[Role]] = None
|
||||
self.__history = history
|
||||
self._url_builder = url_builder
|
||||
self._short_term_cache: dict[tuple[str, ...], Any] = {}
|
||||
# When proxying an existing transaction (see ``proxy_work_context_for_history``)
|
||||
# share its request-scoped cache so work done across proxies of the same
|
||||
# request -- e.g. every step of a workflow Run form build -- is reused.
|
||||
self._short_term_cache: dict[tuple[Hashable, ...], Any] = (
|
||||
short_term_cache if short_term_cache is not None else {}
|
||||
)
|
||||
self.workflow_building_mode = workflow_building_mode
|
||||
self.galaxy_session = galaxy_session
|
||||
|
||||
@@ -195,4 +202,5 @@ def proxy_work_context_for_history(
|
||||
url_builder=trans.url_builder,
|
||||
workflow_building_mode=workflow_building_mode,
|
||||
galaxy_session=trans.galaxy_session,
|
||||
short_term_cache=trans._short_term_cache,
|
||||
)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import contextlib
|
||||
import json
|
||||
from collections.abc import Hashable
|
||||
from typing import (
|
||||
Any,
|
||||
)
|
||||
@@ -1052,6 +1053,7 @@ class MockTrans(ProvidesAppContext):
|
||||
self.history = None
|
||||
self.workflow_building_mode = workflow_building_modes.ENABLED
|
||||
self.tag_handler = app.tag_handler
|
||||
self._short_term_cache: dict[tuple[Hashable, ...], Any] = {}
|
||||
|
||||
@property
|
||||
def galaxy_session(self):
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
"""Test Tool execution and state handling logic."""
|
||||
|
||||
from collections import OrderedDict
|
||||
from typing import cast
|
||||
from collections.abc import Hashable
|
||||
from typing import (
|
||||
Any,
|
||||
cast,
|
||||
)
|
||||
|
||||
import webob.exc
|
||||
from sqlalchemy import select
|
||||
@@ -206,6 +210,7 @@ class MockTrans:
|
||||
self.sa_session = self.app.model.context
|
||||
self.url_builder = None
|
||||
self.galaxy_session = None
|
||||
self._short_term_cache: dict[tuple[Hashable, ...], Any] = {}
|
||||
|
||||
def get_history(self, **kwargs):
|
||||
return self.history
|
||||
|
||||
Reference in New Issue
Block a user