FEATURE: Add more control over io management (#564)

Author: drewm102

src/ansys/edb/core/utility/io_manager.py

@@ -2,7 +2,7 @@
 import abc
 from collections import defaultdict
 from contextlib import contextmanager
-from enum import Enum, auto
+from enum import Enum, Flag, auto
 from sys import modules
 
 from ansys.api.edb.v1.edb_messages_pb2 import EDBObjCollectionMessage, EDBObjMessage
@@ -77,6 +77,7 @@ def __init__(self):
         self._response_cache = {}
         self._msg_type_cache = {}
         self._cached_edb_objs = {}
+        self._allow_invalidation = True
 
     def _extract_msg_from_any_module_msg(self, any_module_msg):
         msg_type_name = any_module_msg.any.TypeName()
@@ -126,7 +127,7 @@ def _hijack_request(self, service_name, rpc_name, request):
         return self._response_cache.get(self._generate_cache_key(service_name, rpc_name, request))
 
     def invalidate(self):
-        if not self._response_cache:
+        if not self._response_cache or not self.allow_invalidation:
             return
         self._response_cache.clear()
         self._cached_edb_objs = self._cached_edb_objs.fromkeys(self._cached_edb_objs, False)
@@ -153,6 +154,14 @@ def refresh_for_request(self):
             for msg in response.items:
                 self.add_from_cache_msg(msg)
 
+    @property
+    def allow_invalidation(self):
+        return self._allow_invalidation
+
+    @allow_invalidation.setter
+    def allow_invalidation(self, allow):
+        self._allow_invalidation = allow
+
 
 class _Buffer(_IOOptimizer):
     class _BufferEntry:
@@ -181,6 +190,7 @@ def _reset(self):
         self._buffer = []
         self._futures = defaultdict(list)
         self._invalidate_cache = False
+        self._allow_flushing = True
 
     def _hijack_request(self, service_name, rpc_name, request):
         if (rpc_info := get_rpc_info(service_name, rpc_name)) is None or rpc_info.is_read:
@@ -203,7 +213,7 @@ def _buffer_request_iterator(buffer):
         )
 
     def flush(self):
-        if not self._buffer:
+        if not self._buffer or not self.allow_flushing:
             return
         with self.block():
             if (cache := get_cache()) is not None and self._invalidate_cache:
@@ -225,6 +235,14 @@ def add_future_ref(self, future):
     def _reset_after_block(self):
         self._reset()
 
+    @property
+    def allow_flushing(self):
+        return self._allow_flushing
+
+    @allow_flushing.setter
+    def allow_flushing(self, allow):
+        self._allow_flushing = allow
+
 
 class ServerNotification(Enum):
     """Provides an enum representing the types of server notifications."""
@@ -234,12 +252,15 @@ class ServerNotification(Enum):
     RESET_FUTURE_TRACKING = auto()
 
 
-class IOMangementType(Enum):
+class IOMangementType(Flag):
     """Provides an enum representing the types of IO management modes."""
 
     READ = auto()
     WRITE = auto()
-    READ_AND_WRITE = auto()
+    READ_AND_WRITE = READ | WRITE
+    NO_CACHE_INVALIDATION = auto()
+    NO_BUFFER_FLUSHING = auto()
+    NO_CACHE_INVALIDATION_NO_BUFFER_FLUSHING = NO_CACHE_INVALIDATION | NO_BUFFER_FLUSHING
 
 
 class _ActiveRequestEdbObjMsgMgr:
@@ -286,16 +307,21 @@ def _enable_caching(enable):
         _get_io_manager_stub().EnableCache(bool_message(enable))
 
     def start_managing(self, mode):
-        if mode == IOMangementType.READ or mode == IOMangementType.READ_AND_WRITE:
+        if IOMangementType.READ in mode:
             self._cache = _Cache()
             self._enable_caching(True)
-        if mode == IOMangementType.WRITE or mode == IOMangementType.READ_AND_WRITE:
+        if IOMangementType.WRITE in mode:
             self._buffer = _Buffer()
