QoS - API and implemention for event callbacks

[emersonknapp]

Depends on ros2/rcl#408

Expose QoS event callbacks to the ROS2 Python client.

[emersonknapp]

For folks looking at this here's a demo app i've been using to try it out

import rclpy
# from rclpy.duration import Duration
from rclpy.executors import SingleThreadedExecutor
from rclpy.node import Node
from rclpy.publisher import PublisherEventCallbacks
# from rclpy.qos import QoSLivelinessPolicy
from rclpy.qos import QoSProfile
from rclpy.subscription import SubscriptionEventCallbacks

from std_msgs.msg import String


class Talker(Node):
    def __init__(self):
        super().__init__('Talker')
        qos = QoSProfile(
            # lifespan=Duration(seconds=0),
            # deadline=Duration(seconds=0),
            # liveliness=QoSLivelinessPolicy.RMW_QOS_POLICY_LIVELINESS_AUTOMATIC,
            # liveliness_lease_duration=Duration(seconds=2),
        )
        self.publisher = self.create_publisher(
            String, 'topic', qos_profile=qos,
            event_callbacks=PublisherEventCallbacks(
                deadline_callback=self.dead_cb,
                liveliness_callback=self.live_cb
            ))
        self.count = 0

        self.timer = self.create_timer(0.6, self.timer_cb)

    def fini(self):
        # Destroy the timer attached to the node explicitly
        # (optional - otherwise it will be done automatically
        # when the garbage collector destroys the node object)
        self.get_logger().info('Shutting down')
        self.destroy_timer(self.timer)
        self.destroy_node()

    def timer_cb(self):
        msg = String()
        msg.data = 'Hello World: %d' % self.count
        self.count += 1
        self.get_logger().info('Publishing: "%s"' % msg.data)
        self.publisher.publish(msg)

    def dead_cb(self, evt):
        self.get_logger().info('DEADLINE missed: "{}"'.format(evt))

    def live_cb(self, evt):
        self.get_logger().info('LIVELINESS failed: "{}"'.format(evt))


class Listener(Node):
    def __init__(self):
        super().__init__('Listener')
        self.subscription = self.create_subscription(
            String, 'topic', self.msg_cb,
            event_callbacks=SubscriptionEventCallbacks(
                deadline_callback=self.dead_cb,
                liveliness_callback=self.live_cb
            ))

    def msg_cb(self, msg):
        self.get_logger().info('Heard: "%s"' % msg.data)

    def dead_cb(self, evt):
        self.get_logger().info('DEADLINE missed: "{}"'.format(evt))

    def live_cb(self, evt):
        self.get_logger().info('LIVELINESS changed: "{}"'.format(evt))

def main(args=None):
    rclpy.init(args=args)

    executor = SingleThreadedExecutor()

    talker = Talker()
    listener = Listener()

    def killa():
        executor.remove_node(talker)
        talker.fini()

    killswitch = talker.create_timer(4.0, killa)  # NOQA

    executor.add_node(talker)
    executor.add_node(listener)

    executor.spin()

    talker.fini()
    rclpy.shutdown()


if __name__ == '__main__':
    main()

[emersonknapp]

@thomas-moulard I think this is ready (except for one last TODO that I am tracking down) it seems like you've got a bit of experience with C-Python interfaces, can you give this a once-over review before I open up the review to openrobotics?

[nburek]

One of the subtleties of default parameters in Python is that they are only evaluated once when the function is defined instead of every time the function is executed. This means that every call to create_publisher will use the same default PublisherEventCallbacks() object instance. The safe option is to use None as the default value and then construct a new instance of the object within the function itself when providing a default for mutable objects.

I don't think it breaks anything right now, so I wouldn't consider this a blocking issue. I wanted to bring it up anyways in case I'm missing something in how it's used.

[thomas-moulard]

+1

[emersonknapp]

Good catch - I'll default to None

[nburek]

Kind of a nitpick, but I think it would be better if there was a function in the PublisherEventCallbacks class that returns a list of QoSEventHandler objects for all the event handlers in the class. My reasoning is that if someone adds a new type of event handle in the future it is more obvious that they need to add it to the list in that class than it is this list in an unrelated file.

[emersonknapp]

I like that idea - will do

[nburek]

Same comment here as in the Publisher.

[thomas-moulard]

+1

[thomas-moulard]

callback_group type is Optional[CallbackGroup] as None is allowed.

[thomas-moulard]

I like this syntax, you don't need to use it but FYI

callback_group = callback_group or self.default_callback_group

[thomas-moulard]

ditto

[thomas-moulard]

Sounds like this could be split into multiple functions, this starts to be a bit long

[emersonknapp]

