Coroutines option in place of threads for SSE

[zigster64]

trafficstars

Am in the early stages of thinking this one through

Running a service on the cheapest possible VPS for the job. Want to be able to handle a large number of open SSE connections

Works fine now using threads, but the overheads can be high as the number of connections grows (say max 100k connections on a single core shared VPS). I can handle 2000 connections with threads now, but it does thrash the machine a bit. Wouldn’t want to push higher than that using threads.

Looking at what’s involved in using something like Xcoro or a libexv custom event loop maybe in place of threads. In theory it should handle it lots of connections with less memory at least.

It’s SSE only so the traffic is outgoing only .. with infrequent dribs and drabs, so no need to listen for incoming data on the open connection yet.

Non blocking IO on the output should be fine. Just need an extra channels like mechanism for the core engine to communicate with the SSE coroutine, and that should handle the whole job I think.

Will do a proof of concept first that enables the http.zig user to provide their own custom event loop or handler to take over SSE connections, and then tackle the channels issue.

Not sure exactly where to start with this (xcoro isnt well maintained?) .. but it should be fun. Might be useful to add to http.zig if it works well.

Opening an issue here as a discussion point. Thx

[karlseguin]

It's pretty opinionated that startEventStream launches a thread. In retrospect, I don't understand why I didn't just make it return the std.net.Stream:

https://github.com/karlseguin/http.zig/blob/cb760afdb50b1e4b3f37570e6ddbbd04bf9d5c32/src/response.zig#L126

changed to

self.disown();
return stream;

Then you can manage the stream however you want. Depending on what you're doing, a single thread for all the streams, or a small thread pool, might be the simplest solution.

[zigster64]

I think the reasoning iirc was to present the connection as a blocking io stream, because that seems appropriate for the most common case.

.. well, most common at the time anyway :)

I will try it with your 2 line suggestion. I think the stream at that point will already be in non blocking mode ? Is that correct ?

Agree - was thinking a single worker thread to hold all sse connections, with a write buffer for each connection. Then a single channel in the thread to receive “events” that trigger more output. But again, that’s going to be quite opinionated as well, so all that logic belong in the app rather than the lib.

[karlseguin]

Ya, it's forced into blocking mode a few lines up.

[zigster64]

cool

Yeah, I just take out the blocking mode transition now, write the header to start the SSE connection (in non-blocking mode), then hand back the stream to the caller. That seems to work fine.

Trying with 2 different handler funcs

1 - Just hangs on to the original thread, and slowly outputs to the connection from there. (no new thread spawned) 2 - (WIP) on establishing a new SSE connection, will pass it to a co-routine. Not working yet

With 1) - it's actually pretty efficient, if I pre-open a silly amount of threads on boot. Just for a test - im pre-opening 4k threads, and it's not too bad at all. About 100kb memory per thread, and about 1% CPU per 100 new connections (M2 air). That's usable up to around 3000 connections all at the same time.

(Which I think is more about my Mac Air config having a hard upper limit of 6000 threads - so 3000 for the web server, and 3000 curl jobs running in the background :) )

I will pick up the co-routine attempt next, after a break. That stuff is out of scope with http.zig then

Ive posted a minimal PR for adding a new SSE variant function startEventStreamSync() whilst keeping the old one. Up to you ...

[zigster64]

Test / Hacking app that I will be adding to if you are interested

https://github.com/zigster64/datastar-zig-train

Will get back into this later - but what Im going to try is to use http.zig as the HTTP engine, and use tardy as the coroutine + channel driver (Tardy being the async lib that was derived from the development of zzz)

So when it works, it will be a Frankenstein mash up of both http.zig and zzz, cooperating over the same sockets. Ha !

[0x546F6D]

Yeah, I just take out the blocking mode transition now, write the header to start the SSE connection (in non-blocking mode), then hand back the stream to the caller. That seems to work fine.

Any way to not automatically finish the header with \r\n\r\n before handing back the stream? In case more headers are required for this SSE connection, such as Content-Encoding: br\r\n.

[karlseguin]

I can add an option to startEventStream and startEventStreamSync for this, but can I ask why you can't use res.headers.add(...) before calling startEventStream ?

[0x546F6D]

I can add an option to startEventStream and startEventStreamSync for this, but can I ask why you can't use res.headers.add(...) before calling startEventStream ?

Brotli encoding can fail if the size of the message you want to stream is too small. So since I might not know what the size of the sse response will be, I only encode the response if it is above a certain treshold.

But you are right, since I only use brotli on short lived sse connections, I believe that I should be able to do all the necessary processing before end and use res.header.add(...) at that time before calling startEventStream.

[zigster64]

Suggest we close this as it’s no longer relevant after writergate changes coming up