[REF] stock: perf orderpoint as batch

Currently orderpoint run one request at a time. However it's not
efficient since compute are some private methods could be executed
one time in batch instead of record by record.

The strategy in order to improve it is:
- Still cut the orderpoints in batch of 1000 (usefull when using a new
cursor with intermediate commit)
- For each batch of 1000, create a list of procurement values and run
them in batch (commit a2a6b099a4 for more info)
- Each time a procurement exception is raised, catch the failing
orderpoint and rollback to the state before run. Remove the failing
order point from the current set and run it again.
- run method try to catch a maximum  errors by continuing to process even it
finds error in order to stack them. It would avoid to run the same set
again and again each time an error is raised.

test_orderpoint_1 took 975s before this patch and 100s after.

Task: 1849912

addons/mrp/models/stock_rule.py

@@ -37,14 +37,18 @@ class StockRule(models.Model):
     @api.model
     def _run_manufacture(self, procurements):
         productions_values_by_company = defaultdict(list)
+        errors = []
         for procurement, rule in procurements:
             bom = self._get_matching_bom(procurement.product_id, procurement.company_id, procurement.values)
             if not bom:
                 msg = _('There is no Bill of Material of type manufacture or kit found for the product %s. Please define a Bill of Material for this product.') % (procurement.product_id.display_name,)
-                raise ProcurementException([(procurement, msg)])
+                errors.append((procurement, msg))
 
             productions_values_by_company[procurement.company_id.id].append(rule._prepare_mo_vals(*procurement, bom))
 
+        if errors:
+            raise ProcurementException(errors)
+
         for company_id, productions_values in productions_values_by_company.items():
             # create the MO as SUPERUSER because the current user may not have the rights to do it (mto product launched by a sale for example)
             productions = self.env['mrp.production'].sudo().with_company(company_id).create(productions_values)

addons/purchase_stock/models/stock_rule.py

@@ -40,6 +40,7 @@ class StockRule(models.Model):
     @api.model
     def _run_buy(self, procurements):
         procurements_by_po_domain = defaultdict(list)
+        errors = []
         for procurement, rule in procurements:
 
             # Get the schedule date in order to find a valid seller
@@ -53,7 +54,7 @@ class StockRule(models.Model):
 
             if not supplier:
                 msg = _('There is no matching vendor price to generate the purchase order for product %s (no vendor defined, minimum quantity not reached, dates not valid, ...). Go on the product form and complete the list of vendors.') % (procurement.product_id.display_name)
-                raise ProcurementException([(procurement, msg)])
+                errors.append((procurement, msg))
 
             partner = supplier.name
             # we put `supplier_info` in values for extensibility purposes
@@ -66,6 +67,9 @@ class StockRule(models.Model):
             domain = rule._make_po_get_domain(procurement.company_id, procurement.values, partner)
             procurements_by_po_domain[domain].append((procurement, rule))
 
+        if errors:
+            raise ProcurementException(errors)
+
         for domain, procurements_rules in procurements_by_po_domain.items():
             # Get the procurements for the current domain.
             # Get the rules for the current domain. Their only use is to create

addons/stock/models/stock_orderpoint.py

@@ -117,6 +117,12 @@ class StockWarehouseOrderpoint(models.Model):
                     raise UserError(_("Changing the company of this record is forbidden at this point, you should rather archive it and create a new one."))
         return super().write(vals)
 
