Refactor parallelism to support PyTorch and only use shared memory

conftest.py

@@ -6,7 +6,6 @@
 
 import logging
 import os
-import time
 import warnings
 
 from seedbank import initialize, numpy_rng

docs/internals.rst

@@ -6,5 +6,4 @@ Neither LensKit users nor algorithm developers are likely to need to use this
 code directly.
 
 .. toctree::
-    sharing
     parallel

docs/parallel.rst

@@ -1,10 +1,10 @@
 Parallel Execution
 ------------------
 
-.. py:module:: lenskit.util.parallel
+.. py:module:: lenskit.parallel
 
-LensKit uses :py:class:`concurrent.futures.ProcessPoolExecutor` to paralellize batch
-operations (see :py:mod:`lenskit.batch`).
+LensKit uses a custom API wrapping  :py:class:`multiprocessing.pool.Pool` to
+paralellize batch operations (see :py:mod:`lenskit.batch`).
 
 The basic idea of this API is to create an *invoker* that has a model and a function,
 and then passing lists of argument sets to the function::
@@ -13,6 +13,7 @@ and then passing lists of argument sets to the function::
         results = list(func.map(args))
 
 The model is persisted into shared memory to be used by the worker processes.
+PyTorch tensors, including those on CUDA devices, are shared.
 
 Parallel Model Ops
 ~~~~~~~~~~~~~~~~~~
@@ -25,9 +26,3 @@ Parallel Model Ops
     :members:
 
 
-Single Process Isolation
-~~~~~~~~~~~~~~~~~~~~~~~~
-
-We also have a single-process isolation function that runs a function in a subprocess.
-
-.. autofunction:: run_sp

envs/lenskit-py3.10-ci.yaml

@@ -35,7 +35,6 @@ dependencies:
   - ruff>=0.2
   - scikit-learn>=1.1
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - setuptools>=64
   - setuptools_scm>=8
   - sphinx-autobuild>=2021
@@ -44,5 +43,7 @@ dependencies:
   - tqdm>=4
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1
       - unbeheader~=1.3

envs/lenskit-py3.10-dev.yaml

@@ -40,7 +40,6 @@ dependencies:
   - ruff>=0.2
   - scikit-learn>=1.1
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - setuptools>=64
   - setuptools_scm>=8
   - sphinx-autobuild>=2021
@@ -53,5 +52,7 @@ dependencies:
   - tqdm>=4
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1
       - unbeheader~=1.3

envs/lenskit-py3.10-doc.yaml

@@ -23,7 +23,6 @@ dependencies:
   - pandas<3,>=1.5
   - pytorch==2.*
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - sphinx>=4.2
   - sphinx_rtd_theme>=0.5
   - sphinxcontrib-bibtex>=2.0
@@ -32,4 +31,6 @@ dependencies:
   - threadpoolctl>=3.0
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1

envs/lenskit-py3.10-test.yaml

@@ -27,9 +27,10 @@ dependencies:
   - pytest==7.*
   - pytorch==2.*
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - tbb
   - threadpoolctl>=3.0
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1

envs/lenskit-py3.11-ci.yaml

@@ -35,7 +35,6 @@ dependencies:
   - ruff>=0.2
   - scikit-learn>=1.1
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - setuptools>=64
   - setuptools_scm>=8
   - sphinx-autobuild>=2021
@@ -44,5 +43,7 @@ dependencies:
   - tqdm>=4
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1
       - unbeheader~=1.3

envs/lenskit-py3.11-dev.yaml

@@ -40,7 +40,6 @@ dependencies:
   - ruff>=0.2
   - scikit-learn>=1.1
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - setuptools>=64
   - setuptools_scm>=8
   - sphinx-autobuild>=2021
@@ -53,5 +52,7 @@ dependencies:
   - tqdm>=4
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1
       - unbeheader~=1.3

envs/lenskit-py3.11-doc.yaml

@@ -23,7 +23,6 @@ dependencies:
   - pandas<3,>=1.5
   - pytorch==2.*
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - sphinx>=4.2
   - sphinx_rtd_theme>=0.5
   - sphinxcontrib-bibtex>=2.0
@@ -32,4 +31,6 @@ dependencies:
   - threadpoolctl>=3.0
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1

envs/lenskit-py3.11-test.yaml

