Performance improvements in datachange callbacks

[erogulenko]

Hi there,

With increasing number of nodes in a server and a lot of value changes in short times, the performance of the MonitoredItemService.datachange_callback starts to matter quite a bit.
I looked into the implementation and would like to suggest minor improvements regarding performance.

First and most importantly, when calling MonitoredItemValues.set_current_datavalue() a deepcopy is performed on the DataValue instance. However, the DataValue itself and its fields are frozen dataclasses and therefore immutable, if I'm not mistaken. Only the DataValue.Value.Value which is the user value can be a mutable object. So my suggestion here would be, to only execute the deepcopy if the DataValue.Value.Value is not an instance of an immutable object, and only executing it on the DataValue.Value.Value itself and creating a new DataValue manually. Something like the following:

class MonitoredItemValues:
    def __init__(self) -> None:
        self.current_dvalue: Optional[ua.DataValue] = None
        self.old_dvalue: Optional[ua.DataValue] = None
        self.__immutable_builtin_types = (int, float, complex, str, tuple, frozenset, bytes)

    def set_current_datavalue(self, cur_val: ua.DataValue):
        self.old_dvalue = self.current_dvalue
        # We need to clone the value, to prevent referencing the inner value
        # Clone only if type is not one of the builtin immutable types
        # Also deepcopy only the inner value and not the whole DataValue object for performance reasons
        if not isinstance(cur_val.Value.Value, self.__immutable_builtin_types):
            variant = ua.Variant(
                Value=copy.deepcopy(cur_val.Value.Value),
                VariantType=cur_val.Value.VariantType,
                Dimensions=cur_val.Value.Dimensions,
                is_array=cur_val.Value.is_array,
            )
            cur_val = ua.DataValue(
                Value=variant,
                StatusCode_=cur_val.StatusCode_,
                SourceTimestamp=cur_val.SourceTimestamp,
                ServerTimestamp=cur_val.ServerTimestamp,
                SourcePicoseconds=cur_val.SourcePicoseconds,
                ServerPicoseconds=cur_val.ServerPicoseconds,
            )
        self.current_dvalue = cur_val

    def get_current_datavalue(self) -> Optional[ua.DataValue]:
        return self.current_dvalue

    def get_old_datavalue(self) -> Optional[ua.DataValue]:
        return self.old_dvalue

The isinstance check increases the function time by a bit for actually mutable objects, however, it saves a lot when the values are immutable with respect to just calling deepcopy on the immutable value.

If it's not desired to do this kind of isinstance check, just deepcopying the DataValue.Value.Value instead of the DataValue already leads to quite substantial performance improvements here.

Secondly, a minor improvement, in the MonitoredItemService.datachange_callback, the MonitoredItemNotification is generated each callback, even if the deadband_flag_pass is not passed and no event is being enqueued. This can be improved by just instantiating the MonitoredItemNotification directly before the enqueuing of the event.

    async def datachange_callback(self, handle: int, value: ua.DataValue, error=None):
        if error:
            self.logger.info("subscription %s: datachange callback called with handle '%s' and error '%s'", self,
                             handle, error)
            await self.trigger_statuschange(error)
        else:
            # self.logger.info(f"subscription {self}: datachange callback called "
            #                 f"with handle '{handle}' and value '{value.Value}'")
            mid = self._monitored_datachange[handle]
            mdata = self._monitored_items[mid]
            mdata.mvalue.set_current_datavalue(value)
            if mdata.filter:
                deadband_flag_pass = self._is_data_changed(
                    mdata.mvalue, mdata.filter.Trigger
                ) and self._is_deadband_exceeded(mdata.mvalue, mdata.filter)
            else:
                # Trigger defaults to StatusValue
                deadband_flag_pass = self._is_data_changed(mdata.mvalue, ua.DataChangeTrigger.StatusValue)

            if deadband_flag_pass:
                event = ua.MonitoredItemNotification(ClientHandle=mdata.client_handle, Value=value)
                await self.isub.enqueue_datachange_event(mid, event, mdata.queue_size)

These are just two suggestions on where to improve. If I'm mistaken in any of my assumptions, feel free to correct me.

Does anyone have other ideas on where we could improve performance further?

With the code changes above, I ran the unit tests from v1.1.0, which seem to be working the same as before those changes. So if the changes are OK from your end, I could create a pull request.