You can compile the project with sbt:
scala-effects> sbt
sbt> project supermarket
sbt> compile
Our aim is to code up a stream that will represent shoppers paying at two checkouts, one of which is for fast shoppers only. You can see the illustration in supermarket/supermarket.png
for a rough idea of the input and output streams.
You can see and edit a diagram here.
Take a look at the App
in the threading
project. Check that you can compile and run it.
sbt
sbt> project threading
sbt> compile
Note: that the supermarket
project will fail to compile. Please only compile threading
.
The app can run two different kinds of work, found in the Work
object.
-
How long does the
writeToTheDatabase
work take to run? -
How long does the
calculateHash
work take to run?You will need to edit the following line of code:
Work.time(Work.writeToTheDatabase)
Find out the number of available processors on your computer:
- Enter the SBT console
sbt
sbt> project threading
sbt> console
scala> Runtime.getRuntime().availableProcessors()
val res0: Int = 16 // This is the number of available processors
-
How long does it take to run the app with this number of threads?
// In App.scala override def runtime: unsafe.IORuntime = Setup.createBasicRuntime(Setup.bounded("global", 16))
-
What about twice this number?
-
What about half this number?
The evalOn
function allows us to execute an IO
on a different thread pool (an ExecutionContext
is another name for a thread pool).
-
Take a look at the new
Work.factorial
function. Time it and see how long it takes:// In App.scala def run: IO[Unit] = Work.time(factorial)
If it takes less than a second, increase the
2000000000L
number within the function.What is printed as the thread name?
-
Now execute it on the
scala.concurrent.ExecutionContext.global
// In App.scala def run: IO[Unit] = Work.time(factorial.evalOn(scala.concurrent.ExecutionContext.global))
What is printed as the thread name?
-
Take a look at
writeToTheDatabase
. It now queries postgres. -
Run docker with
docker compose up -d
. This should start a postgres container. -
Run the application with
sbt
. Check that you can connect to postgres. How long does the query take?
The app runs a single writeToTheDatabase
task. It has:
- an unbounded blocking thread pool as part of
IORuntime
- a bounded compute pool
- an
ec
threadpool with a single thread that is passed to hikari.
-
Predict which threads will be blocked when running the app.
-
Run the app. In the session, we will profile this with visualvm to check your results.
The snooze
task sleeps a thread for 100 seconds.
Consider a factorial task followed by a snooze task:
Work.factorial >> Work.snooze
- If many of these tasks are run in parallel, predict how many factorials will be computed in the first 30 seconds.
Work.doLotsOf(Work.time(Work.factorial) >> Work.snooze)
- Run the app. In the session, we will profile this with visualvm to check your results.
This exercise explores the thread pool used by Hikari.
The hikari threadpool is configured with a single thread. There are only three connections allowed at once (the maximumPoolSize
is 3
). There is a connection timeout of two seconds.
-
Consider:
Work.doLotsOf(Work.handleError(Work.writeToTheDatabase(transactor)))
What errors do you expect to be printed to the console and when?
-
Consider configuring the thread pool with two threads:
val ecResource: Resource[IO, ExecutionContext] = ExecutionContexts.fixedThreadPool[IO](2)
What do you expect to be printed to the console and when?
The threading project now contains a HttpApp
.
- Start the app with
run
. - Query the app with
./work.sh 1
. How many factorial tasks do you expect to run?
The HttpApp
runs items of work
.
-
Start the app with
sbt run
. -
Query the app with
./work.sh 5
. How many work items do you expect to run concurrently? -
Consider the route:
case GET -> Root / "work" => work >> IO.println("Wrote to the db") >> Ok("Wrote to the db\n")
Modify
work
towork.start
. Query the app again with./work.sh 5
. How many work items do you expect to run concurrently?
The app now has two endpoints: sync-work
and async-work
.
- Start the app with
sbt run
.
You can call the endpoints with the shell script, e.g: ./work.sh sync-work 4
.
2. Consider the difference between the sync-work
and async-work
endpoints.
- How do they behave on failure? The fourth request made will fail due to a connection timeout.
- What status codes do they respond with?
- In both cases, how do they schedule work?
The app now has two endpoints under work
.
- The
POST
endpoint starts an async task. - The
GET
endpoint checks its status.
- Think about the code needed to properly implement these endpoints. Draw a rough diagram of the design in Excelidraw (or your preferred tool).
We'll begin today's session by mobbing on a design.
The app has some stubbed code under the work
endpoint.
for {
taskId <- Work.randomUUID
_ <- Work.queueTask(taskId)
_ <- Work.recordTask(taskId)
result <- Ok(taskId.toString)
} yield result
This queues a task (e.g. by sending it a kafka topic) and records it in some data store.
- What possible states can a task be in? You can consider "queued" and "running" to be states.
- What happens if
queueTask
succeeds, butrecordTask
fails? - Can
recordTask
ever succeed ifqueueTask
fails?
The messageQueue
project consumes messages (from kafka, for example), processes them and commits the offset.
Take a look at the processMessages
function.
- Can it ever commit an offset for a task before the task has been processed?
- Can it ever process a task more than once? Consider the case of application failure and restarts.
Consider processMessages
. It processes each message sequentially.
- Could we use
parEvalMap
to process these messages? - What would the consequences be of using
parEvalMapUnordered
?
In this session, we'll take a look at error handling. The code has been amended such that the message time is an Int
.
- Should it be possible for the user to submit negative times? If so, would you expect an error?
- What possible errors can occur when querying the database? For each error, consider whether we should recover from it.
We'll explore error handling with the egg
project.
-
Run the
FryEggApp
:sbt sbt> project egg sbt:egg> run
You should see an exception being thrown indicating
"The yolk broke during frying"
. -
Read through the
FryCook.fry
function to get a gist of what it does. -
Take a look at the
cookWithPower
function.- What is the difference between throwing an exception and returning a value?
- Is this a pure function? If not, how could we make it pure?
- Take a look at the cats API docs for
ApplicativeError
. In particular, look at thehandleError
andrecover
functions. - The
crack
andcook
functions capture errors in anIO
: either function may fail. Consider how you can use the functions onApplicativeError
to perform the following tasks:
- If the yolk is broken during cooking, return a scrambled egg instead
- If the egg is rotten, crack another egg
- If there are any errors, print "Sorry! Something wen't wrong."
In this session, we'll take a look at error handling and scopes.
For reference, here is our current implementation of fry
:
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
.recover { case YolkIsBroken => CookedEgg.Scrambled }
}.handleErrorWith(_ => fry(power, eggBox))
}
- Consider the following implementation of
fry
, paying attention to the position of therecover
function. Is the implementation correct?:
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
}.recover { case YolkIsBroken => CookedEgg.Scrambled }
.handleErrorWith(_ => fry(power, eggBox))
}
- What about the following implementation, paying attention to
handleErrorWith
?
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
.recover { case YolkIsBroken => CookedEgg.Scrambled }
.handleErrorWith(_ => fry(power, eggBox))
}
}
- What about the following implementation?
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
}
.handleErrorWith(_ => fry(power, eggBox))
.recover { case YolkIsBroken => CookedEgg.Scrambled }
}
In this session, we'll experiment with the order in which we handle errors.
For reference, here is our current implementation of fry
:
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg) // Previous position of `recover` handler
}
.recover { case YolkIsBroken => CookedEgg.Scrambled } // Current position
.handleErrorWith(_ => fry(power, eggBox))
}
We saw that moving the recover
handler did not change the behaviour.
- What about the following implementation? Are
YolkIsBroken
exceptions handled in the same way?
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
}
.handleErrorWith(_ => fry(power, eggBox))
.recover { case YolkIsBroken => CookedEgg.Scrambled }
}
- What about the following implementation, paying attention to the position of
handleErrorWith
? AreRottenEgg
exceptions still handled in the same way?
def fry(power: Ref[IO, Boolean], eggBox: Queue[IO, RawEgg]): IO[CookedEgg] = {
crack(eggBox).flatMap { egg =>
cook(power)(egg)
.recover { case YolkIsBroken => CookedEgg.Scrambled }
.handleErrorWith(_ => fry(power, eggBox))
}
}
Take a look at the numbers
project:
sbt
> project numbers
> compile
> test
- Run the code with
sbt run
- Test the code with
sbt test
- You'll see some tests in
NumbersTest
that are failing. How can you use thehandleError
functions to implement the correct behaviour?
Take a look at the numbers.scala
file.
-
In the last session, we changed the signature of
process
from:def process(message: Message): Stream[IO, Unit]
to:
def process(message: Message): IO[Unit]
We did this by "compiling" the stream into an
IO
usingstream.compile.drain
.How is the resulting
IO
different from the stream? - Will it ever time out, where the stream wouldn't? - Will it hold more data in memory than the stream? -
The signature for
run
is as follows:val run: Stream[IO, Unit]
- What is the meaning of
Unit
in this signature? - Would this function signature be better as
val run: IO[Unit]
?
- What is the meaning of
Take a look at the egg
project:
sbt
> project egg
> compile
Run the code with sbt run
.
- What happens when the egg taken from the egg box is rotten?
- Try and write a function with the following signature:
def crackAndRetry(eggBox: Queue[IO, RawEgg]): IO[RawEgg.FreshEgg] = ???
This function should call crack
, but crack another egg if the egg is rotten.
In the previous session, we attempted to retry the following action:
def crack(eggBox: Queue[IO, RawEgg]): IO[RawEgg.FreshEgg] = {
eggBox.take.flatMap {
case re @ RawEgg.RottenEgg => IO.raiseError(RottenEggError)
case egg: RawEgg.FreshEgg => IO.pure(egg)
}
}
We did so as follows:
def crackAndRetry(eggBox: Queue[IO, RawEgg]): IO[RawEgg.FreshEgg] = {
val policy = RetryPolicies.constantDelay[IO](2.seconds)
def onFailure(failedValue: RawEgg, details: RetryDetails): IO[Unit] = {
IO(println(s"Retrying on $failedValue: $details"))
}
def isSuccessful(value: RawEgg): IO[Boolean] =
value match {
case RawEgg.FreshEgg(yolkIsFragile, isSmall) => IO.pure(true)
case RawEgg.RottenEgg => IO.pure(false)
}
val action: IO[RawEgg.FreshEgg] = crack(eggBox)
retryingOnFailures(policy,
isSuccessful,
onFailure
)(action)
}
- Run the app. Does the current solution retry on rotten eggs?
- Will the
action
ever result in anIO[RawEgg.RottenEgg]
? - Will the
isSuccessful
function ever result in anIO(false)
?
In the past few sessions, we've examined error handling with IO
. This time, we'll take a look at error handling with fs2.Stream.
The FrySeveralEggsApp
in egg.scala
is meant to repeatedly cracks and cooks eggs.
- Run the app. Why does it raise an exception?
- Can we recover from the error using
Stream.handleErrorWith
? If not, why?
Stream
.repeatEval(FryCook.crack(eggBox))
.handleErrorWith(err => ...)
.evalMap(FryCook.cook(power))
- What other functions on
Stream
enable us to handle this error?
This session will kick off the topic of resources.
Take a look at the reader
project. This reads a file, cats.txt
, prints the first line, then sleeps.
- Check that you can compile and run the app.
- The following command gets a list of open file descriptors and searches it for
cats.txt
:Run the command while running the app. What does it tell you aboutps | grep App | cut -d' ' -f 1 | head -n 1 | xargs lsof -p | grep cats.txt
scala.io.Source
?
Take a look at the reader.scala
file:
Which of these expressions constructs an IO
that reads five lines from the cats.txt
file?
// Option 1
range.traverse(_ => sourceResource.flatMap(s => printLine(s).toResource)).use_
// Option 2
range.traverse(_ => sourceResource.use(printLine)).void
// Option 3
sourceResource.use(s => range.traverse(_ => printLine(s))).void
// Option 4
sourceResource.flatMap(s => range.traverse(_ => printLine(s).toResource)).use_
For each expression, figure out how many times the source file is opened and closed.
The following code in reader.scala
reads five lines from the file:
val range: List[Int] = (0 until 5).toList
sourceResource.use(s => range.traverse(_ => printLine(s))).void
We can define a sourceStream
as follows:
val sourceStream: Stream[IO, Source] = Stream.resource(sourceResource)
-
How many elements can this stream contain?
-
Combine this stream with the
printLine
function to print lines from the file such that the following code prints five lines:val printLineStream: Stream[IO, Unit] = ??? printLineStream.take(5).compile.drain
Take a look at the ref
project. The ref.scala
file contains its main app.
The app defines a counter
cats-effect Ref
that should contain the number of cats printed.
- Run the app to print out a list of four cat names.
- Update the counter each time a cat is printed by using the
updateCount
pipe.
Run the app. You should see at least 8 cat names printed:
sbt:ref> run
[info] running (fork) ref.CatNamesApp
[info] Mao
[info] Mao
[info] We're incrementing 0
[info] We're incrementing 0
[info] Maru
[info] Maru
[info] We're incrementing 1
[info] We're incrementing 2
[info] Popcorn
[info] Popcorn
[info] We're incrementing 3
[info] We're incrementing 3
[info] Mao
[info] Mao
[info] We're incrementing 4
[info] We're incrementing 4
[info] Maru
[info] Maru
[info] We're incrementing 5
[info] We're incrementing 5
[info] There are 6 cats.
Why did the counter increment 6 times?
Take a look at the streamref.CatNamesApp
in the ref
package. This uses a counter to keep track of the number of times a cat name is printed.
-
Do you need to use a
ref
to keep track of the count? Take a look at the functions in fs2 and see if you can do so using stream utilities instead. -
What happens if an error is raised in the stream? Use
printCatOrError
to raise an error when"Popcorn"
is encountered.
Take a look at the streamref.CatNamesApp
in the ref
package. This uses a ref to keep track of the State
of a stream.
- Read through the code. How does
requestCancellation
cancel the stream? - Run the app. Has the stream been cancelled?
We'll look at some patterns for working with shared state in concurrent systems.
Take a look at the stateRef
in the streamref.CatNamesApp
.
- What are the differences, in terms of concurrency challenges, between using a
Ref
or a database to store shared state? - The
update
function is used in two places. Can you generalize the usage of it?