Brain dump

[mmienko]

No description provided.

[djspiewak]

Sorry this took me so long to dig through! This is awesome.

[djspiewak]

Ah! Hmm. We may need to explicitly set this, then. If we don't have auto read set to false on the server socket, then we won't be able to backpressure new connections.

[mmienko]

Yeah for sure. I discovered perchance when digging through Netty source and saw a scheduled task for autoread. Thought that was weird, so made a note to investigate. We can just print Netty config during runtime. Best case, our setting is always accepted and this is false concern. Worse case, we have to schedule an autoread override at the right time so that it executes after Netty's 🙃 .

[djspiewak]

That would be abjectly horrible. :-D Like complain to Norman on Twitter horrible.

[mmienko]

haha will test it out :)

[djspiewak]

Seriously? What's replacing it? The impact this had was substantial.

[mmienko]

According to Epoll source this doesn't even do anything. For KQueue and NIO ELG this has a meaningful impact. As to what's replacing Epoll's method...🤷‍♂️ ha. So it's not really a universal interface for ELG's, but Netty makes it seem like it is.

One takeaway is that this creates a discrepancy between tuning on non-linux OS'es vs. other OS'es.

[djspiewak]

Oh this is really interesting!

I'm not that fussed about the tuning being a bit different on different OSes, though if it starts getting to the point where people might actually take that into account and build their applications and profiling to those kinds of parameters, then it might be an issue. At some point, differences between platforms are very much unavoidable. Like if you're profiling kqueue and using it to tune a server running against epoll or (even worse) io_uring, you're going to have a bad time with or without our meddling.

[mmienko]

👍

[djspiewak]

Are there Netty methods for doing this kind of detection?

[mmienko]

For example,

import io.netty.channel.epoll.Epoll
import io.netty.channel.kqueue.KQueue
if (Epoll.isAvailable) {epoll elg} else if (KQueue.isAvailable) {kqueue elg} else {nio elg}

// Don't see any the incubator package in current Netty version, but presumably it would look similar to above

Epoll source code for more details.

[djspiewak]

So. Much. Nicer. Let's do this.

[djspiewak]

👍

[djspiewak]

IMO we should have a Pipeline[I, O] which basically just consists of a bunch of handlers which will be added to the pipeline together with a final stage which just takes the messages and shovels them into a Queue using Dispatcher, the same way the current SocketHandler does with ByteBuf. We then generalize Socket to something like Socket[F, I, O], where the default is Socket[F, Byte, Byte] if you don't supply a pipeline.

[mmienko]

Yup, that all makes sense.

This requirement is too strong since we need to allow pipeline mutations, e.g. HTTP to WS. We're back to exposing all of Netty again since users have access to handlers.

One thing to think about is Netty's custom events. Specifically IdleState events, for detecting lack of inbound or outbound channel activity. So this would imply another type Pipeline[I, O, U] where U is some Netty User triggered event. Or it's a custom ADT and users would also need to supply a mapping Any => Either[Throwable, U]. Basically this would interface with override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit in Netty handlers. Then it could be another stream on Socket, i.e. events: Stream[F, U].

[djspiewak]

I like the idea of having some nicer wrappers around the existing pipeline handlers in these areas. So exactly what you're saying here, I think. :-) We should still allow users to wrap their own arbitrary handlers if they need to, but nice experiences out of the box are important.

[mmienko]

yeah exactly, there's the core and some user-friendly default pipelines to make people's lives easier.

[djspiewak]

Pretty much exactly this!

[djspiewak]

I think it can actually be a bit more general than this! It would be something like Socket[F, HttpRequest, HttpResponse]. Which really just boils down to a pair of two things: Stream[F, HttpRequest] and Pipe[F, HttpResponse, Unit]. So to write to the socket, you need to supply a Stream[F, HttpResponse], which you then feed to that pipe. To read from the socket, you simply consume from the stream of requests. Note that your Stream[F, HttpResponse] is probably a transformation of Stream[F, HttpRequest]. Thus, if you have a function HttpRequest => F[HttpResponse], you can get the obvious semantic really easily:

val server: Stream[F, Socket[F, HttpRequest, HttpResponse]] = Network[F].server(host, port, httpPipeline) // or something

val handler: HttpRequest => F[HttpResponse] = ???

val handles = server map { (socket: Socket[F, HttpRequest, HttpResponse]) =>
  socket.reads
    .evalMap(handler)
    .through(socket.writes)
    .onError(e => logF("caught exception during request", e))
    .attempt
    .void
}

handles.parJoin(8196)     // limit to 8196 concurrent connections

Does that kind of make sense? Basically we're getting a lot from Stream for free here.

[mmienko]

Yes it does, I think.
So at a high level?

object ExampleHttpServer extends IOApp {
  private[this] val router = Kleisli[IO, FullHttpRequest, FullHttpResponse] {
    request => IO {
          new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1,
            HttpResponseStatus.NOT_FOUND
          )
        }
  }
  
  override def run(args: List[String]): IO[ExitCode] =
    HttpServer.start[IO](HttpServer.HttpConfigs(
      requestTimeoutPeriod = 500.milliseconds,
      HttpServer.HttpConfigs.Parsing.default
    ))
      .evalMap { httpClientConnections =>
        httpClientConnections
          .map(_.successfullyDecodedReads(router))
          .parJoin(65536)
          .compile
          .drain
      }
      .useForever
      .as(ExitCode.Success)
    
}

and under the hood, the overly specific successfullyDecodedReads method is the handles in your example. The HttpServerConnection in this PR (which I'll now call HttpClientConnection) will handle piping Stream[F, HttpRequest] through Pipe[F, HttpResponse, Unit] via Kleisli[F, FullHttpRequest, FullHttpResponse] transform.

[djspiewak]

😳 This is cool.

[mmienko]

haha and to clarify, "user" is the higher layer sending a close frame via the FS2-Netty Streams.

[djspiewak]

So this is actually where things will get SUPER awesome, because for websockets, I think we'll just expose a Socket[F, WebSocketFrame, WebSocketFrame]. So in other words, a socket where you can send and receive websocket frames, just as how without any pipeline handlers we get a socket where you can send and receive bytes, and with the http handler we get a socket where you can send and receive HttpRequest/HttpResponse, respectively.

[mmienko]

Yup