WebSocket connected() crashing
deavmi opened this issue · 12 comments
Not all to sure what the cause of this is but I am getting the following error when calling connected() on the WebSocket instance:
vibe.http.websockets.WebSocket
core.exception.AssertError@../../../.dub/packages/taggedalgebraic-0.11.22/taggedalgebraic/source/taggedalgebraic/taggedunion.d(308): Attempting to get type StreamSocketSlot from a TaggedUnion with type typeof(null)
----------------
??:? _d_assert_msg [0xb0d3d0]
../../../.dub/packages/taggedalgebraic-0.11.22/taggedalgebraic/source/taggedalgebraic/taggedunion.d:308 inout pure nothrow ref @property @nogc @safe inout(eventcore.drivers.posix.sockets.StreamSocketSlot) taggedalgebraic.taggedunion.TaggedUnion!(eventcore.internal.utils.AlgebraicChoppedVector!(eventcore.drivers.posix.driver.FDSlot, eventcore.drivers.posix.sockets.StreamSocketSlot, eventcore.drivers.posix.sockets.StreamListenSocketSlot, eventcore.drivers.posix.sockets.DgramSocketSlot, eventcore.drivers.posix.dns.DNSSlot, eventcore.drivers.posix.watchers.WatcherSlot, eventcore.drivers.posix.events.EventSlot, eventcore.drivers.posix.signals.SignalSlot, eventcore.drivers.posix.pipes.PipeSlot).AlgebraicChoppedVector.U).TaggedUnion.value!(eventcore.drivers.posix.sockets.StreamSocketSlot).value() [0xadb79f]
../../../.dub/packages/taggedalgebraic-0.11.22/taggedalgebraic/source/taggedalgebraic/taggedalgebraic.d:767 pure nothrow ref @nogc @safe inout(eventcore.drivers.posix.sockets.StreamSocketSlot) taggedalgebraic.taggedalgebraic.get!(eventcore.drivers.posix.sockets.StreamSocketSlot, eventcore.internal.utils.AlgebraicChoppedVector!(eventcore.drivers.posix.driver.FDSlot, eventcore.drivers.posix.sockets.StreamSocketSlot, eventcore.drivers.posix.sockets.StreamListenSocketSlot, eventcore.drivers.posix.sockets.DgramSocketSlot, eventcore.drivers.posix.dns.DNSSlot, eventcore.drivers.posix.watchers.WatcherSlot, eventcore.drivers.posix.events.EventSlot, eventcore.drivers.posix.signals.SignalSlot, eventcore.drivers.posix.pipes.PipeSlot).AlgebraicChoppedVector.U).get(ref inout(taggedalgebraic.taggedalgebraic.TaggedAlgebraic!(eventcore.internal.utils.AlgebraicChoppedVector!(eventcore.drivers.posix.driver.FDSlot, eventcore.drivers.posix.sockets.StreamSocketSlot, eventcore.drivers.posix.sockets.StreamListenSocketSlot, eventcore.drivers.posix.sockets.DgramSocketSlot, eventcore.drivers.posix.dns.DNSSlot, eventcore.drivers.posix.watchers.WatcherSlot, eventcore.drivers.posix.events.EventSlot, eventcore.drivers.posix.signals.SignalSlot, eventcore.drivers.posix.pipes.PipeSlot).AlgebraicChoppedVector.U).TaggedAlgebraic)) [0xadb738]
../../../.dub/packages/eventcore-0.9.21/eventcore/source/eventcore/internal/utils.d-mixin-288:288 pure nothrow ref @property @nogc @safe eventcore.drivers.posix.sockets.StreamSocketSlot eventcore.internal.utils.AlgebraicChoppedVector!(eventcore.drivers.posix.driver.FDSlot, eventcore.drivers.posix.sockets.StreamSocketSlot, eventcore.drivers.posix.sockets.StreamListenSocketSlot, eventcore.drivers.posix.sockets.DgramSocketSlot, eventcore.drivers.posix.dns.DNSSlot, eventcore.drivers.posix.watchers.WatcherSlot, eventcore.drivers.posix.events.EventSlot, eventcore.drivers.posix.signals.SignalSlot, eventcore.drivers.posix.pipes.PipeSlot).AlgebraicChoppedVector.FullField.streamSocket() [0xad4de3]
../../../.dub/packages/eventcore-0.9.21/eventcore/source/eventcore/drivers/posix/sockets.d:309 nothrow @safe eventcore.driver.ConnectionState eventcore.drivers.posix.sockets.PosixEventDriverSockets!(eventcore.drivers.posix.epoll.EpollEventLoop).PosixEventDriverSockets.getConnectionState(eventcore.driver.StreamSocketFD) [0xac9efb]
../../../.dub/packages/vibe-core-1.22.5/vibe-core/source/vibe/core/net.d:579 const nothrow @property @safe bool vibe.core.net.TCPConnection.connected() [0xa78ece]
../../../.dub/packages/vibe-core-1.22.5/vibe-core/source/vibe/internal/interfaceproxy.d-mixin-312:312 const @property @safe bool vibe.internal.interfaceproxy.InterfaceProxy!(vibe.core.stream.ConnectionStream).InterfaceProxy.ProxyImpl!(vibe.core.net.TCPConnection).ProxyImpl.__mixin8.__mixin2.connected(const(void[])) [0x934b13]
../../../.dub/packages/vibe-core-1.22.5/vibe-core/source/vibe/internal/interfaceproxy.d-mixin-201:201 const @property @safe bool vibe.internal.interfaceproxy.InterfaceProxy!(vibe.core.stream.ConnectionStream).InterfaceProxy.__mixin22.__mixin2.connected() [0x907022]
../../../.dub/packages/vibe-d-0.9.5/vibe-d/stream/vibe/stream/wrapper.d:179 const @property @safe bool vibe.stream.wrapper.ConnectionProxyStream.connected() [0xa64ecc]
../../../.dub/packages/vibe-d-0.9.5/vibe-d/http/vibe/http/websockets.d:583 @property @safe bool vibe.http.websockets.WebSocket.connected() [0x9766f2]
source/dnostr/relays.d:66 bool dnostr.relays.NostrRelay.attemptConnection() [0x8380fc]
source/dnostr/relays.d:110 void dnostr.relays.NostrRelay.ensureConnected() [0x8383dc]
source/dnostr/relays.d:124 void dnostr.relays.NostrRelay.run() [0x8384e0]
??:? void core.thread.context.Callable.opCall() [0xb19608]
??:? thread_entryPoint [0xb190e6]
??:? [0x7fd9b09ce12c]
??:? [0x7fd9b0a4fbbf]
Error Program exited with code 1
This could be a case of the issue fixed in vibe-d/vibe-core#338 (1.22.7). If you are able to reproduce it, can you retry this after a dub upgrade?
I didn't call runApplication()... That fixed that I thimk but I have some other issues now. Going to investigate those because I can't yet tell if it is working.
Will report back to you soon - work just has been pumbling me down :)
Appreciate it - and keep the great work up!
Is there information I can read on what runApplication() sets up?
From what I see if I am just opening up a websocket as a client do I need to call runApplication()?
runApplication mainly just handles command line arguments and then calls runEventLoop. But since any blocking operation in vibe.d also creates a temporary event loop, there shouldn't be any reason why it should be necessary to call either runEventLoop or runApplication in general.
Another possibility where this error could happen would be that the underlying socket is used from a different thread than the one that created it. Apart from synchronization primitives, all objects in vibe.d are tied to their own thread. However, I'm unsure how runApplication could help in that case. But judging by the call stack above, this might indeed be the case here?
runApplication mainly just handles command line arguments and then calls runEventLoop. But since any blocking operation in vibe.d also creates a temporary event loop, there shouldn't be any reason why it should be necessary to call either runEventLoop or runApplication in general.
Agreed
Another possibility where this error could happen would be that the underlying socket is used from a different thread than the one that created it.
So there are two D threads being used in my application, the only time I have a mutex that I added in use is when I initialize a new WebSocket (I have an ensureConnected() which will lock the ws object, check if a connection exists, if not then open one and assign the new WebSocket back, if not return).
So yes there is a case whereby I use the I/O calls like waitForData() from another thread than the one who originally created the WebSocket object (I have confirmed this with threadIDs) - is this something that vibe.d doesn't support (I presume I should be doing it in some other manner then? - this is the first time I have used vibed in such a manner - normally I write my own network code but the WebSOcket functionality is very useful).
Regarding the server side of websockets, I was trying to figoure out the HTTPServerOption.distribute because I was wondering if I would be able to atleast then handle the spawning of incoming websockets per-thread and then only operate on them within said threads.
Like a sort of .accept() -> spawnThreadHandler(), but from what I see on the API that option no longer exists?
Yeah, the distribute option wasn't thread safe within D's shared realm, so it had to be removed, unfortunately. For anything that needs to scale seriously, the recommended approach is to spawn multiple processes and use HTTPServerOption.reusePort together with the same port to let the OS distribute requests among the processes (at least Linux does that in a fair manner). This will avoid the GC becoming an issue with high loads, as that will otherwise block all threads during the collection phase.
However, for moderate workloads, or if the GC is not used for per-request allocations, using multiple threads within a single process can of course still be done. The typical approach for this is to use runWorkerTaskDist to reverse the order of spawn and listen:
runWorkerTaskDist({
auto router = new URLRouter;
// ...
auto settings = new HTTPServerSettings;
settings.options |= HTTPServerOption.resusePort;
settings.port = 1234;
listenHTTP(settings, router);
});
This will create one HTTP listener per hardware thread and uses the same reusePort logic for distributing incoming connections. But in contrast to HTTPServerOption.distribute, each thread has its own callback delegate and associated data, so that there are no hidden data races.
Thank you very much @s-ludwig - this seems to be working. I migrated over to working with fibers and I think I now have the hang of it!
Is there a vibe-d forum for questions?
There is the official form at https://forum.rejectedsoftware.com/groups/rejectedsoftware.vibed/ (also reachable via NNTP), but it kind of died off during a time where I was very busy with other things and spam got a real problem (thousands of spam messages before any real message). Then after a server migration, the NNTP access also didn't work for some time, which didn't help either. Nowadays, I try to clean up spam from time to time, but still didn't find the time to implement better anti-spam measures, so it's still not ideal.
For this reason, I'd recommend to use the regular D forums, where others might also be able to provide input. I'm still very busy with other things to this day, so I'm not very active there currently. In doubt, just open an issue here (it may require one or more pings sometimes).
I noticed the forums were overrun with people asking them to write their essays for them 😂😂
I tried signing up but didn't receive an email. I will try asking on the D forums.
My followup question pertained to the following:
I know have a server which has handlers for each new client socket (connection) coming in.
The hanlder does some work then yields (the fiber), what I then do is I loop through all other fibers (saved to an array) and resume them to allow them to progress some more).
However, the problem I face with this is that it is done once after a new connection (basically requiring more connections).
So, I was wondering if there was a way to emulate perhaps .accept(), calling a hanldler and then yielding again, but coupling it with something like canAccept() (some function returning true if the kernel has new socket connections to dequeue).
Any guidance will be greatly appreciated (I have never used a network framework before, always used primitives, fibers are also new but I have the hang of it).
I think I will close this now, I have posted this as a question on the D forums.
Thanks for the help and explanations @s-ludwig !
https://forum.dlang.org/post/mhtirtezsfdryrzadear@forum.dlang.org