Lazyflow logging and fix default operator naming

ilastik/ilastik_logging/default_config.py

@@ -127,6 +127,7 @@ def get_default_config(
                 "maxBytes": 50e6,  # 20 MB
                 "backupCount": 5,
                 "formatter": "verbose",
+                "encoding": "utf-8",
             },
         },
         "root": {"handlers": root_handlers, "level": root_log_level},

ilastik/ilastik_logging/logging_config.json

@@ -27,13 +27,13 @@
         "ilastik.utility":                                          { "level":"INFO" },
         "ilastik.utility.exportingOperator":                        { "level":"INFO" },
         "ilastik.utility.exportFile":                               { "level":"INFO" },
+        "lazyflow.signals":                                         { "level":"INFO" },
         "lazyflow.request":                                         { "level":"INFO" },
         "lazyflow.request.RequestLock":                             { "level":"INFO" },
         "lazyflow.request.SimpleRequestCondition":                  { "level":"INFO" },
         "lazyflow.graph":                                           { "level":"INFO" },
-        "lazyflow.graph.Slot":                                      { "level":"INFO" },
-        "lazyflow.slot.Slot":                                       { "level":"INFO" },
         "lazyflow.classifiers":                                     { "level":"INFO" },
+        "lazyflow.op_debug":                                        { "level":"INFO" },
         "lazyflow.operatorWrapper.OperatorWrapper":                 { "level":"INFO" },
         "lazyflow.operators.ioOperators":                           { "level":"INFO" },
         "lazyflow.operators.ioOperators.opRESTfulVolumeReader":     { "level":"INFO" },

lazyflow/graph.py

@@ -63,6 +63,7 @@ def call_when_setup_finished(self, fn):
             fn()
         else:
             # Subscribe to the next completion.
+            logger.debug(f"Adding to queue in setup={id(self._sig_setup_complete)}: {fn}")
             self._sig_setup_complete.subscribe(fn)
 
     class SetupDepthContext(object):
@@ -79,6 +80,7 @@ def __enter__(self):
                 with self._graph._lock:
                     if self._graph._setup_depth == 0:
                         self._graph._sig_setup_complete = OrderedSignal()
+                        logger.debug(f"New setupDepthContext={id(self._graph._sig_setup_complete)}")
                     self._graph._setup_depth += 1
 
         def __exit__(self, *args):
@@ -88,7 +90,8 @@ def __exit__(self, *args):
                     self._graph._setup_depth -= 1
                     if self._graph._setup_depth == 0:
                         sig_setup_complete = self._graph._sig_setup_complete
-                        # Reset.
+                        logger.debug(f"Finish setupDepthContext={id(sig_setup_complete)}")
                         self._graph._sig_setup_complete = None
                 if sig_setup_complete:
+                    logger.debug(f"Flushing queue of setupDepthContext={id(sig_setup_complete)}")
                     sig_setup_complete()

lazyflow/operator.py

@@ -173,9 +173,6 @@
 
     """
 
-    loggerName = __name__ + ".Operator"
-    logger = logging.getLogger(loggerName)
-
     # definition of inputs slots
     inputSlots = []
 
@@ -221,6 +218,9 @@
                 )
             graph = parent.graph
 
+        if self.name == Operator.name:
+            self.name = type(self).__name__  # better default if subclass does not override self.name
+
         self._cleaningUp = False
         self.graph = graph
         self._children = collections.OrderedDict()
@@ -266,9 +266,8 @@
 
     # continue initialization, when user overrides __init__
     def _after_init(self):
-        # provide simple default name for lazy users
         if self.name == Operator.name:
-            self.name = type(self).__name__
+            self.name = type(self).__name__  # repeat in case subclass.__init__ does not call super().__init__()
         assert self.graph is not None, (
             "Operator {}: self.graph is None, the parent ({})"
             " given to the operator must have a valid .graph attribute!".format(self, self._parent)
@@ -301,6 +300,10 @@
             islot.notifyUnready(self.handleInputBecameUnready)
 
         self._initialized = True
+
+        self.logger = logging.getLogger(f"lazyflow.op_debug.{type(self).__name__}")
+        self.logger.debug(f"Instantiated {self.name} {id(self)} with parent={self.parent.name if self.parent else ''}")
+
         self._setupOutputs()
 
     def _instantiate_slots(self):
@@ -393,6 +396,7 @@
         if self._parent is not None:
             del self._parent._children[self]
 
+        self.logger.debug(f"Cleaning up {self.name} {id(self)}")
         # Disconnect ourselves and all children
         self._disconnect()
 
@@ -489,6 +493,7 @@
                 self._condition.wait()
 
             self._settingUp = True
+            self.logger.debug(f"Starting setupOutputs on {self.name} {id(self)}")
 
             # Keep a copy of the old metadata for comparison.
             #  We only trigger downstream changes if something really changed.
@@ -498,6 +503,7 @@
             self.setupOutputs()
             self._setup_count += 1
 
+            self.logger.debug(f"Finished setupOutputs on {self.name} {id(self)}")
             self._settingUp = False
             self._condition.notifyAll()
 
@@ -588,6 +594,7 @@
             # We are executing the operator. Incremement the execution
             # count to protect against simultaneous setupOutputs()
             # calls.
+            self.logger.debug(f"Executing {self.name} {id(self)} slot={slot.name} for roi={str(roi)} with {kwargs=}")
             self._incrementOperatorExecutionCount()
             return self.execute(slot, subindex, roi, result, **kwargs)
         finally:

lazyflow/operatorWrapper.py

@@ -90,7 +90,7 @@ def __init__(
 
         self._initialized = False
 
-        if operatorClass.name == "":
+        if operatorClass.name == Operator.name:
             self._name = "Wrapped " + operatorClass.__name__
         else:
             self._name = "Wrapped " + operatorClass.name

lazyflow/slot.py

@@ -30,9 +30,7 @@
 ###############################################################################
 # Python
 import logging
-import collections
 import itertools
-import threading
 from functools import partial, wraps
 from contextlib import contextmanager
 import warnings
@@ -51,6 +49,8 @@
 from lazyflow.metaDict import MetaDict
 from lazyflow.utility import slicingtools, OrderedSignal
 
+module_logger = logging.getLogger(__name__)
+
 
 def is_setup_fn(func):
     """
