Blocking subscriber callback and message buffer size

[chengguizi]

I realise that in Python API, the subscriber class will call the callback in a new thread whenever a measurement is received. I could see my callback actually create 8 thread in the background processing the in parallel.

Could I control the behaviour such that the callback is blocking? Namely, the callback will not be called concurrently, as the next callback is only run after the previous one returns. Also at the same time, a subscruber message buffer, pretty much like ROS API would be useful. This allow us to compromise between real-time performance and frame drops etc.

Reference:
https://continental.github.io/ecal/_api_python/ecal.core.subscriber.html

[chengguizi]

Or alternatively, an API of controlling how many thread the subscriber callback could use.

[KerstinKeller]

HI @chengguizi ,

eCAL will internally create one thread per subscriber topic (plus some extra, not 100% sure).
Your callbacks will be blocking, e.g. if you spend tooo much time in the callback and multiple messages arrive in the meantime, you will loose all but the most recent one. There is no internal eCAL queue.

However, callbacks for multiple topics may run concurrently, so you will have to synchronize mutliple callbacks yourself.

However, maybe in your usecase, using the synchronous "receive" function is a better fit. You can poll for data and also give a timeout (to see if you want to block or not).

[chengguizi]

Hi @KerstinKeller Thanks for your quick reply.

Now I understand callback runs at single thread. I went to study a bit more. I realise the threading starts once i start doing imshow within ecal_core.ok() while loop. And the thread does not go away even if the callback stop receive new messages.

1. When the script first start, it is 1 thread
2. once image is being received, and cv.imshow start working, the threading increase to like 12 (the number of CPU thread on my machine).

This really puzzles me. Could you spot a mistake here?

#!/usr/bin/env python3

import sys
import time

import capnp
import numpy as np
import cv2

import ecal.core.core as ecal_core
from byte_subscriber import ByteSubscriber

imshow_map = {}

def callback(topic_name, msg, ts):

    # need to remove the .decode() function within the Python API of ecal.core.subscriber StringSubscriber
    print("callback called")
    mat = np.zeros((800,600))
    imshow_map["mono8"] = mat
       


def main():  

    print("eCAL {} ({})\n".format(ecal_core.getversion(), ecal_core.getdate()))
    
    # initialize eCAL API
    ecal_core.initialize(sys.argv, "test_image_sub")
    
    # set process state
    ecal_core.set_process_state(1, 1, "I feel good")

    # create subscriber and connect callback
    sub = ByteSubscriber("S0/camd")
    sub.set_callback(callback)
    
    # idle main thread
    while ecal_core.ok():
        for im in imshow_map:
            cv2.imshow(im, imshow_map[im])
        cv2.waitKey(3)
        
    
    # finalize eCAL API
    ecal_core.finalize()

if __name__ == "__main__":
    main()

[chengguizi]

My bad, seems I could reproduce the problem without eCAL at all

import numpy as np
import cv2

imshow_map = {}


def main():

    while True:
        mat = np.zeros((800,600))
        imshow_map["mono8"] = mat


        cv2.imshow("mono8", imshow_map["mono8"])
        if cv2.waitKey(3) == ord('q'):
            break



if __name__ == "__main__":
    main()

With this script, I have all my cpu cores spinning for no reasons. Not a eCAL related issue.