Cleanup after instance cache rework. (#6209)

* Cleanup after instance cache rework.
* Improve instance finalizer.

---------

Signed-off-by: Michal Zientkiewicz <michalz@nvidia.com>

Author: mzient

dali/python/nvidia/dali/experimental/dynamic/_invocation.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import threading
+import weakref
 from typing import TYPE_CHECKING, Any, Optional
 
 import nvtx
@@ -96,14 +97,16 @@ def __init__(
             else None
         )
 
-    def __del__(self):
-        self._return_op_to_cache()
+        if hasattr(self._operator, "_cache"):
+            self._return_op_to_cache = weakref.finalize(
+                self, Invocation._return_op_to_cache_impl, self._operator
+            )
+        else:
+            self._return_op_to_cache = None
 
-    def _return_op_to_cache(self):
-        if (cache := getattr(self._operator, "_cache", None)) is not None:
-            cache[self._operator._key] = self._operator
-        self._operator = None
-        self._return_op_to_cache = lambda: None
+    @staticmethod
+    def _return_op_to_cache_impl(op):
+        op._cache[op._key] = op
 
     def device(self, result_index: int):
         if self._output_devices is None:
@@ -305,7 +308,9 @@ def output_device(x):
                     assert output_device(self._results[i]) == d
 
             ctx.cache_results(self, self._results)
-            self._return_op_to_cache()  # the operator instance is ready for a new invocation
+            if self._return_op_to_cache:
+                self._return_op_to_cache()  # the operator instance is ready for a new invocation
+            self._operator = None
 
     def values(self, ctx: Optional[_EvalContext] = None):
         """

dali/python/nvidia/dali/experimental/dynamic/_ops.py

@@ -285,6 +285,14 @@ def _input_device(
 
     @classmethod
     def _process_params(cls, backend, op_device, batch_size, *raw_args, **raw_kwargs):
+        """
+        Processes run-time parameters passed to the operator to ones that can be consumed DALI
+        (Batch or Tensor).
+
+        This is a class method, as it doesn't require an operator instance - and this method
+        is essential for proper operator instance caching, as input/argument metadata is a part
+        of the operator cache key.
+        """
         is_batch = batch_size is not None
         if cls._has_random_state_arg:
             from . import random
@@ -348,12 +356,6 @@ def _pre_call(self, *inputs, **args):
     def _is_backend_initialized(self):
         return self._op_backend is not None
 
-    def _reset_backend(self):
-        self._op_backend = None
-        self._op_spec = None
-        self._input_meta = []
-        self._arg_meta = {}
-
     def _init_spec(self, inputs, args):
         if self._op_spec is None:
             import nvidia.dali as dali