@@ -27,9 +27,10 @@ dependencies:
   - pytest==7.*
   - pytorch==2.*
   - scipy>=1.9.0
-  - seedbank>=0.1.0
   - tbb
   - threadpoolctl>=3.0
   - pip
   - pip:
+      - manylog>=0.1.0a3
       - progress-api>=0.1.0a6
+      - seedbank>=0.2.0a1

lenskit/algorithms/item_knn.py

@@ -16,10 +16,11 @@
 import numpy as np
 import pandas as pd
 import torch
-from progress_api import Progress, make_progress
+from progress_api import Progress
 
 from lenskit import ConfigWarning, DataWarning, util
 from lenskit.data.matrix import DimStats, sparse_ratings, sparse_row_stats
+from lenskit.parallel import ensure_parallel_init
 from lenskit.util.logging import pbh_update, progress_handle
 
 from . import Predictor
@@ -465,7 +466,7 @@ def fit(self, ratings, **kwargs):
             ratings(pandas.DataFrame):
                 (user,item,rating) data for computing item similarities.
         """
-        util.check_env()
+        ensure_parallel_init()
         # Training proceeds in 2 steps:
         # 1. Normalize item vectors to be mean-centered and unit-normalized
         # 2. Compute similarities with pairwise dot products

lenskit/batch/__init__.py

@@ -10,4 +10,3 @@
 
 from ._predict import predict  # noqa: F401
 from ._recommend import recommend  # noqa: F401
-from ._train import train_isolated  # noqa: F401

lenskit/batch/_predict.py

@@ -10,6 +10,7 @@
 import pandas as pd
 
 from .. import util
+from ..parallel import invoker
 
 _logger = logging.getLogger(__name__)
 _rec_context = None
@@ -81,7 +82,7 @@ def predict(algo, pairs, *, n_jobs=None, **kwargs):
     nusers = pairs["user"].nunique()
 
     timer = util.Stopwatch()
-    with util.parallel.invoker(algo, _predict_user, n_jobs=n_jobs) as worker:
+    with invoker(algo, _predict_user, n_jobs=n_jobs) as worker:
         del algo  # maybe free some memory
 
         _logger.info(

lenskit/batch/_recommend.py

@@ -12,7 +12,7 @@
 
 from .. import util
 from ..algorithms import Recommender
-from ..sharing import PersistedModel
+from ..parallel import invoker
 
 _logger = logging.getLogger(__name__)
 
@@ -72,19 +72,18 @@
         n_jobs = kwargs["nprocs"]
         warnings.warn("nprocs is deprecated, use n_jobs", DeprecationWarning)
 
-    if not isinstance(algo, PersistedModel):
-        rec_algo = Recommender.adapt(algo)
-        if candidates is None and rec_algo is not algo:
-            warnings.warn("no candidates provided and algo is not a recommender, unlikely to work")
-        algo = rec_algo
-        del rec_algo
+    rec_algo = Recommender.adapt(algo)
+    if candidates is None and rec_algo is not algo:
+        warnings.warn("no candidates provided and algo is not a recommender, unlikely to work")
+    algo = rec_algo
+    del rec_algo
 
     if "ratings" in kwargs:
         warnings.warn("Providing ratings to recommend is not supported", DeprecationWarning)
 
     candidates = __standard_cand_fun(candidates)
 
-    with util.parallel.invoker(algo, _recommend_user, n_jobs=n_jobs) as worker:
+    with invoker(algo, _recommend_user, n_jobs=n_jobs) as worker:
         _logger.info("recommending with %s for %d users (n_jobs=%s)", str(algo), len(users), n_jobs)
         del algo
         timer = util.Stopwatch()

lenskit/parallel/__init__.py

@@ -0,0 +1,22 @@
+# This file is part of LensKit.
+# Copyright (C) 2018-2023 Boise State University
+# Copyright (C) 2023-2024 Drexel University
+# Licensed under the MIT license, see LICENSE.md for details.
+# SPDX-License-Identifier: MIT
+
+"""
+LensKit parallel computation support.
+"""
+
+from __future__ import annotations
+
+from .config import ensure_parallel_init, get_parallel_config, initialize
+from .invoker import ModelOpInvoker, invoker
+
+__all__ = [
+    "initialize",
+    "get_parallel_config",
+    "ensure_parallel_init",
+    "invoker",
+    "ModelOpInvoker",
+]