Cancel culture

Cancellation is a thorny issue in software - especially when we use containers as they're constantly being killed. 

First, let's look at how some Scala frameworks handle it. This Discord chat is about where the boundaries lie in Cats. Basically, if code running flatMaps that are chained together is cancelled, the next flatMap may be executed (you can see that with a println) but the IO created in it is not.

We're going to SIGTERM this code in CatsCleanup:

  val resource = Resource.make(IO.println("acquire"))(_ => IO.println("release"))

  def run: IO[Unit] = for {

    _ <- resource.use(_ => IO.readLine)

  } yield {

    ()

with 

kill -SIGTERM $(jps | grep CatsCleanup | awk '{print $1}')

In 3.5.4 the Resource is released and the output looks like:

acquire

release

Process finished with exit code 143 (interrupted by signal 15:SIGTERM)

(The exit code when killing a process is 128 + the sigterm code. You can see the last exit code of a process in a Unix-like system with echo $?).

The equivalent code in ZIO (2.0.21):

  val release = ZIO.logInfo("release")

  val resource = ZIO.acquireRelease(ZIO.logInfo("acquire"))(_ => {

      release

  })

  override def run: ZIO[Any & ZIOAppArgs & Scope, Any, Any] = for {

    _ <- resource

    _ <-zio.Console.readLine("press a key")

  } yield ()

does nothing.

Why it's important

With many workloads moving to, say, Kubernetes, cancellation comes with the territory. 

"What happens when a pod starts up, and what happens when a pod shuts down? 

"When a pod starts in a rolling deployment without the readiness probe configured ... the pod starts receiving traffic even though the pod is not ready. The absence of a readiness probe makes the application unstable. ... 

[HashNode]

The problem is"it takes more time to update the iptables rules than for the containers to be terminated by the Kubelet... The Kubelet immediately sends a SIGTERM signal to the container, and the endpoints controller sends a request back to the API server for the pod endpoints to be removed from all service objects... Due to the difference in task completion time, Services still route traffic to the endpoints of the terminating pods. 

[The solution involves] "adding a preStop hook to the deployment configuration. Before the container shuts down completely, we will configure the container to wait for 20 seconds. It is a synchronous action, which means the container will only shut down when this wait time is complete". 

This gives your application time to clean itself up.

Exception, Exception, Exceptions

I asked this question in the Discord channel about ZIO 1.0.3:

If I have a ZIO[Any, Throwable, String] that just throws an Exception then I can handle the exception with ZIO.either. 

But if I have a ZIO[Any, Throwable, String] that throws an Exception in the release of bracket I get:

    Fiber failed.
    An unchecked error was produced.

Is that expected? [Code to demonstrate here on GitHub]

Adam Fraser responded:

@PhillHenry The release action of bracket should never fail so it has a type of URIO[R, Any]. If there is an error there you need to expose the full cause and either handle it or ignore it. 

You are kind of lying to the compiler by doing URIO(x.close()).

URIO promises that something will never fail but close can fail so that is going to create an unchecked exception. 

If you make that something like Task(x.close()) to reflect the fact that the effect can fail then you are going to get a compilation error when you try to put that in bracket. 

And then you have a choice as the user. If a finalizer fails, which really shouldn't happen, how do you want to treat it? One option is to say just fail at that point. Not successfully running the finalizer indicates a potential resource leak and better to fail fast than die slowly later. So then you would call orDie on the effect, and you could potentially handle the cause at a higher level of your application. Another option is to say we made our best efforts, we just need to move on, and then you would do ignore or maybe log the error somewhere and then ignore it.

Why couldn't the Exception manifest itself in the error channel of the ZIO?

Because the error channel models the failure of the acquire and use actions of bracket. It is possible that the use action fails and then the release action runs and also fails, so we need different channels for those errors.

Could ZIO not use Throwable.addSuppressed? I notice that this is what Java's try-with-resource does in these situations

Well we have to be polymotphic. The error of use may not be a Throwable at all so we definitely can't add a suppressed exception to that.

That also makes it really easy to lose failures in finalizers when normally a failure in a finalizer is very bad and something you should explicitly handle.

Well the only way you are cheating the compiler is by doing UIO(somethingThatThrows). I'm not sure how that can be prevented other than by preventing users from constructing UIO values at all, which prevents users from describing effects that really can't fail. 

I think the lesson is just to only use UIO for effects that really don't throw exceptions (or if they do throw exceptions you are comfortable treating those as defects).

Can we not force bracket to be available only if the error channel is a Throwable?

But we want to use bracket in a ton of situations where it is not.

The bracket operator is core to safe resource management. Saying we could only use it when E was Throwable would prevent us from writing resource safe code with a polymorphic error type, which is basically all code we want to write.

Disquieting

If a ZIO can throw any Exception to bring the whole machinary to a crashing halt, why have an error channel in the first place? This seems like ZIO is a leaky abstraction. And this is what bothers me although colleagues have told me they don't find it a worry at all.

The Cats way

Drew Boardman @drewboardman Feb 23 22:15

Basically I'm trying to see if there exists something like MonadError that signals whether the error has been handled but I think this is only possible with datatypes, not with typeclasses. I was having a discussion about how MonadError doesn't really signal, at the type-level, that anything has been handled. I basically got around to just creating a datatype that signals this - but that effectively just re-invents Either

Adam Rosien @arosien Feb 23 22:18

"whether the error has been handled" - do you mean knowing this statically? MonadError and the like don't distinguish, in the type, error-handling.

Fabio Labella @SystemFw Feb 23 22:18

It's kinda possible with cats-mtl , but in general yeah it's a lot easier with datatypes... So in this case you'd have a MonadError constraint, and handle it (eliminate it) by instantiating to EitherT (locally) and then you handle that. 

To recap, MonadError tells you that the called code has the right to raiseError and if it does, what type this will be. But there is no guarantee that an IO will not throw an Exception that brings the JVM crashing down. 

IOs that throw Exceptions  and MonadErrors that raiseError can be .attempted to get an IO[Either[. That is, the datatype not the effect indicates whether the exception has been handled or not.

Cancellation idioms

Java IO interrupt refresher

"The InterruptibleChannel interface is a marker that, when implemented by a channel, indicates that the channel is interruptible... Most, but not all, channels are interruptible.

"Channels introduce some new behaviors related to closing and interrupts. If a channel implements the InterruptibleChannel interface, then it's subject to the following semantics. If a thread is blocked on a channel, and that thread is interrupted (by another thread calling the blocked thread's interrupt() method), the channel will be closed, and the blocked thread will be sent a ClosedByInterruptException.  Additionally, if a thread's interrupt status is set, and that thread attempts to access a channel, the channel will immediately be closed, and the same exception will be thrown." [Java NIO, Hitchens]

Summarising, if a thread is interrupted before or during a blocking call on a channel, the channel is closed and an exception is thrown.

"It may seem rather draconian to shut down a channel just because a thread sleeping on that channel was interrupted. But this is an explicit design decision made by the NIO architects.  Experience has shown that it's impossible to reliably handle interrupted I/O operations consistently across all operating systems."

"Interruptible channels are also asynchronously closable. A channel that implements InterruptibleChannel can be closed at any time, even if another thread is blocked waiting for an I/O to complete on that channel. When a channel is closed, any threads sleeping on that channel will be awakened and receive an AsynchronousCloseException. The channel will then be closed and will be no longer usable." [ibid]

The problems with Java

Daniel Spiewak @djspiewak Apr 24 19:47, 2020

There are a couple things with thread interruption that are horrible:

There's no way to build "uninterruptible" code. Meaning that you cannot have a critical section which acquires a resource atomically in several steps. Or in other words, there is no analogue to the acquire action in bracket.

The only way to detect self-cancelation for valid purposes (e.g. resource cleanup) is catching the InterruptedException, but doing this immediately flips the interrupted bit on the Thread back to false! 

The only solution to this is to do Thread.currentThread().interrupt() at the end of your exception handler, which almost no one knows to do. To make matters more annoying, even if you do this correctly, you mess up the stack trace on the interruption, because it's technically a new interrupt.

Oh, and exception handlers are not critical regions either, so if someone is just hammering the interrupt() button over and over externally, you could catch the exception, try to clean things up, and then get immediately interrupted again. This actually happens a lot because of the next point.

Catching Exception or Error will silently catch InterruptedException, even when that's almost guaranteed to not be what you want to do. This leads to silently ignoring interruption in most code paths, which is why people repeatedly hammer interrupt() in the first place.

The problems also go deeper than just Java but down to the OS level. "The underlying stream may not know it its closed until you attempt to write to it (e.g. if the other end of a socket closes it)" [SO]. "There's no API for determining whether a stream has been closed." [SO]

The semantics of cancelling

This is a complicated area. (see the interruption model proposed for Cats Effects 3 at https://github.com/typelevel/cats-effect/issues/681)

Basically interruptible/uninterruptible are not composable. The best way to think about it is that "interruptable means always accept the interrupt, no matter what", while "uninterruptible means always suppress the interrupt no matter what". But the uninterruptible(fa >> interruptible(fb) >> fc) breaks either one of the guarantees.
So you have to choose: do you want resource leaks (by biasing in favor of the innermost in that context), or do you want possible deadlocks (by biasing in favor of the outermost)?
And you can't even phrase it as inner/outer, because you can do the same thing in reverse: uninterruptible(interruptible(fa >> uninterruptible(fb) >> fc))
..
as prior art here, Haskell tried all of these and ultimately decided that mask/poll was the sanest solution [Daniel Spiewak, Gitter]

An alternative architecture

"it's a mistake to think of interruptibility as being an attribute of threads or fibers. Instead it should be an attribute of the activities which run on the threads/fibers, and of necessity, that means that any interruptible activity must have it's own first class interrupt channel. If we go down that route then an activity is interruptible if it 1) has an interrupt channel and 2) it's interrupt channel is accessible. If it doesn't have an interrupt channel, or the channel is hidden somehow, then it's uninterruptible.

"Exposing an explicit interrupt channel on every blocking operation that we want to be interruptible is obviously a lot more laborious than just firing random interrupts at globally visible threads/fibers and hoping for the best, but I think it's the only way to go."
[Miles Sabin]

How does this affect Effectful Systems?

Integrating Scala code that uses effectful libraries with Java IO code can cause problems.

Gavin Bisesi @Daenyth Feb 05 21:16, 2020

InputStream is always a blocking api

(note you don't need much to make a blocker; Blocker.apply gives a Resource of one)

Daniel Spiewak @djspiewak Feb 05 21:46, 2020

FYI, all things involving files are blocking except on Windows, and even then they're blocking most of the time.

So the "NIO stuff" that is inside of getResourceAsStream is actually not NIO but rather regular IO wrapped up with a thread pool :-(

I generally use Blocker just to be safe on resource access. It doesn't really cost that much in terms of syntax

... non-blocking things must have a callback-driven API

either directly (via callbacks passed to functions) or indirectly (via Future or CompletableFuture)

If something doesn't have a callback API, then you know it's blocking

When can you cancel?

The effect of a cancellation is felt at every asynch boundary or every N flatMaps (where N=1 for ZIO, it seems).

Note that there is no code in ZIO nor Cats that calls Thread.interrupt() that I could find. Note however that ZIO still gives you the ability to wrap your code in a Future and cancelling this will lead to an InterruptedException (see this gist).

Note that there are still undefined areas in Cats regarding cancellation:

Raas Ahsan @RaasAhsan Jul 07 20:26

calling cancel right after start results in non-deterministic behavior

Fabio Labella @SystemFw Jul 08 19:38

yeah I wanted to say

fa.guarantee(foo) doesn't guarantee that foo will always happen

it guarantees that if fa happens, then foo always happen

In particular if you have fa.guarantee(foo).start.flatMap(_.cancel), foo might happen or not, because the program can be cancelled before fa gets scheduled to run

See tip #2 at this Cats video (An Introduction to Interruption by Jakub Kozlowskiat 11'03" ) where starting and joining in a for comprehension is an anti-pattern (what if one fails to complete?). Instead, one should use (ioa, iob).parTupled.

Unit testing with ZIO

ZIO has the concept of Layers. These allow you to inject requirements to your ZIO monad (the R in ZIO[R, E, A]). The nice thing about them is that they make testing easier.

Say, I have an application with an intialisation piece and a data flow piece. My code would look like this:

import zio.{Has, Task, ZIO}

  type Init = Has[Init.Service]

  object Init {

    trait Service {

      def initializeWith(configFile:  String,

                         session:     SparkSession,

                         fs:          FileSystem): ZIO[Any, Throwable, Settings]

    }

...

  type Flow = Has[Flow.Service]

  object Flow {

    trait Service {

      def flow(s:       Settings,

               paths:   Filenames,

               session: SparkSession,

               fs:      FileSystem): Task[Results[String]]

    }

Now, the data structure I want to test is something like a ZIO[Init with Flow, Throwable, A] and the code that creates it is probably some large and complicated for-comprehension that's begging for a test. 

In production, the code to "run" it would look like:

    val prodInitialization: ULayer[Init]    = ZLayer.succeed(InitProd)

    val prodDataFlow:       ULayer[Flow]    = ZLayer.succeed(FlowProd)

...

    zio.provideLayer(prodInitialization ++ prodDataFlow)

where InitProd and FlowProd are my production objects that implement Init and Flow. 

My test code, however, look like:

import zio.test.Assertion._

import zio.test.environment.TestEnvironment

import zio.test.junit.JUnitRunnableSpec

import zio.test.{ZSpec, assertM, suite, testM}

import zio.{Layer, Task, ZIO, ZLayer}

class InitFlowSpec extends JUnitRunnableSpec {

  def unhappyPathInitLayer(e: Exception): Layer[Nothing, Init] = ZLayer.succeed(

    new Init.Service {

      override def initializeWith(configFile: String, session: SparkSession, fs: FileSystem) = ZIO.fail(e)

    }

  )

...

  val error                         = new Exception("Boo")

  val withNoErrors: Results[String] = Map.empty

  override def spec: ZSpec[TestEnvironment, Any] = suite("Orchestration")(

    testM("returns error if initialisation fails"){

      val layers    = unhappyPathInitLayer(error) ++ flowLayer(withNoErrors)

      val result    = zio.provideLayer(layers)

      assertM(result.run)(fails(equalTo(error)))

    }

  )

Et voila, we have a mocking framework for the requirement channel of a ZIO monad. 

Cracked Pipe

Pipes can get blocked as an old post of mine illustrates. So, how do we ensure that we can read and write to and from pipes in the effectful world?

Well, my first (rather poor) attempt was to use Java's PipedInputStream and PipedOutputStream and have ZIO handle the multithreading aspects of it. Although the test passes it is far from exhaustive.

One of the problems is that unbeknown to me that Java's piped IO classes are very thread sensitive. Consequently, my ZIO test was failing with:

    Fiber failed.

    A checked error was not handled.

    java.io.IOException: Read end dead

        at java.io.PipedInputStream.checkStateForReceive(PipedInputStream.java:262)

        at java.io.PipedInputStream.awaitSpace(PipedInputStream.java:268)

        at java.io.PipedInputStream.receive(PipedInputStream.java:231)

        at java.io.PipedOutputStream.write(PipedOutputStream.java:149)

        at java.io.OutputStream.write(OutputStream.java:75)

        at uk.co.odinconsultants.fp.zio.streams.LargePipeMain$.writing$1(LargePipeMain.scala:42)

        at uk.co.odinconsultants.fp.zio.streams.LargePipeMain$.$anonfun$piping$3(LargePipeMain.scala:58)

        at zio.internal.FiberContext.evaluateNow(FiberContext.scala:458)

        at zio.internal.FiberContext.$anonfun$evaluateLater$1(FiberContext.scala:687)

        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

        at java.lang.Thread.run(Thread.java:748)

This happens because the JDK code has something in it that looks like this:

...

        } else if (readSide != null && !readSide.isAlive()) {

            throw new IOException("Read end dead");

        }

...

and to be fair, the Java API documentation says:

"A pipe is said to be broken if a thread that was providing data bytes to the connected piped output stream is no longer alive."

But if your effect system uses fibres instead of threads then it is free to stop and start threads as it pleases. The bottom line is that wrapping side effecting code in a ZIO or a Cats IO might not be the first step in making your code more FP.

Applicative and Effectful Types

Typical use case: we have several legacy java.io.OutputStreams to close after we have done some work. We want to try to close all of them even if some might fail to close. Can we encapsulate these individual units of work in monads?

Recall a key distinction between the type classes Applicative and Monad - Applicative captures the idea of independent computations, whereas Monad captures that of dependent computations. Put differently Applicatives cannot branch based on the value of an existing/prior computation. Therefore when using Applicatives, we must hand in all our data in one go. [Cats documentation]

So, not only is monad inappropriate, it's impossible to perform this use case with monads at all.

Ok, let's look more closely at these Applicatives:

Rob Norris @tpolecat There's already a common misconception that applicative composition implies order-independence, which isn't necessarily (or even commonly) true. I feel like we're still kind of struggling for a way to talk about effects. I am, anyway.
Fabio Labella @SystemFw Because most education resources state that Applicative does represent non ordered computation whereas it can represents non ordered computation.
Rob Norris @tpolecat Even Validated is order-dependent

[From Gitter]

My reading of this is that calling Apply's *> will "Compose two actions, discarding any value produced by the first." (Cats Apply Scaladoc). The order is important in that all but the last result is thrown away.

To demonstrate this, I wrote some code here [GitHub] that uses Validated from Cats which is an Applicative as its code [GitHub] proves here.

My code uses a filthy var but as even the esteemed Fabio Labella "you need a Ref, but not a Ref of aWriter, just a Ref with IO. You can also use a var, whether you use one or another often depends on how complex the test is (var works well for simple cases with no concurrency)". So, with that caveat, let's proceed.

      var i = 0
      def failure(): Validated[String, String] = {
        println("failure")
        i = i + 1
        Invalid("failure message")
      }
      def success(): Validated[String, String] = {
        println("valid")
        i = i + 1
        Valid("success message")
      }

      success() *> failure() *> success()
      i shouldBe 3

This passes. Now, let's look at a monadic version in the same test class:

    type     MonadType = Either[String, Int]
    val aye: MonadType = Right(1)
    val nay: MonadType = Left("nope")

    aye *> nay *> aye shouldBe nay

You can demonstrate if you like that the last aye is not actually called but this is left as an exercise.

ZIO

ZIO is another Effects system. It is totally independent of Cats (unlike Monix, yet another Effects system, which depends on Cats).

"A quick summary: efforts to use Pure Functional Programming in Scala began with Scalaz lib which was then continued by Cats. These libraries are great, but in essence they try to replicate the whole Haskell experience in Scala. This comes at a cost mainly because Haskell is a non-strict, lazy-by-default language while the JVM is both eager and strict, so in order to make some things work some techniques were used that affect your performance and Scala's ability to infer types.
ZIO strives to give you a PFP experience but taking adavantage of Scala's paradigms (such as variance) to help both with performance and type inference.
it also tries to be more friendly for newcomers" ToxicaFunk (12 April 2020) on Discourse

Now, although all monads are applicatives, not all applicatives are monads (cats.data.Validated being an example of such a type). "We sometimes use the terms monadic effects or applicative effects to mean types with an associated Monad or Applicative instance." [Functional Programming in Scala]

In ZIO, there is "one monad to rule them all", also called ZIO. So, how do you get Applicative behaviour? None other than ZIO's creator himself answered me on Discourse:

jdegoes 12 April 2020 at 2:10 PM
@PhillHenry x1.ignore &> x2.ignore &> x3. This will execute x1, x2, and x3 in parallel, using zipRightPar (that's the &> operator, zips two effects together in parallel, returning whatever is produced on the right), and ignore the result of x1 and x2 so their failures don't influence the result of the computation.
@PhillHenry In ZIO, even parallel zip or collect or foreach operations will "kill" the other running effects if one of them fails. Because that's often what you want. To get the other behavior, just use .ignore in the right places to ignore the failures you don't care about.

My equivalent ZIO code can be found here on GitHub. Looking at the thread names and stack traces, these operations do appear to execute on different thread/fibres. This is not the case on the Cats implementation.