Seeing Delays of up to 10-15 seconds from emit function on tornado server.

[AShriki]

Hello,
I'm using a socketio server to transmit a 1mb video file and a ~500 byte playlist file from a socketio server to a client (via a successful web socket connection) on the same LAN. I'm seeing sporadic delays of up to ~15 seconds with all of the data from multiple emit calls being transmitted after the delay. In effect the network transactions (data from each event) are building up and then released between 1-4 at a time with ms in between each transaction (this has been verified with tcpdump). The desired behavior is to have a video file transmitted every 6 seconds. I've been scratching my head at this for a while now.
Help would be greatly appreciated!

The video directory is watched by watchdog and the server socketio logs show the following at 6 second intervals:

emitting event "file_transfer" to all [/video]
INFO:socketio.server:emitting event "file_transfer" to all [/video]
0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data 52-/video,["file_transfer",{"video":{"_placeholder":true,"num":0},"playlist":{"_placeholder":true,"num":1}}]
INFO:engineio.server:0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data 52-/video,["file_transfer",{"video":{"_placeholder":true,"num":0},"playlist":{"_placeholder":true,"num":1}}]
0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data <binary>
INFO:engineio.server:0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data <binary>
0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data <binary>
INFO:engineio.server:0HmKJndVO5rjuWkoAAAA: Sending packet MESSAGE data <binary>

Initialization code:

    sio = socketio.AsyncServer(logger=True,engineio_logger=True,async_mode='tornado',async_handlers=True,max_http_buffer_size=3000000)
    sio.register_namespace(videoNamespace(namespace='/video'))
    application = tornado.web.Application([
        (r"/data", DataHandler, {"shared":shared}),
        (r"/unsent", UnsentHandler, {"shared":shared}),
        (r"/socket.io/", socketio.get_tornado_handler(sio)),
    ])
    server = tornado.httpserver.HTTPServer(application, max_buffer_size=268435456) 
    server.listen(int(env_var["base_core_port"]))
    IOLoop.current().start()

namespace class:

class videoNamespace(socketio.AsyncNamespace):
    def __init__(self, namespace: str = None):
        super().__init__(namespace)
        self.streamer = Streamer()
        manager = multiprocessing.Manager()
        self.last_altered = manager.Value(c_float,0.0)
        self.exit_flag = manager.Value(c_bool,False)

    def on_connect_error(self,*args):
        logger.info("The connection failed!")

    async def on_set_cameras(self,event,cameras: dict):
        logger.info(f"received cams: {cameras}, {event}")
        loop = asyncio.get_running_loop()
        executor = concurrent.futures.ThreadPoolExecutor()
        await loop.run_in_executor(executor, self.streamer.set_cameras, cameras)
        return 'OK'

    async def on_send_directory_changes(self, event, cam_num):
        logger.info(f"received cams: {cam_num}, {event}")
        loop = asyncio.get_running_loop()
        executor = concurrent.futures.ThreadPoolExecutor()
        await loop.run_in_executor(executor, video_file_watcher, cam_num, self.last_altered, self.exit_flag, self)
        return 'DONE'

    def on_connect(self, *args):
        logger.info(f'client connected to local cameras')

    def on_catch_all(self, event=None, data=None):
        logger.info(f"oops :::: {event},{data}")

    async def on_disconnect(self, *args):
        logger.error(f'client disconnected')
        self.exit_flag.value = True
        time.sleep(0.25)
        self.exit_flag.value = False

Associated call to video_file_watcher and the watchdog PatternMatchingEventHandler

class ChangeHandler(PatternMatchingEventHandler):
    def __init__(self, obs: Observer, last_altered, exit_flag, cam_num: int, sio: socketio.AsyncNamespace):
        self.obs = obs
        self.sio = sio
        self.cam_num = cam_num
        self.last_ts = None
        super().__init__(ignore_directories=True)
        self.last_altered = last_altered
        self.exit_flag = exit_flag

    def on_closed(self, event: FileClosedEvent):
        try:
            if "m3u8" in event.src_path and os.path.exists(event.src_path[:-4]):
                ts_size = 0
                with open(self.last_ts,'rb') as f1:
                    ts_size = os.fstat(f1.fileno()).st_size
                    with open(event.src_path[:-4],'rb') as f2:
                        f_size = os.fstat(f2.fileno()).st_size
                        payload = {'video':f1.read(),'playlist':f2.read()}
                        # emit happens here <---
                        asyncio.run(self.sio.emit('file_transfer', data=payload, namespace='/video'))
                        logger.info(f"sent {self.last_ts}: {ts_size}, {event.src_path[:-4]}: {f_size}")
                self.last_altered.value = time.time()
            elif "ts" in event.src_path:
                self.last_ts = event.src_path
        except Exception as e:
            # stop emitting events
            self.obs._clear_emitters()
            logger.info(f"socket send exception from change handler: {e}\n")
            logger.info(traceback.print_exc())
            self.exit_flag.value = True

def video_file_watcher(cam_num: int, last_altered, exit_flag, sio: socketio.AsyncNamespace):
    try:
        logger.info("Watcher spawned")
        obs = Observer()
        change_handler = ChangeHandler(obs, last_altered, exit_flag, cam_num, sio)
        watch_dir = env_var["live_video_path"] + str(cam_num) + '/'
        obs.schedule(change_handler, watch_dir)
        obs.start()
        logger.info("work loop starting")
        while True:
            if exit_flag.value:
                obs._clear_emitters()
                raise Exception(f"Camera scheduled shutdown with exit_flag:{exit_flag.value}")
    except Exception as e:
        logger.error(f"video_watcher exception: {e}")
    finally:
        exit_flag.value = False
        logger.info(f"Got stop for watcher with pid {os.getpid()}")
        try:          
            for f in glob.glob(env_var["live_video_path"] + str(cam_num)+ '/' + 'stream.*'):
                os.remove(f)
            logger.info(f"artifacts left over: {obs.event_queue.unfinished_tasks}")
            obs.stop()
            logger.info("Watchdog stopped")
            obs.join()
            logger.info("done")
        except Exception as e:
            logger.error(f"Error during cleanup: {e}")

[miguelgrinberg]

Are all the libraries that you are using compatible with asyncio? The most likely cause of your delays is that your application is blocking, which is not related to socket.io.

[AShriki]

Thanks Miguel. The issue ended up being that I was emitting events into the wrong async loop. The solutions was to pass the loop to the thread process and use it in the handler to emit.