I'm confused with callback_groups and threads

asked 2023-07-18 06:35:53 -0600

SébastienL
32 ●3 ●10 ●11

Hi,

I want to get a value from a topic and after perform some operations on the value retrieved (can't be in the callback function).

I imagine a subscriber and this simple loop (python)

while self.__var is None:
    self.get_logger().info("__var has no value yet.")
    # wait

On my first attempt, I used time.sleep for the "wait", which pause the whole Node, including the subscriber if I don't misunderstand. Then I used a callback_group to handle the subscription, it didn't work either. I read that time.sleep should never be used and prefer rate.sleep instead, inside another thread. So i tried this

def retrieve_value(self):
    self.__var = None
    # Spin in a separate thread
    thread = threading.Thread(target=rclpy.spin, args=(self, ), daemon=True)
    thread.start()
    self.get_logger().info("thread starts.")

    rate = self.create_rate(1)
    self.__sub_var = self.create_subscription(
        String,
        f'/var_values',
        self.__set_var_value,  # simply self.__var = msg.data
        10,
        callback_group=self.__listener_cb_grp
    )
    while self.__var is None:
        self.get_logger().info("¤¤¤__var has not value yet.")
        rate.sleep()

    # rclpy.shutdown()  # the node crash if uncomment and the node seem to stop when commented
    self.get_logger().info("thread stops.")
    thread.join()

Last I used a MultiThreadedExecutor and it still did not work. Obviously I don't understand the thread management here.

Can someone help me ?

edit retag flag offensive close merge delete

Comments

Can you provide the full program context? How is retrieve_value being run?

jdlangs ( 2023-07-18 13:19:28 -0600 )edit

I'll post more code. I suspect I did not understand how ROS2 nodes are supposed to be used.

SébastienL ( 2023-07-19 03:28:32 -0600 )edit

add a comment

0

answered 2023-07-19 09:51:38 -0600

jdlangs
971 ●23 ●47 ●45

Looking at the additional info posted, I think your second design more-or-less makes sense. You should not use spin though because that's intended to move program control to callbacks only, and it only ends when the node shuts down. Instead, you can create the subscription and then in the while self.__var is None loop you can call spin_once on an executor. This will let callbacks run but return control to you afterwards. I don't think you need a reentrant callback group or a multi-threaded executor in this scenario.

edit flag offensive delete link

Comments

Thank you for the explanations. I tried to spin_once and it worked. I'll work with that on my real case.

SébastienL ( 2023-07-19 10:33:30 -0600 )edit

add a comment

0

answered 2023-07-18 16:04:23 -0600

slim71
18 ●3 ●6 ●8

Are you actually using rclpy.spin() in the spawned thread? If I'm getting this right, in the spawned thread you're trying to spin() once again the same node calling this function (as for the argument self). With the info provided, I'm not sure that will work.

I could be wrong, given I don't have the full picture, but... I don't actually see the point of using an explicit thread here. Even if the node you're coding can be both the subscriber and the publisher to the /var_values thread (in two different configurations), using the MultiThreadExecutor you can just add two nodes before spinning. Something like:

executor = MultiThreadedExecutor()
executor.add_node(sub_node)
executor.add_node(pub_node)
rclpy.spin()

Some useful stuff about executors and callback groups:

An open class by The Construct

A post on ROS Discourse

ROS2 documentation (humble only because that's what I'm using now)

As an additional information judging by the log print, the join() does not terminate the thread, it actually waits for its termination before going on.

edit flag offensive delete link

Comments

thanks for the answer, I tried indeed to spin again the same node.

I think sleeping on it make it clearer to me, but still unsolved. like @jdlangs suspected, to context on how the function is called is probably the key. -> i'll post more code after his reply

SébastienL ( 2023-07-19 03:26:57 -0600 )edit

add a comment

0

answered 2023-07-19 03:31:30 -0600

SébastienL
32 ●3 ●10 ●11

Here are some complement

The retrieve_value method is called in my __init__ function which is

def __init__(self, ):
    super().__init__(__class__.__name__)

    self.__var = None
    self.__listener_cb_grp = ReentrantCallbackGroup()
    self.retrieve_value()

    # I want __var set here
    self.get_logger().info(f"init: var = {self.__var}")

What I understand, is that the callback will be run only once the spin method is called, so calling a time.sleep in this init method makes this init never ends.

Let's say, my init defines only the callback group and move the retrieved+"use of __var" in another methods work

def __init__(self, ):
    super().__init__(__class__.__name__)

    self.__var = None
    self.__listener_cb_grp = ReentrantCallbackGroup()

def work(self,):
    self.get_logger().info("work IN !")
    self.retrieve_value()
    # I want __var set here
    self.get_logger().info(f"work: var = {self.__var}")
    self.get_logger().info("work OUT !")

And the main will init the node, spin the executor and last call the work method

node = TestCB()  # my node

exe = MultiThreadedExecutor()
exe.add_node(node)
exe.spin()

node.get_logger().info("init done, work now !")
node.work()

node.destroy_node()
rclpy.shutdown()

This time the node is stuck right after the init. I suspect everything has to be done in callbacks and methods can't be called by ROS2 design, is that right ?

edit flag offensive delete link

add a comment

I'm confused with callback_groups and threads

Comments

3 Answers

Comments

Comments

Question Tools

Stats

Related questions

I'm confused with callback_groups and threads edit

Comments

3 Answers

Comments

Comments

Question Tools

Stats

Related questions

I'm confused with callback_groups and threads