|
| 1 | +# ____________________________________________________________________________________ |
| 2 | +# |
| 3 | +# Pyomo: Python Optimization Modeling Objects |
| 4 | +# Copyright (c) 2008-2026 National Technology and Engineering Solutions of Sandia, LLC |
| 5 | +# Under the terms of Contract DE-NA0003525 with National Technology and Engineering |
| 6 | +# Solutions of Sandia, LLC, the U.S. Government retains certain rights in this |
| 7 | +# software. This software is distributed under the 3-clause BSD License. |
| 8 | +# ____________________________________________________________________________________ |
| 9 | + |
| 10 | +import pyomo.environ as pyo |
| 11 | +import pyomo.common.unittest as unittest |
| 12 | +import gc |
| 13 | +from pyomo.contrib.appsi.cmodel import cmodel_available |
| 14 | + |
| 15 | +from pyomo.common.dependencies import attempt_import |
| 16 | + |
| 17 | +# Note: tracemalloc is not always available, e.g., under PyPy |
| 18 | +tracemalloc, tracemalloc_available = attempt_import('tracemalloc') |
| 19 | + |
| 20 | + |
| 21 | +class TestAppsiLegacyLeak(unittest.TestCase): |
| 22 | + @unittest.skipIf(not tracemalloc_available, "tracemalloc not available") |
| 23 | + @unittest.skipIf(not cmodel_available, "APPSI C-extension not available") |
| 24 | + def test_legacy_solver_wrapper_memory_leak(self): |
| 25 | + tracemalloc.start() |
| 26 | + # 1. Create a minimal structure |
| 27 | + model = pyo.ConcreteModel() |
| 28 | + model.I = pyo.RangeSet(100) |
| 29 | + model.x = pyo.Var(model.I) |
| 30 | + model.obj = pyo.Objective(expr=sum(model.x[i] for i in model.I)) |
| 31 | + model.c = pyo.ConstraintList() |
| 32 | + for i in model.I: |
| 33 | + model.c.add(model.x[i] >= 0) |
| 34 | + |
| 35 | + # 2. Instantiate Legacy Wrapper |
| 36 | + solver = pyo.SolverFactory('appsi_cbc') |
| 37 | + if not solver.available(): |
| 38 | + raise unittest.SkipTest("appsi_cbc solver is not available") |
| 39 | + |
| 40 | + solver.set_instance(model) |
| 41 | + |
| 42 | + # Warm-up solve |
| 43 | + solver.solve(model) |
| 44 | + gc.collect() |
| 45 | + |
| 46 | + # 3. Take initial memory snapshot |
| 47 | + |
| 48 | + s1 = tracemalloc.take_snapshot() |
| 49 | + |
| 50 | + # 4. Perform iterative solves |
| 51 | + iterations = 10 |
| 52 | + for _ in range(iterations): |
| 53 | + solver.solve(model) |
| 54 | + |
| 55 | + gc.collect() |
| 56 | + s2 = tracemalloc.take_snapshot() |
| 57 | + tracemalloc.stop() |
| 58 | + |
| 59 | + # 5. Measure memory delta |
| 60 | + stats = s2.compare_to(s1, 'lineno') |
| 61 | + total_leak = sum(stat.size_diff for stat in stats) |
| 62 | + |
| 63 | + initial_size = sum(stat.size for stat in s1.statistics('lineno')) |
| 64 | + final_size = sum(stat.size for stat in s2.statistics('lineno')) |
| 65 | + percentage_increase_per_solve = (total_leak / iterations) / initial_size * 100 |
| 66 | + |
| 67 | + # We allow a small tolerance for memory use growth, set here |
| 68 | + threshold_pct = 3 |
| 69 | + print(f"Percentage increase per solve: {percentage_increase_per_solve}%") |
| 70 | + # Check if the leak is substantial |
| 71 | + self.assertLess( |
| 72 | + percentage_increase_per_solve, |
| 73 | + threshold_pct, |
| 74 | + f"More than {threshold_pct}% memory leak detected across iterations", |
| 75 | + ) |
| 76 | + |
| 77 | + |
| 78 | +if __name__ == "__main__": |
| 79 | + unittest.main() |
0 commit comments