+        if IOMangementType.NO_CACHE_INVALIDATION in mode:
+            self._cache.allow_invalidation = False
+        if IOMangementType.NO_BUFFER_FLUSHING in mode:
+            self._buffer.allow_flushing = False
 
     def end_managing(self):
         if self._cache is not None:
             self._enable_caching(False)
         if self._buffer is not None:
+            self._buffer.allow_flushing = True
             self._buffer.flush()
         self._reset()
 

tests/e2e/scratch/io_performance_scratch.py

@@ -6,16 +6,18 @@
 from ansys.edb.core.layer.layer import LayerType
 from ansys.edb.core.layer.stackup_layer import StackupLayer
 from ansys.edb.core.layout.cell import Cell, CellType
+from ansys.edb.core.layout.layout import Layout
 from ansys.edb.core.net.net import Net
 from ansys.edb.core.primitive.rectangle import Rectangle, RectangleRepresentationType
 from ansys.edb.core.session import session
 from ansys.edb.core.utility.io_manager import IOMangementType, enable_io_manager
+import tests.e2e.settings as settings
 
 db_name = "io_performance_test.aedb"
 lyr_name = "signal_1"
 net_name = "net_1"
 db = None
-lyt = None
+lyt: Layout = None
 
 
 def query_prims():
@@ -31,7 +33,7 @@ def query_prims():
 
 
 def create_prims():
-    num_prims = 10000
+    num_prims = 1000
     for i in range(num_prims):
         rect = Rectangle.create(
             lyt,
@@ -103,18 +105,52 @@ def teardown():
     Database.delete(db_name)
 
 
+def caching_flushing_blocking_test():
+    blocking_start = time()
+    with enable_io_manager(
+        IOMangementType.READ_AND_WRITE | IOMangementType.NO_CACHE_INVALIDATION_NO_BUFFER_FLUSHING
+    ):
+        for prim in lyt.primitives:
+            rect: Rectangle = prim
+            geom = rect.polygon_data
+            ll = geom.points[3]
+            ur = geom.points[1]
+            scaled_geom = geom.scale(2, (((ur.x - ll.x) / 2) + ll.x, ((ur.y - ll.y) / 2) + ll.y))
+            scaled_ll = scaled_geom.points[3]
+            scaled_ur = scaled_geom.points[1]
+            rect.set_parameters(
+                RectangleRepresentationType.LOWER_LEFT_UPPER_RIGHT,
+                scaled_ll.x,
+                scaled_ll.y,
+                scaled_ur.x,
+                scaled_ur.y,
+                0,
+                0,
+            )
+    blocking_end = time()
+    print(f"total blocking time: {blocking_end - blocking_start} seconds")
+
+
+def read_write_test():
+    rw_start = time()
+    num_created_prims = write_test()
+    num_queried_prims = read_test()
+    prim_io_op_counts = read_and_write_test()
+    rw_end = time()
+    total_created_prims = num_created_prims + prim_io_op_counts[0]
+    total_queried_prims = num_queried_prims + prim_io_op_counts[1]
+    print(
+        f"total rw time: {rw_end - rw_start} seconds (created {total_queried_prims} primitives, "
+        f"queried {total_created_prims} primitives)"
+    )
+
+
 if __name__ == "__main__":
-    with session(settings.server_exe_dir(), 50051):
+    with session(settings.server_exe_dir()):
         setup()
         start = time()
-        num_created_prims = write_test()
-        num_queried_prims = read_test()
-        prim_io_op_counts = read_and_write_test()
+        read_write_test()
+        caching_flushing_blocking_test()
         end = time()
-        total_created_prims = num_created_prims + prim_io_op_counts[0]
-        total_queried_prims = num_queried_prims + prim_io_op_counts[1]
-        print(
-            f"total time: {end - start} seconds (created {total_queried_prims} primitives, "
-            f"queried {total_created_prims} primitives)"
-        )
+        print(f"total time: {end - start} seconds")
         teardown()