Search for a command to run...
![Synchronisation and Concurrent Access in Cats Effect 3 [Part 7]](/_next/image?url=https%3A%2F%2Fcdn.hashnode.com%2Fres%2Fhashnode%2Fimage%2Fupload%2Fv1653462070395%2F9fItyXLFg.jpg&w=3840&q=75)
Travel | Movies | History | Nature
I am a software developer. Started my career as a Java developer, but later switched to Scala. I wish to write Scala articles which are easier for newbies to follow. I love to travel, prefer to be alone or in very small group. Love to watch movies/series.
Function "accessWashroom" currentTime code is incomplete. Something like:
def accessWashroom(person: String, sem: Semaphore[IO]): IO[Unit] = for {
t0 <- IO(System.currentTimeMillis())
_ <- IO(s"[+ 0 ms] $person wants to access the washroom, waiting for getting the access").trace
_ <- sem.acquire
t1 <- IO(System.currentTimeMillis())
_ <- IO(s"[+ \({t1 - t0} ms] \)person got access to washroom and using it now").trace
_ <- IO.sleep(5.second)
t2 <- IO(System.currentTimeMillis())
_ <- IO(s"[+ \({t2 - t0} ms] \)person left the washroom").trace
_ <- sem.release
} yield ()
does the job.
Yes, definitely it is better to use it that way to be more functional. I updated it.
In this series, I will try to introduce different concepts in Cats Effects. This is going to be a list of many short articles regarding the key features of Cats Effect.
Introduction This is the 8th part of the Cats Effect 3 blog series. In this part, let's discuss about the other common operators/methods available in Cats Effect which were not discussed in the previous parts. Blocking As discussed in the previous se...
Implementing Continuous Data Flows with FS2 Streaming
Building Resumable Data Pipelines with Redis-backed Offset Tracking and Batch Processing
Creating Flexible Data Integration Patterns for Cross-System Data Movement
How to leverage Scala 3 Mirrors to generically parse CSV files into case classes, making data handling more flexible and type-safe.
Introduction Recently, at work, there was a requirement to list out the direct dependencies of the our SBT project, something like Bill Of Materials. SBT itself provides the dependencies as list or tree or searchable HTML page using the built-in comm...
This is the part 7 of the Cats Effect 3 blog series. In this part, we will be looking at the synchronisation primitives available in CE3.
When we have multiple fibers/threads working on the same datasets, it becomes important to have a safe concurrent access and synchronisations to the shared states. Cats Effect provides a wide variety of options to handle such cases.
Ref provides a thread-safe concurrent access and modification of the shared states. Under the hood, it uses Java's AtomicReference, but provides a functional way to handle it. Let's look at how to create and modify states using Ref.
We can mainly use 2 different ways to create a Ref:
val mutableState: IO[Ref[IO, Int]] = Ref[IO].of(100)
val mutableStateV2: IO[Ref[IO, Int]] = IO.ref(100)
We can use the method get to get the value from a Ref. Similarly, we can use set method to set value of a Ref variable. The method getAndSet returns the current value and also set a new value to the Ref. Let's look at these methods in action:
val refOps = for {
ref <- IO.ref(0) //initialises a Ref
value <- ref.get.trace // returns 0
_ <- ref.set(1) //sets the ref with value 1
oldValue <- ref.getAndSet(5).trace //returns 1 and set 5
_ <- ref.get.trace //returns 5
} yield ()
CE also provides some additional methods such as update, updateAndGet and getAndUpdate. They are similar to get and set methods, but it allows to pass a function to modify the state:
val refWithFns = for {
ref <- IO.ref("Hello World")
cap <- ref.getAndUpdate(_.toUpperCase).trace //returns str and then make ref upper case
str <- ref.get.trace // prints str in uppercase
_ <- ref.update(_.toLowerCase) //update the state
strV2 <- ref.get.trace // prints str in all lower
firstWord <- ref.updateAndGet(_.split("\\s").head.toUpperCase()).trace
} yield ()
There is a method called modify, which does similar as update. But modify allows to return a different return type after the state is updated. For example, we can use the ref from previous example and see how to use modify:
val refWithModify = for {
ref <- IO.ref("Hello World")
currentStr <- ref.get.trace // prints Hello World
_ <- IO.println("Length of current string is :" + currentStr.length) // returns 11
len <- ref.modify(value => (value.toUpperCase + "!", value.length)).trace //similar to getAndUpdate, but returns length
newStr <- ref.get.trace //prints HELLO WORLD!
_ <- IO.println("Length of updated string is: " + newStr.length) //returns 12
} yield ()
Deferred is something similar to the concept of Promise. It stores a value which is not yet available. While creating a Deferred, it will be empty. It can then be completed once with the value.
It has 2 important methods, get and complete. The get method blocks(sematically) until the value is available in the deferred variable. The method complete is used to complete a deferred variable and all the fibers which invoked get method will be notified of the value being ready. We can create a deferred variable by using the method deferred on IO. Let's create a program to make coffee using Deferred:
def developer(coffeeSignal: Deferred[IO, String]) = for {
_ <- IO("Developer wants to drink coffee and waiting for it to be ready").trace
_ <- coffeeSignal.get.trace // impatiantly waiting on coffee machine for it to be prepared
_ <- IO("Started sipping the divine coffee.. ").trace
} yield ()
def coffeeMachine(coffeeSignal: Deferred[IO, String]) = for {
_ <- IO("Verifying water and coffee beans").trace
grindingFib <- (IO("Grinding Coffee Beans").trace >> IO.sleep(Random.nextInt(500).millis) >> IO("Grinding complete").trace ).trace.start
boilingFib <- (IO("Boiling Water").trace >> IO.sleep(Random.nextInt(500).millis) >> IO("Boiling complete").trace).start
_ <- grindingFib.join
_ <- boilingFib.join
_ <- IO("Adding milk").trace
_ <- IO.sleep(100.millis)
_ <- coffeeSignal.complete("Coffee Ready!")
} yield ()
def makeCoffee = for {
coffeeSignal <- IO.deferred[String]
fib1 <- developer(coffeeSignal).start
fib2 <- coffeeMachine(coffeeSignal).start
_ <- fib1.join
_ <- fib2.join
} yield ()
Semaphore is a concurrency synchronisation concept where we can restrict the number of threads that can access some resource simultaneously. Semaphore contains a positive number of permits. Any fibers/threads that need to access thee guarded resource need to acquire a permit. After the access is done, the permit is released. If the available permits is zero, all the incoming fibers will be sematically blocked until something becomes available.
Semaphore defines methods such as acquire, release, acquireN and releaseN.
Let's try to implement a bathroom scenario with semaphores. Assume that there are only 2 washrooms. If more than 2 people needs to access the washrooms at the same time, some of them will have to wait for the others to complete and get out:
def accessWashroom(person: String, sem: Semaphore[IO]): IO[Unit] = for {
_ <- IO(s"[\(currentTime] \)person wants to access the washroom, waiting for getting the access").trace
_ <- sem.acquire
_ <- IO(s"[\(currentTime] \)person got access to washroom and using it now").trace
_ <- IO.sleep(5.second)
_ <- IO(s"[\(currentTime] \)person left the washroom").trace
_ <- sem.release
} yield ()
val persons = (1 to 5).map("Person-" + _).toList
val washroomAccessPgm = for {
washrooms <- Semaphore[IO](2)
fibers <- persons.map(p => accessWashroom(p, washrooms).start).sequence
_ <- fibers.map(_.join).sequence
} yield ()
If we run the washroomAccessPgm, we can see that some of the persons are made to wait outside the washroom until some of them becomes available.
Countdown latch is a synchronisation primitive that awaits until a pre-defined number of fibers are awaiting on it. A countdown latch is initialised with a positive number. When release is invoked on it, the latch number is reduced by 1. When the latch value becomes 0, the latch is opened and the execution is resumed. That means, the execution will continue only if the desired number of fibers releases on the latch. Let's implement a multi approval scenario with the countdown latch:
def accessSafeLocker(approvals: CountDownLatch[IO]) = for {
_ <- IO("Need to access safe locker.").trace
_ <- approvals.await
_ <- IO("Safe Locker opened and accessing the contents now").trace
} yield ()
def getApprovals(approvals: CountDownLatch[IO]) = for {
_ <- IO("Requesting approvals for safe access").trace
_ <- IO("Officer 1 Approval in progress").trace >> IO.sleep(Random.between(500, 1500).millis) >> IO("Officer 1 Approved").trace
_ <- approvals.release
_ <- IO("Officer 2 Approval in progress").trace >> IO.sleep(Random.between(500, 1500).millis) >> IO("Officer 2 Approved").trace
_ <- approvals.release
_ <- IO("Officer 3 Approval in progress").trace >> IO.sleep(Random.between(500, 1500).millis) >> IO("Officer 3 Approved").trace
_ <- approvals.release
} yield ()
def safeAccessProcess = for {
approvals <- CountDownLatch[IO](3)
fib <- accessSafeLocker(approvals).start
_ <- getApprovals(approvals)
_ <- fib.join
} yield ()
In the above process, only if 3 officers approve the request, then only the access is granted. If any one of the officers doesn't grant access(release latch), then the access to safe is not possible.
Countdown latch is a single use synchronisation method. Once the latch is opened, then we can't reuse that CountDownLatch. Cyclic Barrier is almost similar to countdown latch, but it is re-usable.
We can create a cyclic barrier by providing a positive number as:
val cyclicBarrier: IO[CyclicBarrier[IO]] = CyclicBarrier[IO](2)
We can use the method await on cyclicBarrier to wait for the desired number of participants. In the above code, when 2 fibers invoke await on cyclicBarrier, it continues the execution.
In this part, we looked at different primitives for handling synchronisation in out concurrent application. The sample code used in this blog is available in GitHub under the package part7.