[IMP] orm: add optional // attr call to iter_browse

In some cases, e.g. if it is known that calling a certain method on the model
will only trigger inserts or it is clear that updates will be disjunct, such
method calls can be done in parallel.

Author: cawo-odoo

src/util/orm.py

@@ -9,8 +9,11 @@
 on this module work along the ORM of *all* supported versions.
 """
 
+import collections
 import logging
+import os
 import re
+from concurrent.futures import ProcessPoolExecutor
 from contextlib import contextmanager
 from functools import wraps
 from itertools import chain
@@ -27,9 +30,9 @@
     except ImportError:
         from odoo import SUPERUSER_ID
     from odoo import fields as ofields
-    from odoo import modules, release
+    from odoo import modules, release, sql_db
 except ImportError:
-    from openerp import SUPERUSER_ID, modules, release
+    from openerp import SUPERUSER_ID, modules, release, sql_db
 
     try:
         from openerp import fields as ofields
@@ -41,8 +44,8 @@
 from .const import BIG_TABLE_THRESHOLD
 from .exceptions import MigrationError
 from .helpers import table_of_model
-from .misc import chunks, log_progress, version_between, version_gte
-from .pg import column_exists, format_query, get_columns, named_cursor
+from .misc import chunks, log_progress, str2bool, version_between, version_gte
+from .pg import column_exists, format_query, get_columns, get_max_workers, named_cursor
 
 # python3 shims
 try:
@@ -52,6 +55,8 @@
 
 _logger = logging.getLogger(__name__)
 
+UPG_PARALLEL_ITER_BROWSE = str2bool(os.environ.get("UPG_PARALLEL_ITER_BROWSE", "0"))
+
 
 def env(cr):
     """
@@ -338,6 +343,31 @@ def get_ids():
         invalidate(records)
 
 
+def _mp_iter_browse_cb(ids_or_values):
+    # init upon first call. Done here instead of initializer callback, because py3.6 doesn't have it
+    if not hasattr(_mp_iter_browse_cb, "env"):
+        sql_db._Pool = None  # children cannot borrow from copies of the same pool, it will cause protocol error
+        _mp_iter_browse_cb.env = env(sql_db.db_connect(_mp_iter_browse_cb.dbname).cursor())
+        _mp_iter_browse_cb.env.clear()
+    # process
+    if _mp_iter_browse_cb.mode == "browse":
+        getattr(
+            _mp_iter_browse_cb.env[_mp_iter_browse_cb.model_name].browse(ids_or_values), _mp_iter_browse_cb.attr_name
+        )(*_mp_iter_browse_cb.args, **_mp_iter_browse_cb.kwargs)
+    if _mp_iter_browse_cb.mode == "create":
+        _mp_iter_browse_cb.env[_mp_iter_browse_cb.model_name].create(ids_or_values)
+    _mp_iter_browse_cb.env.cr.commit()
+
+
+def _init_mp_iter_browse_cb(dbname, model_name, attr_name, args, kwargs, mode):
+    _mp_iter_browse_cb.dbname = dbname
+    _mp_iter_browse_cb.model_name = model_name
+    _mp_iter_browse_cb.attr_name = attr_name
+    _mp_iter_browse_cb.args = args
+    _mp_iter_browse_cb.kwargs = kwargs
+    _mp_iter_browse_cb.mode = mode
+
+
 class iter_browse(object):
     """
     Iterate over recordsets.
@@ -372,8 +402,8 @@ class iter_browse(object):
     :param model: the model to iterate
     :type model: :class:`odoo.model.Model`
     :param iterable(int) ids: iterable of IDs of the records to iterate
-    :param int chunk_size: number of records to load in each iteration chunk, `200` by
-                           default
+    :param int chunk_size: number of records to load in each iteration chunk, `200` by default
+    :param bool multiprocessing: whether to process chunks in parallel
     :param logger: logger used to report the progress, by default
                    :data:`~odoo.upgrade.util.orm._logger`
     :type logger: :class:`logging.Logger`
@@ -384,7 +414,17 @@ class iter_browse(object):
     See also :func:`~odoo.upgrade.util.orm.env`
     """
 
-    __slots__ = ("_chunk_size", "_cr_uid", "_it", "_logger", "_model", "_patch", "_size", "_strategy")
+    __slots__ = (
+        "_chunk_size",
+        "_cr_uid",
+        "_it",
+        "_logger",
+        "_model",
+        "_multiprocessing",
+        "_patch",
+        "_size",
+        "_strategy",
+    )
 
     def __init__(self, model, *args, **kw):
         assert len(args) in [1, 3]  # either (cr, uid, ids) or (ids,)
@@ -398,9 +438,24 @@ def __init__(self, model, *args, **kw):
             except TypeError:
                 raise ValueError("When passing ids as a generator, the size kwarg is mandatory")
         self._chunk_size = kw.pop("chunk_size", 200)  # keyword-only argument
+        self._multiprocessing = kw.pop("multiprocessing", False)
         self._logger = kw.pop("logger", _logger)
-        self._strategy = kw.pop("strategy", "flush")
+        self._strategy = kw.pop("strategy", "commit" if self._multiprocessing and UPG_PARALLEL_ITER_BROWSE else "flush")
         assert self._strategy in {"flush", "commit"}
+        if self._multiprocessing:
+            if self._strategy == "flush":
+                raise ValueError("With multiprocessing, strategy must be 'commit'")
+            if self._size > 100000 and self._logger and not UPG_PARALLEL_ITER_BROWSE:
+                self._logger.warning(
+                    "Browsing %d %s, which may take a long time. "
+                    "This can be sped up by setting the env variable UPG_PARALLEL_ITER_BROWSE to 1. "
+                    "If you do, be sure to examine the results carefully.",
+                    self._size,
+                    self._model._name,
+                )
+            if UPG_PARALLEL_ITER_BROWSE:
+                self._task_size = self._chunk_size
+                self._chunk_size = max(get_max_workers() * 10 * self._task_size, 1000000)
         if kw:
             raise TypeError("Unknown arguments: %s" % ", ".join(kw))
 
@@ -452,6 +507,18 @@ def __getattr__(self, attr):
 
         def caller(*args, **kwargs):
             args = self._cr_uid + args
+            if self._multiprocessing:
+                _init_mp_iter_browse_cb(self._model.env.cr.dbname, self._model._name, attr, args, kwargs, "browse")
+                with ProcessPoolExecutor(max_workers=get_max_workers()) as executor:
+                    for chunk in it:
+                        collections.deque(
+                            executor.map(_mp_iter_browse_cb, chunks(chunk._ids, self._task_size, fmt=tuple)),
+                            maxlen=0,
+                        )
+                next(self._end(), None)
+                # do not return results in // mode, we expect it to be used for huge numbers of
+                # records and thus would risk MemoryError
+                return None
             return [getattr(chnk, attr)(*args, **kwargs) for chnk in chain(it, self._end())]
 
         self._it = None