poll_* methods to support custom futures implementations

[dgrr]

No description provided.

[dgrr]

I'll implement the poll_* methods for the WebSocketRead & WebSocketWrite

[dgrr]

btw, if you implement Stream + Sink from futures you can call split. It's difficult for Stream to return a Frame<'_>, or maybe impossible

[littledivy]

Is this intentional to remove unstable-split? I would like to keep it

[dgrr]

Sure, I can put it back in

[dgrr]

well, the question is, why is it unstable?

[matheus23]

Would love to see this land. This library would become a great alternative to tokio-tungstenite and tokio-websockets in that case. Especially since it supports true, lock-free stream splitting.

[matheus23]

FWIW, it'd be nice to have the same poll_ treatment for FragmentCollectorRead 👀 😁

[matheus23]

btw, if you implement Stream + Sink from futures you can call split. It's difficult for Stream to return a Frame<'_>, or maybe impossible

This is true, but internally it means it'll simply create two references to the same thing using a BiLock. So you end up locking every time you use either the Stream or Sink.

Having a "true" split implementation that is lock free would be a lot cooler! :)

---

Also, while it's impossible to write an implementation of Stream<Item = Frame<'???>>, it's possible to map the Frame<'f> into your own domain-specific application struct that doesn't have a lifetime, e.g. by parsing binary and text frames into your own types, before returning them in your stream implementation as Items.

Just spelling this out for anyone reading this ✌️

[dgrr]

Having a "true" split implementation that is lock free would be a lot cooler! :)

How is this not lock free? If you use tokio::io::split it will use a Mutex internally.

Also, while it's impossible to write an implementation of Stream<Item = Frame<'???>>, it's possible to map the Frame<'f> into your own domain-specific application struct that doesn't have a lifetime, e.g. by parsing binary and text frames into your own types, before returning them in your stream implementation as Items.

Yes, that's what I do mostly. Because I want my code to be compatible with futures::Stream and Sink. I shouldn't but I had to code it in a bit of a rush 👀

[matheus23]

Having a "true" split implementation that is lock free would be a lot cooler! :)

How is this not lock free? If you use tokio::io::split it will use a Mutex internally.

Yeah no I'm agreeing with you and that's exactly what I'm trying to say.
You wrote "if you implement Stream + Sink from futures you can call split.", but that gives you worse performance than having the stream and sink already split from the start ✌️ (which this PR enables/makes easier!)

[dgrr]

Yeah no I'm agreeing with you and that's exactly what I'm trying to say. You wrote "if you implement Stream + Sink from futures you can call split.", but that gives you worse performance than having the stream and sink already split from the start ✌️ (which this PR enables/makes easier!)

Ah yes, but it is mainly for compatibility because sometimes it might be useful to use futures::StreamExt. Anyways, the user can implement it on their own, given that Frame<'_> cannot be returned as the Item in futures::Stream

[dgrr]

@matheus23 about poll in FragmentCollectorRead, what would you expect? Do you expect a callback also or that it returns early with the mandatory reply frame?

[matheus23]

Personally, I'd highly prefer returning the mandatory frame :)

[dgrr]

Personally, I'd highly prefer returning the mandatory frame :)

Indeed. Me too

[littledivy]

Thanks for the PR and sorry for slow replies, I hope you don't mind me taking some time here as this is a relatively big change :)

It seems there is a regression in echo_server benchmark with larger payloads:

# this pr
$ ./load_test 10 0.0.0.0 8080 0 0 102400
Msg/sec: 35997.750000
Msg/sec: 35021.000000

# main
$ ./load_test 10 0.0.0.0 8080 0 0 102400
Msg/sec: 42045.750000
Msg/sec: 42146.500000

(measured on an M1 macbook; similar numbers on x64 Linux server)

[matheus23]

Just as an FYI, I've been trying to get the benchmarks to compile & run here on my NixOS to help diagnose the regression, but I'm still fighting my way through with the linker. Will of course post once I got something.

[dgrr]

Just as an FYI, I've been trying to get the benchmarks to compile & run here on my NixOS to help diagnose the regression, but I'm still fighting my way through with the linker. Will of course post once I got something.

I managed to reproduce the benchmarks, not in macos in Linux, but I cannot find where the regression is

[conradludgate]

Using Vec is faster in my testing. To replicate buffer.advance(written) you can do buffer.splice(..written, [0u8; 0]). Because this is expensive what I found is better is to instead have a buf_pos: usize that I increment, and only in start_send do I run

self.buffer.splice(..self.buf_pos, [0u8; 0]);
self.buf_pos = 0;

before the fmt_head call to write the frame into the buffer. This is because BytesMut::advance is kinda expensive it seems.

[conradludgate]

Further testing shows that, for large buffers, removing the vectored write support is a significant source of the regressions. Now the difference it just 43500 Msg/sec compared to 44000 Msg/sec

Opened a PR against this PR 😅 dgrr#1

[dgrr]

@littledivy it seems that this PR is ok now thanks to the contributions of @conradludgate

[ShabbirHasan1]

@littledivy All checks have passed successfully 🚀. Could you please review and merge this PR at your convenience 🙏.

Thank you @dgrr and @conradludgate for your kind contributions in making this library even better. 🫡

[dgrr]

@littledivy