@@ -81,9 +81,6 @@
     Base class for InputSlot, OutputSlot
     """
 
-    loggerName = __name__ + ".Slot"
-    logger = logging.getLogger(loggerName)
-
     # Allow slots to be sorted by their order of creation for debug
     # output and diagramming purposes.
     _global_counter = itertools.count()
@@ -108,6 +105,7 @@
         allow_mask=False,
         subindex=(),
         top_level_slot=None,
+        write_logs=False,
     ):
         """Constructor of the Slot class.
 
@@ -162,6 +160,13 @@
         self._optional = optional
         self.operator = operator
         self.allow_mask = allow_mask
+        self.logger = None
+        if write_logs:
+            # Using module-like dot separation allows turning on/off all slot loggers at once
+            logger_name = f"lazyflow.slot.NoOperator.{self.name}"
+            if self.operator is not None:
+                logger_name = f"lazyflow.slot.{self.operator.name}.{self.name}"
+            self.logger = logging.getLogger(logger_name)
 
         # in the case of an InputSlot this is the slot to which it is
         # connected
@@ -519,6 +524,8 @@
                         )
                     )
 
+                    if self.logger:
+                        self.logger.debug(f"Connecting to {upstream_slot}")
                     self.upstream_slot = upstream_slot
                     notifyReady = self.upstream_slot.meta._ready and not self.meta._ready
                     self.meta = self.upstream_slot.meta.copy()
@@ -604,11 +611,10 @@
                     # Well, this is bad.  We caused an exception while handling an exception.
                     # We're more interested in the FIRST exception, so print this one out and
                     #  continue unwinding the stack with the first one.
-                    self.logger.error("Error: Caught a secondary exception while handling a different exception.")
+                    module_logger.error("Caught a secondary exception while handling a different exception.")
                     import traceback
 
                     traceback.print_exc()
-                    pass
 
     @is_setup_fn
     def disconnect(self):
@@ -626,6 +632,8 @@
         had_upstream_slot = False
         if self.upstream_slot is not None:
             had_upstream_slot = True
+            if self.logger:
+                self.logger.debug(f"Disconnecting from {self.upstream_slot}")
             # safe to unsubscribe, even if not subscribed
             self.upstream_slot._sig_unready.unsubscribe(self._handleUpstreamUnready)
             try:
@@ -672,8 +680,8 @@
             return
 
         self._resizing = True
-        if self.operator is not None:
-            self.logger.debug("Resizing slot {} of operator {} to size {}".format(self.name, self.operator.name, size))
+        if self.logger:
+            self.logger.debug(f"Resizing to {size=}")
 
         # call before resize callbacks
         self._sig_resize(self, oldsize, size)
@@ -718,19 +726,14 @@
         # call after insert callbacks
         self._sig_insert(self, position, finalsize)
 
+        if self.logger:
+            self.logger.debug(f"Inserting slot {position} to {finalsize=}")
+
         slot = self._insertNew(position)
 
         # New slot inherits our settings
         slot.backpropagate_values = self.backpropagate_values
 
-        operator_name = "<NO OPERATOR>"
-        if self.operator:
-            operator_name = self.operator.name
-        self.logger.debug(
-            "Inserting slot {} into slot {} of operator {} to size {}".format(
-                position, self.name, operator_name, finalsize
-            )
-        )
         if propagate:
             if self.upstream_slot is not None and self.upstream_slot.level == self.level:
                 self.upstream_slot.insertSlot(position, finalsize)
@@ -754,12 +757,8 @@
         if len(self) <= finalsize:
             return None
         assert position < len(self)
-        if self.operator is not None:
-            self.logger.debug(
-                "Removing slot {} into slot {} of operator {} to size {}".format(
-                    position, self.name, self.operator.name, finalsize
-                )
-            )
+        if self.logger:
+            self.logger.debug(f"Removing slot {position} to {finalsize=}")
 
         # call before-remove callbacks
         self._sig_remove(self, position, finalsize)