+    def _get_product_context(self):
+        """Used to call `virtual_available` when running an orderpoint."""
+        self.ensure_one()
+        return {
+            'location': self.location_id.id,
+        }
 
     def _prepare_procurement_values(self, product_qty, date=False, group=False):
         """ Prepare specific key for moves or other components that will be created from a stock rule
@@ -130,3 +136,6 @@ class StockWarehouseOrderpoint(models.Model):
             'orderpoint_id': self,
             'group_id': group or self.group_id,
         }
+
+    def _post_process_scheduler(self):
+        return True

addons/stock/models/stock_rule.py

@@ -1,14 +1,15 @@
 # -*- coding: utf-8 -*-
 # Part of Odoo. See LICENSE file for full copyright and licensing details.
 
-from collections import namedtuple, OrderedDict, defaultdict
+from collections import namedtuple, defaultdict
+from datetime import datetime, time
 from dateutil.relativedelta import relativedelta
 from odoo.tools.misc import split_every
 from psycopg2 import OperationalError
 
 from odoo import api, fields, models, registry, SUPERUSER_ID, _
 from odoo.osv import expression
-from odoo.tools import DEFAULT_SERVER_DATETIME_FORMAT, float_compare, float_round
+from odoo.tools import float_compare, float_round, frozendict
 from odoo.exceptions import UserError
 
 import logging
@@ -364,10 +365,19 @@ class ProcurementGroup(models.Model):
 
     @api.model
     def run(self, procurements, raise_user_error=True):
-        """ Method used in a procurement case. The purpose is to supply the
-        product passed as argument in the location also given as an argument.
-        In order to be able to find a suitable location that provide the product
-        it will search among stock.rule.
+        """Fulfil `procurements` with the help of stock rules.
+
+        Procurements are needs of products at a certain location. To fulfil
+        these needs, we need to create some sort of documents (`stock.move`
+        by default, but extensions of `_run_` methods allow to create every
+        type of documents).
+
+        :param procurements: the description of the procurement
+        :type list: list of `~odoo.addons.stock.models.stock_rule.ProcurementGroup.Procurement`
+        :param raise_user_error: will raise either an UserError or a ProcurementException
+        :type raise_user_error: boolan, optional
+        :raises UserError: if `raise_user_error` is True and a procurement isn't fulfillable
+        :raises ProcurementException: if `raise_user_error` is False and a procurement isn't fulfillable
         """
 
         def raise_exception(procurement_errors):
@@ -506,21 +516,7 @@ class ProcurementGroup(models.Model):
 
     @api.model
     def _procurement_from_orderpoint_get_order(self):
-        return 'location_id'
-
-    @api.model
-    def _procurement_from_orderpoint_get_grouping_key(self, orderpoint_ids):
-        orderpoints = self.env['stock.warehouse.orderpoint'].browse(orderpoint_ids)
-        return orderpoints.location_id.id
-
-    @api.model
-    def _procurement_from_orderpoint_get_groups(self, orderpoint_ids):
-        """ Make groups for a given orderpoint; by default schedule all operations in one without date """
-        return [{'to_date': False, 'procurement_values': dict()}]
-
-    @api.model
-    def _procurement_from_orderpoint_post_process(self, orderpoint_ids):
-        return True
+        return 'location_id,company_id'
 
     def _get_orderpoint_domain(self, company_id=False):
         domain = [('company_id', '=', company_id)] if company_id else []
