Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix async_ usages for new uncancelable semantic #3063

Merged
merged 5 commits into from
Feb 20, 2023
Merged

Fix async_ usages for new uncancelable semantic #3063

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
db5b861
Bump CE
armanbilge Nov 22, 2022
a3ffb43
Fix JVM `SocketGroupPlatform`
armanbilge Nov 22, 2022
54d0189
Fix JS stream converters
armanbilge Nov 22, 2022
9e6a59b
Use intermediate `Channel` for accepting sockets
armanbilge Nov 22, 2022
1c0be5c
Bump base version
armanbilge Nov 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions build.sbt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,9 +208,9 @@ lazy val core = crossProject(JVMPlatform, JSPlatform, NativePlatform)
libraryDependencies ++= Seq(
"org.typelevel" %%% "cats-core" % "2.9.0",
"org.typelevel" %%% "cats-laws" % "2.9.0" % Test,
"org.typelevel" %%% "cats-effect" % "3.4.1",
"org.typelevel" %%% "cats-effect-laws" % "3.4.1" % Test,
"org.typelevel" %%% "cats-effect-testkit" % "3.4.1" % Test,
"org.typelevel" %%% "cats-effect" % "3.5-4b87497",
"org.typelevel" %%% "cats-effect-laws" % "3.5-4b87497" % Test,
"org.typelevel" %%% "cats-effect-testkit" % "3.5-4b87497" % Test,
"org.scodec" %%% "scodec-bits" % "1.1.34",
"org.typelevel" %%% "scalacheck-effect-munit" % "2.0.0-M2" % Test,
"org.typelevel" %%% "munit-cats-effect" % "2.0.0-M3" % Test,
Expand Down
5 changes: 3 additions & 2 deletions io/js/src/main/scala/fs2/io/ioplatform.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,8 +161,9 @@ private[fs2] trait ioplatform {
val end =
if (endAfterUse)
Stream.exec {
F.async_[Unit] { cb =>
writable.end(e => cb(e.toLeft(()).leftMap(js.JavaScriptException)))
F.async[Unit] { cb =>
F.delay(writable.end(e => cb(e.toLeft(()).leftMap(js.JavaScriptException))))
.as(Some(F.unit))
}
}
else Stream.empty
Expand Down
46 changes: 27 additions & 19 deletions io/jvm-native/src/main/scala/fs2/io/net/SocketGroupPlatform.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,17 +57,21 @@ private[net] trait SocketGroupCompanionPlatform { self: SocketGroup.type =>

def connect(ch: AsynchronousSocketChannel): F[AsynchronousSocketChannel] =
to.resolve[F].flatMap { ip =>
Async[F].async_[AsynchronousSocketChannel] { cb =>
ch.connect(
ip.toInetSocketAddress,
null,
new CompletionHandler[Void, Void] {
def completed(result: Void, attachment: Void): Unit =
cb(Right(ch))
def failed(rsn: Throwable, attachment: Void): Unit =
cb(Left(rsn))
Async[F].async[AsynchronousSocketChannel] { cb =>
Async[F]
.delay {
ch.connect(
ip.toInetSocketAddress,
null,
new CompletionHandler[Void, Void] {
def completed(result: Void, attachment: Void): Unit =
cb(Right(ch))
def failed(rsn: Throwable, attachment: Void): Unit =
cb(Left(rsn))
}
)
}
)
.as(Some(Async[F].delay(ch.close())))
}
}

Expand Down Expand Up @@ -105,16 +109,20 @@ private[net] trait SocketGroupCompanionPlatform { self: SocketGroup.type =>
): Stream[F, Socket[F]] = {
def go: Stream[F, Socket[F]] = {
def acceptChannel: F[AsynchronousSocketChannel] =
Async[F].async_[AsynchronousSocketChannel] { cb =>
sch.accept(
null,
new CompletionHandler[AsynchronousSocketChannel, Void] {
def completed(ch: AsynchronousSocketChannel, attachment: Void): Unit =
cb(Right(ch))
def failed(rsn: Throwable, attachment: Void): Unit =
cb(Left(rsn))
Async[F].async[AsynchronousSocketChannel] { cb =>
Async[F]
.delay {
sch.accept(
null,
new CompletionHandler[AsynchronousSocketChannel, Void] {
def completed(ch: AsynchronousSocketChannel, attachment: Void): Unit =
cb(Right(ch))
def failed(rsn: Throwable, attachment: Void): Unit =
cb(Left(rsn))
}
)
}
)
.as(Some(Async[F].delay(sch.close())))
Copy link
Member Author

@armanbilge armanbilge Nov 22, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's potentially some weird life-cycle stuff going on here: AFAIK the only way to cancel an on-going accept is to close the server socket. As the signature is roughly Resource[F, Stream[F, Socket]], where the server socket is tied to the outer resource, in theory you might expect to be able to cancel the inner stream and then start it again.

One way to fix this is to introduce a Channel.synchronous as an intermediary: that way, the stream can be canceled gracefully, without forcing to kill the server socket.

Thoughts?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's some related history here: #2300

}

def setOpts(ch: AsynchronousSocketChannel) =
Expand Down