diff --git a/assume/common/outputs.py b/assume/common/outputs.py index 764443564..f60acf8a4 100644 --- a/assume/common/outputs.py +++ b/assume/common/outputs.py @@ -658,6 +658,13 @@ async def on_stop(self): # insert left records into db await self.store_dfs() + # Dispose the SQLAlchemy engine to release pooled connections + if self.db is not None: + try: + self.db.dispose() + except Exception: + pass + if self.db is None: return diff --git a/assume/common/units_operator.py b/assume/common/units_operator.py index 913cf6d84..e7a33684f 100644 --- a/assume/common/units_operator.py +++ b/assume/common/units_operator.py @@ -234,7 +234,9 @@ def handle_market_feedback(self, content: ClearingMessage, meta: MetaDict) -> No order["market_id"] = content["market_id"] marketconfig = self.registered_markets[content["market_id"]] - self.valid_orders[marketconfig.product_type].extend(orderbook) + # Only keep accepted orders for dispatch tracking — + # rejected orders are not needed and cause unbounded growth. + self.valid_orders[marketconfig.product_type].extend(accepted_orders) self.set_unit_dispatch(orderbook, marketconfig) self.write_actual_dispatch(marketconfig.product_type) diff --git a/assume/common/utils.py b/assume/common/utils.py index 8c6cbe8ea..91d87693b 100644 --- a/assume/common/utils.py +++ b/assume/common/utils.py @@ -737,16 +737,52 @@ def parse_duration(duration_str): def calculate_content_size(content: list | dict) -> int: """ - Calculate the size of a content in bytes. + Calculate the deep memory size of content in bytes. + + Uses recursive traversal to account for nested structures, + DataFrames, numpy arrays, and torch tensors — unlike sys.getsizeof + which only measures the shallow container size. """ - if isinstance(content, dict): # For dictionaries + try: + import pandas as pd + + if isinstance(content, pd.DataFrame): + return int(content.memory_usage(deep=True).sum()) + if isinstance(content, pd.Series): + return int(content.memory_usage(deep=True)) + except Exception: + pass + + try: + import numpy as np + + if isinstance(content, np.ndarray): + return content.nbytes + except Exception: + pass + + try: + import torch + + if isinstance(content, torch.Tensor): + return content.nelement() * content.element_size() + except Exception: + pass + + if isinstance(content, dict): return sys.getsizeof(content) + sum( - sys.getsizeof(value) for value in content.values() + calculate_content_size(k) + calculate_content_size(v) + for k, v in content.items() ) - elif isinstance(content, list): # For lists, including lists of dicts + elif isinstance(content, (list, tuple)): return sys.getsizeof(content) + sum( calculate_content_size(item) for item in content ) + elif isinstance(content, str): + return sys.getsizeof(content) + elif isinstance(content, (int, float, bool, type(None))): + return sys.getsizeof(content) + # Fallback for unknown types return sys.getsizeof(content) diff --git a/assume/markets/base_market.py b/assume/markets/base_market.py index 760e15c8d..1defd5fb5 100644 --- a/assume/markets/base_market.py +++ b/assume/markets/base_market.py @@ -670,7 +670,7 @@ async def clear_market(self, market_products: list[MarketProduct]): 0, ) - self.open_auctions - set(market_products) + self.open_auctions -= set(market_products) accepted_orderbook = sorted( accepted_orderbook, key=lambda x: str(x["agent_addr"]) @@ -725,6 +725,12 @@ async def clear_market(self, market_products: list[MarketProduct]): await self.store_market_results(market_meta) + # Prevent unbounded growth of in-memory results. + # Keep only results that data-request handlers may still need + # (those whose product_start has not yet passed). + now = timestamp2datetime(self.context.current_timestamp) + self.results = [r for r in self.results if r.get("product_start", now) >= now] + if flows is not None and len(flows) > 0: await self.store_flows(flows) diff --git a/assume/reinforcement_learning/learning_role.py b/assume/reinforcement_learning/learning_role.py index 9c14db92b..b8b1c582e 100644 --- a/assume/reinforcement_learning/learning_role.py +++ b/assume/reinforcement_learning/learning_role.py @@ -616,6 +616,14 @@ def init_logging( train_start (str): The start time of simulation. """ + # Explicitly close the previous logger to release its DB engine + # and SummaryWriter file handles before creating a new one. + if ( + hasattr(self, "tensor_board_logger") + and self.tensor_board_logger is not None + ): + self.tensor_board_logger.close() + self.tensor_board_logger = TensorBoardLogger( simulation_id=simulation_id, db_uri=db_uri, diff --git a/assume/reinforcement_learning/tensorboard_logger.py b/assume/reinforcement_learning/tensorboard_logger.py index 56fca3f68..f9dfe24df 100644 --- a/assume/reinforcement_learning/tensorboard_logger.py +++ b/assume/reinforcement_learning/tensorboard_logger.py @@ -363,10 +363,28 @@ def update_tensorboard(self): logger.error(f"Unexpected error in update_tensorboard: {e}") return + def close(self): + """ + Explicitly release resources held by this logger. + Call this before discarding the instance to avoid leaked + file handles (SummaryWriter) and DB connection pools. + """ + if hasattr(self, "writer") and self.writer is not None: + try: + self.writer.flush() + self.writer.close() + except Exception: + pass + self.writer = None + if hasattr(self, "db") and self.db is not None: + try: + self.db.dispose() + except Exception: + pass + self.db = None + def __del__(self): """ Deletes the WriteOutput instance. """ - if hasattr(self, "writer") and self.writer is not None: - self.writer.flush() - self.writer.close() + self.close() diff --git a/assume/world.py b/assume/world.py index 3c3e2d7b4..7f560bd6f 100644 --- a/assume/world.py +++ b/assume/world.py @@ -864,6 +864,7 @@ def run(self): def reset(self): """ Reset the market operators, markets, unit operators, and forecast providers to empty dictionaries. + Also clears caches and large data structures to prevent memory leaks across episodes. Returns: None @@ -874,6 +875,26 @@ def reset(self): self.units = {} self.forecast_providers = {} + # Clear forecast algorithm lru_caches that retain full DataFrames + # and references to old unit/config objects from previous episodes. + from assume.common.forecast_algorithms import ( + calculate_naive_congestion_signal, + calculate_naive_price, + calculate_naive_price_elastic, + calculate_naive_price_inelastic, + calculate_naive_renewable_utilisation, + calculate_naive_residual_load, + sort_units, + ) + + sort_units.cache_clear() + calculate_naive_price_inelastic.cache_clear() + calculate_naive_price_elastic.cache_clear() + calculate_naive_price.cache_clear() + calculate_naive_residual_load.cache_clear() + calculate_naive_congestion_signal.cache_clear() + calculate_naive_renewable_utilisation.cache_clear() + def add_unit( self, id: str, diff --git a/docs/source/release_notes.rst b/docs/source/release_notes.rst index 50ced9f0f..170d12e13 100644 --- a/docs/source/release_notes.rst +++ b/docs/source/release_notes.rst @@ -25,6 +25,7 @@ Upcoming Release - **Fix bug in forecasts**, that occurred when using complex clearing - **Fix infeasible power output in PowerPlant**: ``calculate_min_max_power`` now correctly accounts for base load, positive/negative capacity reserves, and heat demand when computing additional power. If reduced availability makes the unit infeasible to run, both min and max power are set to 0. A warning is issued if previous dispatch exceeded available power. - **Fix upward redispatch potential**, so that availabilities are now correctly considered instead of the nominal power output of the unit + - **Reduce memory growth in long-running workflows**: Several likely object-retention points were cleaned up so that running many simulations in parallel for long periods no longer accumulates memory as aggressively. Functional behaviour is intended to remain unchanged. 0.6.1 - (25th March 2026) =========================