Going to implement as function on EventCallbacks to keep centralized - as per Nick's comment below

[thomas-moulard]

Use typed namedtuple instead: https://docs.python.org/3/library/typing.html#typing.NamedTuple

[thomas-moulard]

Order sounds wrong there, freeing the C underlying object before the python one?

[emersonknapp]

This is called automatically as the destructor of the allocated Capsule when it is destroyed by Python https://github.com/aws-robotics/rclpy/pull/6/files#diff-80dda03110b5be606b823d6b0558b5c2R1416

There isn't a Python object for me to free.

[emersonknapp]

Unless I have misunderstood your comment - and you mean event should be freed first - but I have to send event to fini before that, so its members can be freed - if I were to free event first I couldn't access its members any more to free them.

[mm318]

I think we both got the naming convention wrong. In rclcpp it's QOSEventHandler, here it's QoSEventHandler, but the Google style guide would've wanted QosEventHandler (https://google.github.io/styleguide/cppguide.html#General_Naming_Rules). Let's change it?

[emersonknapp]

The existing python codebase uses QoS (AaA) for the functionality that was already here. I'm going to keep it this way to match the existing convention.

See https://github.com/ros2/rclpy/blob/master/rclpy/rclpy/qos.py#L21

[thomas-moulard]

We should run ASAN on this PR before merging it, it's hard to get the Python ref-counting right.

[thomas-moulard]

Big function - can we split this?

[emersonknapp]

C is pretty verbose - do you really think this is actually more than one logical unit of code?

[thomas-moulard]

Here is a re-implementation of your function (coding style, passing precisely the args etc. is left as an exercise for the reader :)

bool _is_pycapsule_publisher(PyObject* pyobject) {
  return PyCapsule_IsValid(pyparent, "rcl_subscription_t");
}

bool _is_pycapsule_subscription(PyObject* pyobject) {
  return PyCapsule_IsValid(pyparent, "rcl_publisher_t");
}

rcl_event_t * _allocate_and_zero_initialize_rcl_event() {
  rcl_event_t * event = (rcl_event_t *)PyMem_Malloc(sizeof(rcl_event_t));
  if (!event) {
    PyErr_Format(PyExc_MemoryError, "Failed to allocate memory for event");
    return NULL;
  }
  *event = rcl_get_zero_initialized_event();
  return event;
}

// forcing yourself to divide your code in smaller functions will help build
// generally useful utility functions such as this one instead of considering
// the code "write-only".
bool _check_rcl_errors_status() {
  if (ret != RCL_RET_OK) {
    rcl_reset_error();
    PyErr_Format(PyExc_RuntimeError, rcutils_get_error_string());
    return true;
  } else {
    return false;
  }
}

bool _initialize_event_publisher(event) {
  const rcl_ret_t ret = rcl_subscription_event_init(event, subscription, pyevent_type);
  return _check_rcl_error_status();
}

bool _initialize_event_subscription(event) {
  const rcl_ret_t ret = rcl_subscription_event_init(event, subscription, pyevent_type);
  return _check_rcl_error_status();
}

bool _initialize_event(event) {
  if(_is_pycapsule_publisher(event)) {
    return _initialize_event_publisher();
  } else if(_is_pycapsule_subscription(event)) {
    return _initialize_event_subscription();
  }
}

static PyObject *
rclpy_create_event(PyObject * Py_UNUSED(self), PyObject * args) {
  const rcl_event_t * event = _allocate_and_zero_initialize_rcl_event();
  if (!event) {
    return NULL;
  }
  if (_initialize_event(event)) {
    return event;
  } else {
    PyMem_Free(event);  // Nesting functions will help you ensuring cleanup code is needed only ONCE.
    return NULL;
  }
}

[thomas-moulard]

Big function - can we split this?

[thomas-moulard]

This file looks huge, time to break into one file per python function or something? (at least partially split this)

[emersonknapp]

It is huge - but i'm kind of in favor of big code files for feature sets - single compilation unit usually builds faster than code with lots of headers and you always know where to find what you're looking for. Regardless - I'd be hesitant to break it out as part of this feature - that feels like its own dedicated change.

[thomas-moulard]

Let's have a single cleanup logic using Py_XDECREF on all variables to reduce the complexity of this code.

[emersonknapp]

ok - I'll restructure it a bit

[thomas-moulard]

Additionally, this PR needs unit tests. It can be python code, but it should call directly each bound symbol one by one.

[emersonknapp]

Working on adding some unit tests still - but @thomas-moulard do you like the look of it better this way? I tried to make the functions more readable, and I see the value in abstracting away some of the string-based functions, so that those potential mistakes can at least be centralized.

[emersonknapp]

Closing in favor of ros2#316