@@ -793,6 +792,8 @@
 
         """
         if self._value is not None:
+            if self.logger:
+                self.logger.debug(f"Returning value={self._value}")
             # this handles the case of an inputslot
             # having a ._value
             # --> construct cheaper request object for this case
@@ -801,6 +802,8 @@
         elif self.upstream_slot is not None:
             # this handles the case of an inputslot
             # --> just relay the request
+            if self.logger:
+                self.logger.debug(f"Passing request to {self.upstream_slot}.")
             return self.upstream_slot.get(roi)
         else:
             if not self.ready():
@@ -826,6 +829,8 @@
                 ), "This inputSlot has no value and no upstream_slot.  You can't ask for its data yet!"
             # normal (outputslot) case
             # --> construct heavy request object..
+            if self.logger:
+                self.logger.debug(f"Getting data for {roi=}")
             execWrapper = Slot.RequestExecutionWrapper(self, roi)
             request = Request(execWrapper)
 
@@ -961,9 +966,6 @@
                 # Something is wrong.  Are we cancelled?
                 Request.raise_if_cancelled()
                 if not self.ready():
-                    # msg = "This slot ({}.{}) isn't ready yet, which means " \
-                    #      "you can't ask for its data.  Is it connected?".format(self.operator and self.operator.name, self.name)
-                    # self.logger.error(msg)
                     problem_slot = Slot._findUpstreamProblemSlot(self)
                     problem_str = str(problem_slot)
                     if isinstance(problem_slot, Slot):
@@ -978,7 +980,6 @@
                         "First upstream problem slot is: {}"
                         "".format(self.operator and self.operator.__class__, self.name, problem_str)
                     )
-                    # self.logger.error(slotInfoMsg)
                     raise Slot.SlotNotReadyError(slotInfoMsg)
                 assert self.meta.shape is not None, (
                     "Can't ask for slices of this slot yet:"
@@ -1099,6 +1100,8 @@
                     " Call disconnect first if that's what you really wanted."
                 )
             elif self.upstream_slot is not None:
+                if self.logger:
+                    self.logger.debug("Propagating setValue upstream.")
                 self.upstream_slot.setValue(value, notify, check_changed)
                 return
 
@@ -1139,6 +1142,8 @@
                     changed = not same
 
             if changed:
+                if self.logger:
+                    self.logger.debug(f"Setting {value=}")
                 # call disconnect callbacks
                 self._sig_disconnect(self)
                 self._value = value
@@ -1155,10 +1160,14 @@
                 # a slot with a value is ready unless the value is None.
                 if self._value is not None:
                     if self.meta._ready != True:
+                        if self.logger:
+                            self.logger.debug("Now ready.")
                         self.meta._ready = True
                         self._sig_ready(self)
                 else:
                     if self.meta._ready != False:
+                        if self.logger:
+                            self.logger.debug("Now unready.")
                         self.meta._ready = False
                         self._sig_unready(self)
 
@@ -1191,11 +1200,10 @@
                     # Well, this is bad.  We caused an exception while handling an exception.
                     # We're more interested in the FIRST excpetion, so print this one out and
                     #  continue unwinding the stack with the first one.
-                    self.logger.error("Error: Caught a secondary exception while handling a different exception.")
+                    module_logger.error("Caught a secondary exception while handling a different exception.")
                     import traceback
 
                     traceback.print_exc()
-                    pass
 
     @is_setup_fn
     def setValues(self, values):
@@ -1269,6 +1277,7 @@
         return ready
 
     def _setReady(self):
+        """Called at the end of Operator._setupOutputs to set all outputs ready."""
         wasReady = self.ready()
 
         for p in self._subSlots:
@@ -1278,6 +1287,8 @@
 
         # If we just became ready...
         if not wasReady and self.meta._ready:
+            if self.logger:
+                self.logger.debug("Now ready. Notifying downstream slots.")
             # Notify downstream_slots of changed readystatus
             self._changed()
             self._sig_ready(self)
@@ -1338,6 +1349,14 @@
     def _changed(self):
         old_ready = self.ready()
         if self.upstream_slot is not None and self.meta != self.upstream_slot.meta:
+            if self.logger:
+                msg = (
+                    f"Copying meta."
+                    f" Ready: {self.meta._ready} -> {self.upstream_slot.meta._ready}."
+                    f" Shape: {self.meta.shape} -> {self.upstream_slot.meta.shape}."
+                    f" From {self.upstream_slot}. Previous: {self.meta}."
+                )
+                self.logger.debug(msg)
             self.meta = self.upstream_slot.meta.copy()
 
         if self._type == "output":
@@ -1377,6 +1396,8 @@
 
         """
         if self.operator is not None:
+            if self.logger:
+                self.logger.debug(f"Calling setupOutputs of {id(self.operator)}.")
             # check whether all slots are connected and notify operator
             self.operator._setupOutputs()