From 5462496433c276c8656a6509321b1d42afb95097 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Sat, 23 Dec 2023 20:25:16 +0000 Subject: [PATCH 1/2] Remove `SuspendedStream`, `MicrotaskExecutor` hacks in Node.js `Process` --- io/js/src/main/scala/fs2/io/ioplatform.scala | 77 ++++++------- .../fs2/io/process/ProcessesPlatform.scala | 105 ++++++++---------- 2 files changed, 83 insertions(+), 99 deletions(-) diff --git a/io/js/src/main/scala/fs2/io/ioplatform.scala b/io/js/src/main/scala/fs2/io/ioplatform.scala index 3bf0158b5b..055c78d437 100644 --- a/io/js/src/main/scala/fs2/io/ioplatform.scala +++ b/io/js/src/main/scala/fs2/io/ioplatform.scala @@ -57,7 +57,36 @@ private[fs2] trait ioplatform { def suspendReadableAndRead[F[_], R <: Readable]( destroyIfNotEnded: Boolean = true, destroyIfCanceled: Boolean = true - )(thunk: => R)(implicit F: Async[F]): Resource[F, (R, Stream[F, Byte])] = { + )(thunk: => R)(implicit F: Async[F]): Resource[F, (R, Stream[F, Byte])] = + Resource + .makeCase { + F.delay { + val readable = thunk + (readable, unsafeReadReadable(readable)) + } + } { + case ((readable, _), Resource.ExitCase.Succeeded) => + F.delay { + if (!readable.readableEnded & destroyIfNotEnded) + readable.destroy() + } + case ((readable, _), Resource.ExitCase.Errored(_)) => + // tempting, but don't propagate the error! + // that would trigger a unhandled Node.js error that circumvents FS2/CE error channels + F.delay(readable.destroy()) + case ((readable, _), Resource.ExitCase.Canceled) => + if (destroyIfCanceled) + F.delay(readable.destroy()) + else + F.unit + } + .adaptError { case IOException(ex) => ex } + + /** Unsafely creates a `Stream` that reads all bytes from a `Readable`. + */ + private[io] def unsafeReadReadable[F[_]]( + readable: Readable + )(implicit F: Async[F]): Stream[F, Byte] = { final class Listener { private[this] var readable = false @@ -118,45 +147,17 @@ private[fs2] trait ioplatform { go.streamNoScope } - } - Resource - .eval(F.delay(new Listener)) - .flatMap { listener => - Resource - .makeCase { - F.delay { - val readable = thunk - readable.on("readable", () => listener.handleReadable()) - readable.once("error", listener.handleError(_)) - readable.once("end", () => listener.handleEnd()) - readable - } - } { - case (readable, Resource.ExitCase.Succeeded) => - F.delay { - if (!readable.readableEnded & destroyIfNotEnded) - readable.destroy() - } - case (readable, Resource.ExitCase.Errored(_)) => - // tempting, but don't propagate the error! - // that would trigger a unhandled Node.js error that circumvents FS2/CE error channels - F.delay(readable.destroy()) - case (readable, Resource.ExitCase.Canceled) => - if (destroyIfCanceled) - F.delay(readable.destroy()) - else - F.unit - } - .fproduct { readable => - listener.readableEvents.adaptError { case IOException(ex) => ex } >> - Stream.evalUnChunk( - F.delay(Option(readable.read()).fold(Chunk.empty[Byte])(Chunk.uint8Array(_))) - ) - } - } - .adaptError { case IOException(ex) => ex } + val listener = new Listener + readable.on("readable", () => listener.handleReadable()) + readable.once("error", listener.handleError(_)) + readable.once("end", () => listener.handleEnd()) + + listener.readableEvents.adaptError { case IOException(ex) => ex } >> + Stream.evalUnChunk( + F.delay(Option(readable.read()).fold(Chunk.empty[Byte])(Chunk.uint8Array(_))) + ) } /** `Pipe` that converts a stream of bytes to a stream that will emit a single `Readable`, diff --git a/io/js/src/main/scala/fs2/io/process/ProcessesPlatform.scala b/io/js/src/main/scala/fs2/io/process/ProcessesPlatform.scala index f6eebec00d..dc54026941 100644 --- a/io/js/src/main/scala/fs2/io/process/ProcessesPlatform.scala +++ b/io/js/src/main/scala/fs2/io/process/ProcessesPlatform.scala @@ -26,8 +26,6 @@ package process import cats.effect.kernel.Resource import cats.effect.kernel.Async import cats.syntax.all._ -import fs2.io.internal.MicrotaskExecutor -import fs2.io.internal.SuspendedStream import fs2.io.internal.facade import scala.scalajs.js @@ -36,27 +34,50 @@ import scala.scalajs.js.JSConverters._ private[process] trait ProcessesCompanionPlatform { def forAsync[F[_]](implicit F: Async[F]): Processes[F] = new UnsealedProcesses[F] { def spawn(process: ProcessBuilder): Resource[F, Process[F]] = - Resource - .make { - F.async_[facade.child_process.ChildProcess] { cb => - val subprocess = facade.child_process.spawn( - process.command, - process.args.toJSArray, - new facade.child_process.SpawnOptions { - cwd = process.workingDirectory.fold[js.UndefOr[String]](js.undefined)(_.toString) - env = - if (process.inheritEnv) - (facade.process.env ++ process.extraEnv).toJSDictionary - else - process.extraEnv.toJSDictionary + Resource { + F.async_[(Process[F], F[Unit])] { cb => + val childProcess = facade.child_process.spawn( + process.command, + process.args.toJSArray, + new facade.child_process.SpawnOptions { + cwd = process.workingDirectory.fold[js.UndefOr[String]](js.undefined)(_.toString) + env = + if (process.inheritEnv) + (facade.process.env ++ process.extraEnv).toJSDictionary + else + process.extraEnv.toJSDictionary + } + ) + + val fs2Process = new UnsealedProcess[F] { + + def isAlive: F[Boolean] = F.delay { + (childProcess.exitCode eq null) && (childProcess.signalCode eq null) + } + + def exitValue: F[Int] = F.asyncCheckAttempt[Int] { cb => + F.delay { + (childProcess.exitCode: Any) match { + case i: Int => Right(i) + case _ => + val f: js.Function1[Any, Unit] = { + case i: Int => cb(Right(i)) + case _ => // do nothing + } + childProcess.once("exit", f) + Left(Some(F.delay(childProcess.removeListener("exit", f)))) + } } - ) + } + + def stdin = writeWritable(F.delay(childProcess.stdin)) + + def stdout = unsafeReadReadable(childProcess.stdout) - subprocess.once("spawn", () => cb(Right(subprocess))) - subprocess.once[js.Error]("error", e => cb(Left(js.JavaScriptException(e)))) + def stderr = unsafeReadReadable(childProcess.stderr) } - } { childProcess => - F.asyncCheckAttempt[Unit] { cb => + + val finalize = F.asyncCheckAttempt[Unit] { cb => F.delay { if ((childProcess.exitCode ne null) || (childProcess.signalCode ne null)) { Either.unit @@ -67,48 +88,10 @@ private[process] trait ProcessesCompanionPlatform { } } } - } - .flatMap { childProcess => - def suspend(readable: => fs2.io.Readable) = - SuspendedStream( - Stream.resource(fs2.io.suspendReadableAndRead()(readable)).flatMap(_._2) - ) - - (suspend(childProcess.stdout), suspend(childProcess.stderr)).mapN { - (suspendedOut, suspendedErr) => - new UnsealedProcess[F] { - def isAlive: F[Boolean] = F.delay { - (childProcess.exitCode eq null) && (childProcess.signalCode eq null) - } - - def exitValue: F[Int] = F.asyncCheckAttempt[Int] { cb => - F.delay { - (childProcess.exitCode: Any) match { - case i: Int => Right(i) - case _ => - val f: js.Function1[Any, Unit] = { - case i: Int => cb(Right(i)) - case _ => // do nothing - } - childProcess.once("exit", f) - Left(Some(F.delay(childProcess.removeListener("exit", f)))) - } - } - } - - def stdin: Pipe[F, Byte, Nothing] = - fs2.io.writeWritable(F.delay(childProcess.stdin)) - - def stdout: Stream[F, Byte] = suspendedOut.stream - - def stderr: Stream[F, Byte] = suspendedErr.stream - - } - - } + childProcess.once("spawn", () => cb(Right(fs2Process -> finalize))) + childProcess.once[js.Error]("error", e => cb(Left(js.JavaScriptException(e)))) } - .evalOn(MicrotaskExecutor) // guarantee that callbacks are setup before yielding to I/O - + } } } From e44325026eb12834f5ed80bc8c0c7d0d8e08b898 Mon Sep 17 00:00:00 2001 From: Arman Bilge Date: Sat, 23 Dec 2023 20:29:35 +0000 Subject: [PATCH 2/2] Remove `MicrotaskExecutor` --- .../fs2/io/internal/MicrotaskExecutor.scala | 41 ------------------- 1 file changed, 41 deletions(-) delete mode 100644 io/js/src/main/scala/fs2/io/internal/MicrotaskExecutor.scala diff --git a/io/js/src/main/scala/fs2/io/internal/MicrotaskExecutor.scala b/io/js/src/main/scala/fs2/io/internal/MicrotaskExecutor.scala deleted file mode 100644 index ded5c883c4..0000000000 --- a/io/js/src/main/scala/fs2/io/internal/MicrotaskExecutor.scala +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Copyright (c) 2013 Functional Streams for Scala - * - * Permission is hereby granted, free of charge, to any person obtaining a copy of - * this software and associated documentation files (the "Software"), to deal in - * the Software without restriction, including without limitation the rights to - * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of - * the Software, and to permit persons to whom the Software is furnished to do so, - * subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS - * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR - * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER - * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN - * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. - */ - -package fs2.io.internal - -import org.typelevel.scalaccompat.annotation._ - -import scala.scalajs.js -import scala.scalajs.js.annotation.JSGlobal -import scala.concurrent.ExecutionContext - -private[io] object MicrotaskExecutor extends ExecutionContext { - - def execute(runnable: Runnable): Unit = queueMicrotask(() => runnable.run()) - - def reportFailure(cause: Throwable): Unit = cause.printStackTrace() - - @JSGlobal("queueMicrotask") - @js.native - @nowarn212("cat=unused") - private def queueMicrotask(function: js.Function0[Any]): Unit = js.native - -}