feat: dynamic frame skipping

[eliteprox]

This change adds a queue in ComfyStreamClient to dynamically skip inference on frames to acheive maximum FPS

• Queue size: 50 frames
• When queue is full, the last frame is returned, and one frame is removed from the stack to ensure the next frame is processed without delay.

Dynamically uses as many FPS as possible, however, due to last frame being returned the output FPS will mirror the source

[yondonfu]

I agree taht this architecture is preferable to the previous one - just left some comments.

[yondonfu]

I'm not sure that you need two queues here - I think you can just use a single asyncio.Queue to manage the buffer of input frames since you're not consuming anything from the output queue right now.

If the queue is full, the last frame is returned thereby also dropping the current input frame. Otherwise, the input frame is added to the queue and the processor task can handle it.

[eliteprox]

@yondonfu See my comment below about two queues #29 (comment). Should we proceed with merge?

[yondonfu]

This runs an infinite loop but we probably want some way for it to exit cleanly i.e. when the event loop closes. I think you can catch an asyncio.CancelledError and just re-raise the error. For testing, worth adding a print statement for debugging to see if the loop is actually exiting cleanly.

[eliteprox]

Added handling for CancelledError which will break the loop, while allowing for other errors to continue

comfystream/src/comfystream/client.py

Lines 43 to 50 in a070e88

	except Exception as e:
	logger.error(f"Error processing queue item: {str(e)}")
	if output_fut and not output_fut.done():
	output_fut.set_exception(e)
	self.output_queue.task_done()
	except asyncio.CancelledError:
	logger.info("Stopped frame processor loop")
	raise

[yondonfu]

Reviewing this makes me realize there is probably a problem with how locking works right now. I'm not that familiar with the details of that at the moment and whether the lock usage is truly needed, but AFAICT given the current structure there can only be a single queue_prompt() call at a time meaning that appending to the input queue will not happen until the previous queue_prompt() call completes. I don't think that is the desired behavior...you probably want to keep appending to the queue (until its full) even if the previous queue_prompt() is still awaiting the output of an input frame.

If you just want to enable frame skipping then I suggest addressing that in a separate PR.

[eliteprox]

@yondonfu That's correct, I believe this is why the solution only works with two queues:

1. Incoming frames are appended to the input queue, but we do not wait for it to return. This allows for a buffer of incoming frames that get dropped when the queue fills up:
   

   comfystream/src/comfystream/client.py

   Line 63 in ef67569

   self.input_queue.append(output_fut)

Then, we must await the output queue for the result, since we need to return a frame from _process_queue

comfystream/src/comfystream/client.py

Line 69 in ef67569

await self.output_queue.put((input, output_fut))

. If we use a single queue, I believe we would have to await the input and output

If we want to address this differently, I think it would affect how apps are implementing ComfyStreamClient, which could be a breaking change. Are you fine with addressing the duplicate queues in a separate PR after we implement release tagging?

[eliteprox]

Closing in favor of #10