@@ -535,97 +531,80 @@ class ProcurementGroup(models.Model):
             This is appropriate for batch jobs only.
         """
         self = self.with_company(company_id)
-        OrderPoint = self.env['stock.warehouse.orderpoint']
         domain = self._get_orderpoint_domain(company_id=company_id)
-        orderpoints_noprefetch = OrderPoint.with_context(prefetch_fields=False).search(domain,
-            order=self._procurement_from_orderpoint_get_order()).ids
-        while orderpoints_noprefetch:
+        orderpoints_noprefetch = self.env['stock.warehouse.orderpoint'].search_read(
+            domain, fields=['id'], order=self._procurement_from_orderpoint_get_order())
+        orderpoints_noprefetch = [orderpoint['id'] for orderpoint in orderpoints_noprefetch]
+
+        for orderpoints_batch in split_every(1000, orderpoints_noprefetch):
             if use_new_cursor:
                 cr = registry(self._cr.dbname).cursor()
                 self = self.with_env(self.env(cr=cr))
-            OrderPoint = self.env['stock.warehouse.orderpoint']
+            orderpoints_batch = self.env['stock.warehouse.orderpoint'].browse(orderpoints_batch)
+            orderpoints_exceptions = []
+            while orderpoints_batch:
 
-            orderpoints = OrderPoint.browse(orderpoints_noprefetch[:1000])
-            orderpoints_noprefetch = orderpoints_noprefetch[1000:]
+                # Calculate groups that can be executed together
+                orderpoints_contexts = defaultdict(lambda: self.env['stock.warehouse.orderpoint'])
 
-            # Calculate groups that can be executed together
-            location_data = OrderedDict()
+                procurements = []
+                for orderpoint in orderpoints_batch:
+                    orderpoint_context = orderpoint._get_product_context()
+                    product_context = frozendict({**self.env.context, **orderpoint_context})
+                    orderpoints_contexts[product_context] |= orderpoint
 
-            def makedefault():
-                return {
-                    'products': self.env['product.product'],
-                    'orderpoints': self.env['stock.warehouse.orderpoint'],
-                    'groups': []
-                }
+                for orderpoint_context, orderpoints_by_context in orderpoints_contexts.items():
+                    substract_quantity = orderpoints_by_context._quantity_in_progress()
+                    product_quantity = orderpoints_by_context.product_id.with_context(orderpoint_context)._product_available()
 
-            for orderpoint in orderpoints:
-                key = self._procurement_from_orderpoint_get_grouping_key([orderpoint.id])
-                if not location_data.get(key):
-                    location_data[key] = makedefault()
-                location_data[key]['products'] += orderpoint.product_id
-                location_data[key]['orderpoints'] += orderpoint
-                location_data[key]['groups'] = self._procurement_from_orderpoint_get_groups([orderpoint.id])
+                    for orderpoint in orderpoints_by_context:
+                        op_product_virtual = product_quantity[orderpoint.product_id.id]['virtual_available']
+                        if op_product_virtual is None:
+                            continue
+                        if float_compare(op_product_virtual, orderpoint.product_min_qty, precision_rounding=orderpoint.product_uom.rounding) <= 0:
+                            qty = max(orderpoint.product_min_qty, orderpoint.product_max_qty) - op_product_virtual
+                            remainder = orderpoint.qty_multiple > 0 and qty % orderpoint.qty_multiple or 0.0
 
-            for location_id, location_data in location_data.items():
-                location_orderpoints = location_data['orderpoints']
-                product_context = dict(self._context, location=location_orderpoints[0].location_id.id)
-                substract_quantity = location_orderpoints._quantity_in_progress()
+                            if float_compare(remainder, 0.0, precision_rounding=orderpoint.product_uom.rounding) > 0:
+                                qty += orderpoint.qty_multiple - remainder
 
-                for group in location_data['groups']:
-                    if group.get('from_date'):
-                        product_context['from_date'] = group['from_date'].strftime(DEFAULT_SERVER_DATETIME_FORMAT)
-                    if group['to_date']:
-                        product_context['to_date'] = group['to_date'].strftime(DEFAULT_SERVER_DATETIME_FORMAT)
-                    product_quantity = location_data['products'].with_context(product_context)._product_available()
-                    for orderpoint in location_orderpoints:
-                        try:
-                            op_product_virtual = product_quantity[orderpoint.product_id.id]['virtual_available']
-                            if op_product_virtual is None:
+                            if float_compare(qty, 0.0, precision_rounding=orderpoint.product_uom.rounding) < 0:
                                 continue
-                            if float_compare(op_product_virtual, orderpoint.product_min_qty, precision_rounding=orderpoint.product_uom.rounding) <= 0:
-                                qty = max(orderpoint.product_min_qty, orderpoint.product_max_qty) - op_product_virtual
-                                remainder = orderpoint.qty_multiple > 0 and qty % orderpoint.qty_multiple or 0.0
 
-                                if float_compare(remainder, 0.0, precision_rounding=orderpoint.product_uom.rounding) > 0:
-                                    qty += orderpoint.qty_multiple - remainder
+                            qty -= substract_quantity[orderpoint.id]
+                            qty_rounded = float_round(qty, precision_rounding=orderpoint.product_uom.rounding)
+                            if qty_rounded > 0:
+                                date = datetime.combine(orderpoint_context.get('to_date'), time.min)
+                                values = orderpoint._prepare_procurement_values(qty_rounded, date=date)
+                                procurements.append(self.env['procurement.group'].Procurement(
+                                    orderpoint.product_id, qty_rounded, orderpoint.product_uom,
+                                    orderpoint.location_id, orderpoint.name, orderpoint.name,
+                                    orderpoint.company_id, values))
 
-                                if float_compare(qty, 0.0, precision_rounding=orderpoint.product_uom.rounding) < 0:
-                                    continue
+                try:
+                    with self.env.cr.savepoint():
+                        self.env['procurement.group'].with_context(from_orderpoint=True).run(procurements, raise_user_error=False)
+                except ProcurementException as errors:
+                    for procurement, error_msg in errors.procurement_exceptions:
+                        orderpoints_exceptions += [(procurement.values.get('orderpoint_id'), error_msg)]
+                    failed_orderpoints = self.env['stock.warehouse.orderpoint'].concat(*[o[0] for o in orderpoints_exceptions])
+                    if not failed_orderpoints:
+                        _logger.error('Unable to process orderpoints')
+                        break
+                    orderpoints_batch -= failed_orderpoints
 
-                                qty -= substract_quantity[orderpoint.id]
-                                qty_rounded = float_round(qty, precision_rounding=orderpoint.product_uom.rounding)
-                                if qty_rounded > 0:
-                                    values = orderpoint._prepare_procurement_values(qty_rounded, **group['procurement_values'])
-                                    try:
-                                        with self._cr.savepoint():
-                                            #TODO: make it batch
-                                            self.env['procurement.group'].with_context(from_orderpoint=True).run([self.env['procurement.group'].Procurement(
-                                                orderpoint.product_id, qty_rounded, orderpoint.product_uom,
-                                                orderpoint.location_id, orderpoint.name, orderpoint.name,
-                                                orderpoint.company_id, values)], raise_user_error=False)
-                                    except UserError as error:
-                                        self.env['stock.rule']._log_next_activity(orderpoint.product_id, error.name)
-                                    self._procurement_from_orderpoint_post_process([orderpoint.id])
-                                if use_new_cursor:
-                                    cr.commit()
-
-                        except OperationalError:
-                            if use_new_cursor:
-                                orderpoints_noprefetch += [orderpoint.id]
-                                cr.rollback()
-                                continue
-                            else:
-                                raise
-
-            try:
-                if use_new_cursor:
-                    cr.commit()
-            except OperationalError:
-                if use_new_cursor:
-                    cr.rollback()
-                    continue
+                except OperationalError:
+                    if use_new_cursor:
+                        cr.rollback()
+                        continue
+                    else:
+                        raise
                 else:
-                    raise
+                    orderpoints_batch._post_process_scheduler()
+                    break
+
+            for orderpoint, error_msg in orderpoints_exceptions:
+                self.env['stock.rule']._log_next_activity(orderpoint.product_id, error_msg)
 
             if use_new_cursor:
                 cr.commit()

addons/stock/tests/test_proc_rule.py

@@ -166,3 +166,91 @@ class TestProcRule(TransactionCase):
         self.assertAlmostEqual(move_orig.date_expected, move_orig_initial_date, delta=timedelta(seconds=10), msg='schedule date should not be impacted by action_done')
         self.assertAlmostEqual(move_orig.date, datetime.now(), delta=timedelta(seconds=10), msg='date should be now')
         self.assertAlmostEqual(move_dest.date_expected, move_dest_initial_date + timedelta(days=6), delta=timedelta(seconds=10), msg='date should be propagated')
+
+
+class TestProcRuleLoad(TransactionCase):
+    def setUp(cls):
+        super(TestProcRuleLoad, cls).setUp()
+        cls.skipTest("Performance test, too heavy to run.")
+
+    def test_orderpoint_1(self):
+        """ Try 500 products with a 1000 RR(stock -> shelf1 and stock -> shelf2)
+        Also randomly include 4 miss configuration.
+        """
+        warehouse = self.env['stock.warehouse'].create({
+            'name': 'Test Warehouse',
+            'code': 'TWH'
+        })
+        warehouse.reception_steps = 'three_steps'
+        supplier_loc = self.env.ref('stock.stock_location_suppliers')
+        stock_loc = warehouse.lot_stock_id
+        shelf1 = self.env['stock.location'].create({
+            'location_id': stock_loc.id,
+            'usage': 'internal',
+            'name': 'shelf1'
+        })
+        shelf2 = self.env['stock.location'].create({
+            'location_id': stock_loc.id,
+            'usage': 'internal',
+            'name': 'shelf2'
+        })
+
+        products = self.env['product.product'].create([{'name': i, 'type': 'product'} for i in range(500)])
+        self.env['stock.warehouse.orderpoint'].create([{
+            'product_id': products[i // 2].id,
+            'location_id': (i % 2 == 0) and shelf1.id or shelf2.id,
+            'warehouse_id': warehouse.id,
+            'product_min_qty': 5,
+            'product_max_qty': 10,
+        } for i in range(1000)])
+
+        self.env['stock.rule'].create({
+            'name': 'Rule Shelf1',
+            'route_id': warehouse.reception_route_id.id,
+            'location_id': shelf1.id,
+            'location_src_id': stock_loc.id,
+            'action': 'pull',
+            'procure_method': 'make_to_order',
+            'picking_type_id': warehouse.int_type_id.id,
+        })
+        self.env['stock.rule'].create({
+            'name': 'Rule Shelf2',
+            'route_id': warehouse.reception_route_id.id,
+            'location_id': shelf2.id,
+            'location_src_id': stock_loc.id,
+            'action': 'pull',
+            'procure_method': 'make_to_order',
+            'picking_type_id': warehouse.int_type_id.id,
+        })
+        self.env['stock.rule'].create({
+            'name': 'Rule Supplier',
+            'route_id': warehouse.reception_route_id.id,
+            'location_id': warehouse.wh_input_stock_loc_id.id,
+            'location_src_id': supplier_loc.id,
+            'action': 'pull',
+            'procure_method': 'make_to_stock',
+            'picking_type_id': warehouse.in_type_id.id,
+        })
+
+        wrong_route = self.env['stock.location.route'].create({
+            'name': 'Wrong Route',
+        })
+        self.env['stock.rule'].create({
+            'name': 'Trap Rule',
+            'route_id': wrong_route.id,
+            'location_id': warehouse.wh_input_stock_loc_id.id,
+            'location_src_id': supplier_loc.id,
+            'action': 'pull',
+            'procure_method': 'make_to_order',
+            'picking_type_id': warehouse.in_type_id.id,
+        })
+        (products[50] | products[99] | products[150] | products[199]).write({
+            'route_ids': [(4, wrong_route.id)]
+        })
+        self.env['procurement.group'].run_scheduler()
+        self.assertTrue(self.env['stock.move'].search([('product_id', 'in', products.ids)]))
+        for index in [50, 99, 150, 199]:
+            self.assertTrue(self.env['mail.activity'].search([
+                ('res_id', '=', products[index].product_tmpl_id.id),
+                ('res_model_id', '=', self.env.ref('product.model_product_template').id)
+            ]))