Last updated on

Futures

Welcome to week 12 of CS-214 — Software Construction!

As usual, ⭐️ indicates the most important exercises and questions and 🔥 indicates the most challenging ones.

You do not need to complete all exercises to succeed in this class, and you do not need to do all exercises in the order they are written.

We strongly encourage you to solve the exercises on paper first, in groups. After completing a first draft on paper, you may want to check your solutions on your computer. To do so, you can download the scaffold code (ZIP).

Predicting future runtimes ⭐️

Let’s start our exploration of future with a series of check-yourself question: can you predict the runtime of various combinations of futures?

We assume the following two utility functions:

def sleep(durationSec: Int): Future[Unit] =
  Future(Thread.sleep(durationSec * 1000))

src/main/scala/futures/runtime/runtimePrediction.scala

def wait[T](message: String)(f: Future[T]): Unit =
  val start = System.currentTimeMillis
  Await.result(f, Duration.Inf)
  val elapsed = (System.currentTimeMillis - start) / 1000.0
  println(f"    $message: ${elapsed}%.0f seconds")

src/main/scala/futures/runtime/runtimePrediction.scala

The first sleeps for durationSec sections; the second measures how long f takes to exit and then prints message along with the elapsed time.

Given these, can you predict the output of each of the lines below?

@main def predictionMain(): Unit =
  println("Single operators:")

  println("  flatMap:")

  wait("2flatMap4"):
    sleep(2).flatMap(_ => sleep(4))
  wait("4flatMap2"):
    sleep(4).flatMap(_ => sleep(2))

  println("  zip:")
  wait("2 zip 4"):
    sleep(2).zip(sleep(4))
  wait("4 zip 2"):
    sleep(4).zip(sleep(2))

  println("  sequence (;):")
  wait("2 ; 4"):
    sleep(2); sleep(4)
  wait("4 ; 2"):
    sleep(4); sleep(2)

  println("Combinations:")

  wait("2flatMap4 zip 1flatMap6"):
    sleep(2).flatMap(_ => sleep(4))
      .zip(sleep(1).flatMap(_ => sleep(6)))

  wait("2zip1 flatMap 4zip6"):
    sleep(2).zip(sleep(1)).flatMap(_ =>
      sleep(4).zip(sleep(6))
    )

  wait("2zip1 flatMap (4;6)"):
    sleep(2).zip(sleep(1)).flatMap(_ =>
      sleep(4); sleep(6)
    )

  wait("2zip1 flatMap (6;4)"):
    sleep(2).zip(sleep(1)).flatMap(_ =>
      sleep(6); sleep(4)
    )

  println("Early start:")

  wait("A=2 B=4 AflatMapB"):
    val a = sleep(2)
    val b = sleep(4)
    a.flatMap(_ => b)

  wait("A=2 B=4 AzipB"):
    val a = sleep(2)
    val b = sleep(4)
    a.zip(b)

src/main/scala/main.scala

Making new friends ⭐️

In this exercise, we will explore different styles of writing asynchronous code in Scala: the “callback” style, the Future “monadic” style, and the Future “direct” style.

The Social Network

Consider a directed graph where nodes are users of a social network, and edges represent friendship. In our model, friendship is a symmetric relation: if a is friends with b, then b is friends with a.

Let Database be a trait that provides methods to get and update users:

trait Database:
  /** Gets a user by its id asynchronously.
    *
    * @param id
    *   Id of the user.
    * @return
    *   A `Future` that will resolve to the queried [[User]].
    */
  def getUser(id: String): Future[User]

  /** Updates a user in the database asynchronously.
    *
    * @param currentValue
    *   Current value of the user.
    * @param newValue
    *   New value of the user.
    * @return
    *   A `Future` that will resolve to `true` if the operation was successful,
    *   or `false` otherwise.
    */
  def updateUser(currentValue: User, newValue: User): Future[Boolean]

  /** Gets a user by its id asynchronously.
    *
    * @param id
    *   Id of the user.
    * @param callback
    *   Callback to call with the queried [[User]].
    */
  def getUserWithCallback(id: String)(callback: User => Unit): Unit

  /** Updates a user in the database asynchronously.
    *
    * @param currentValue
    *   Current value of the user.
    * @param newValue
    *   New value of the user.
    * @param callback
    *   Callback to call with the result of the operation. The result will be
    *   `true` if the operation was successful, or `false` otherwise.
    */
  def updateUserWithCallback(currentValue: User, newValue: User)(callback: Boolean => Unit): Unit

src/main/scala/futures/friends/package.scala

Your task is to write a function makeFriends. The function should take two user identifiers user1Id: String anduser2Id: String, and a db: Database as parameters, and update the list of friends of both users to add the other user asynchronously. This means that the function should return immediately, and the friend list should be updated eventually.

In order to achieve this, you will call asynchronous functions to get and update users.

The User class is defined as follows:

case class User(id: String, friends: List[String])

src/main/scala/futures/friends/package.scala

To make two users friends, you will need to:

  1. Get the first user from the database.
  2. Add the second user to the first user’s friend list.
  3. Update the first user in the database.
  4. Get the second user from the database.
  5. Add the first user to the second user’s friend list.
  6. Update the second user in the database.
  7. Notify the caller that the operation is complete.

To copy a User with a new list of friends, you can use its copy: method:

user.copy(friends = ???)

Sequential execution

In the first versions of makeFriends, we will perform the operations sequentially, one after the other:

Directed acyclic graph the steps “Get user aisha”, “Update user aisha”, “Get user carlos”, “Update user carlos”, “print”, where each step depends on the previous one.

In this representation, the arrows represent dependencies between operations. If there is an arrow from a to b, it means that b should only be executed after a has completed.

Callback style

The “callback” style was mostly used in the past, and is still used in some libraries. In this style, each asynchronous operation takes an additional callback function as a parameter, which is called with the desired result when the operation completes.

In this style, the makeFriendsWithCallback function takes an additional parameter callback: (Boolean, Boolean) => Unit which is called when the operation completes:

def makeFriendsWithCallback(
    user1Id: String,
    user2Id: String,
    db: Database
)(
    callback: ((Boolean, Boolean)) => Any
): Unit =
  ???

src/main/scala/futures/friends/package.scala

The result of the operation is a pair of booleans, indicating whether the updates of the two users were successful.

Implement makeFriendsWithCallback by calling getUserCallback and updateUserCallback appropriately.

Testing

To test your code, copy your implementations along with the scaffold code in your favorite editor and run makeFriendsMain with the appropriate version identifier, here callbacks:

sbt:futures> runMain makeFriendsMain callbacks

The main function is run with a mock database where each operation takes 1 second to complete. As you you implemented the operations sequentially, the total time to complete the operation should be 4 seconds.

A correct implementation should print the following:

[info] running makeFriendsMain callbacks
Using callbacks
Waiting for callback to complete...
Get user aisha
Get user carlos
Update user aisha
Update user carlos
Aihsa and Carlos are now friends!
result: (true,true)
Completed in: 4 seconds

Monadic style, with flatMap and map

Let’s now use Futures, in the “monadic” style. In this style, each asynchronous operation returns a Future object, which represents the result of the operation. We can then use methods such as map and flatMap one the Future to transform and combine it with other Futures.

In this style, we never await on a Future directly, except at the very end of the program; your implementation must not call Await.result, Await.ready or await.

Complete the makeFriendsMonadicFlatMap function by calling the getUser and updateUser, and combining the resulting Futures using flatMap and map:

def makeFriendsMonadicFlatMap(
    user1Id: String,
    user2Id: String,
    db: Database
): Future[(Boolean, Boolean)] =
  ???

src/main/scala/futures/friends/package.scala

Testing

Yon can test your implementation by running the runMain makeFriendsMain monadic-flatmap. A correct implementation should print the following:

Using monadic style and flatMap
Waiting for future to complete...
Get user aisha
Update user aisha
Get user carlos
Update user carlos
Aihsa and Carlos are now friends!
result: (true,true)
Completed in: 4 seconds

Monadic style, with for-comprehensions

As you have learned, combinations of flatMap and map can be written more concisely using for-comprehensions. Rewrite the previous solution using for-comprehensions:

def makeFriendsMonadicFor(
    user1Id: String,
    user2Id: String,
    db: Database
): Future[(Boolean, Boolean)] =
  ???

src/main/scala/futures/friends/package.scala

Testing

You can test your implementation by running the runMain makeFriendsMain monadic-for. It should complete in 4 seconds.

Direct style

Finally, let’s consider Futures in the “direct” style, which is a more recent style that is gaining popularity in many languages. In this style, we also use Futures, but we allow calling await from inside Futures whenever we want. This is the future of Futures in Scala, but is not yet ready for production use for the reasons explained in the lecture.

In this style, the makeFriends function also returns a Future[(Boolean, Boolean)]:

def makeFriendsDirect(
    user1Id: String,
    user2Id: String,
    db: Database
): Future[(Boolean, Boolean)] =
  ???

src/main/scala/futures/friends/package.scala

Implement it by calling the getUser and updateUser functions appropriately.

We also provide you with the following await extension method on Future that you can use to block until the result of a future is available:

extension [T](self: Future[T])
  def await: T = Await.result(self, 3.seconds)

src/main/scala/futures/await.scala

In the direct style, you can call await, as long you are inside a Future, so that you don’t block the main thread.

Testing

You can test your implementation by running the runMain makeFriendsMain direct. It should complete in 4 seconds.

Parallel execution

Can we make the makeFriends function faster by executing some of the operations in parallel? Yes, we can!

Updating the two users can be done in parallel, since they are independent. Only the final operation, notifying the caller, depends on the two updates:

Directed acyclic graph showing the steps “Get user aisha”, “Update user aisha”, “Get user carlos”, “Update user carlos”, “print”, where “Update user aisha” depends on “Get user aisha”, “Update user carlos” depends on “Get user carlos”, and “print” depends on “Update user aisha” and “Update user carlos”.

Monadic style

Implement makeFriendsMonadicParallel:

def makeFriendsMonadicParallel(
    user1Id: String,
    user2Id: String,
    db: Database
): Future[(Boolean, Boolean)] =
  ???

src/main/scala/futures/friends/package.scala

Hint

You can use a combination of flatMap and map, or zip, to combine two Futures into a single Future that completes when both Futures have completed.

Testing

You can test your implementation by running the runMain makeFriendsMain monadic-flatmap-par. It should complete in 2 seconds.

Direct style

Implement makeFriendsDirectParallel:

def makeFriendsDirectParallel(
    user1Id: String,
    user2Id: String,
    db: Database
): Future[(Boolean, Boolean)] =
  ???

src/main/scala/futures/friends/package.scala

Testing

You can test your implementation by running the runMain makeFriendsMain direct-par. It should complete in 2 seconds.

Reimplementing parallel collections ⭐️

Remember parallel collections? How could you implement your own parallel collections using Futures?

In this exercise, you will implement ParList, a wrapper around a List that provides map, flatMap and filter methods that execute their argument functions in parallel.

def map[U](f: T => U): ParList[U] =
  ???

src/main/scala/futures/parallel/ParList.scala

def flatMap[U](f: T => ParList[U]): ParList[U] =
  ???

src/main/scala/futures/parallel/ParList.scala

def filter(p: T => Boolean): ParList[T] =
  ???

src/main/scala/futures/parallel/ParList.scala

You can use Future.sequence to combine a list of Futures into a single Future that completes when all the Futures have completed.

Task manager

In this exercise, you will implement a simple command-line task manager. To start the REPL of the task manager, run runMain tasksMain in the sbt console.

Adding tasks

The first command to implement is add. It has the form of add <time> <text>, which runs a task that first waits for <time> seconds then prints out <text>. Right now the tasks run synchronously. The REPL is blocked while the task is running, and you can only enter a new command after the task has completed.

sbt:futures> runMain tasksMain
[info] compiling 1 Scala source to /Users/linyxus/Workspace/course-material/exercises/futures/target/scala-3.3.1/classes ...
> add 10 hello
Started task
Finished task: hello
>

Now your task is to make the tasks run asynchronously, by modifying the following definition:

case ADD_RE(duration, result) =>
  println(s"Started task")
  Thread.sleep(duration.toLong * 1000)
  println(s"Finished task: $result")
  loop()

src/main/scala/futures/tasks/managerRepl.scala

Start by taking a look at the code in managerRepl.scala to get familiar with how the following is used throughout the program:

val tasks = collection.mutable.ArrayBuffer.empty[Future[String]]

src/main/scala/futures/tasks/managerRepl.scala

Here, ADD_RE is a regex that matches the syntax of the add command.

Try numbering the tasks to distinguish them. The REPL should now work like the following:

sbt:futures> runMain tasksMain
[info] running tasksMain 
> add 5 hello
> Started task #0
add 6 world
> Started task #1
Finished task #0: hello
Finished task #1: world

Tasks with dependencies

Implement another command addafter <depid> <time> <text>, which creates a tasks that waits for <time> seconds then prints out <text>, but starts running only after the task with id <depid> has completed.

sbt:futures> runMain tasksMain
[info] running tasksMain 
> add 5 hello
> Started task #0
addafter 0 1 world
> Finished task #0: hello
Started task #1
Finished task #1: world

In this example, we start a task (numbered 0) that prints hello after 5 seconds, then starts another task (numbered 1) which runs after the first one and prints world after 1 second. Task #1 starts only after #0 has completed.

Complete the following sketch.

case ADD_AFTER_RE(after, duration, result) =>
  ???

src/main/scala/futures/tasks/managerRepl.scala

Utility methods of future ⭐️

Implement the following utility methods of Future:

Hint

You may want to manipulate a Promise rather than futures directly.

extension [T](self: Future[T])
  def map[U](f: T => U): Future[U] =
    ???

src/main/scala/futures/ops/futureOps.scala

extension [T](self: Future[T])
  def flatMap[U](f: T => Future[U]): Future[U] =
    ???

src/main/scala/futures/ops/futureOps.scala

extension [T](self: Future[T])
  def zip[U](other: Future[U]): Future[(T, U)] =
    ???

src/main/scala/futures/ops/futureOps.scala

def sequence[T](futures: List[Future[T]]): Future[List[T]] =
  ???

src/main/scala/futures/ops/futureOps.scala

def race[T](futures: List[Future[T]]): Future[T] =
  ???

src/main/scala/futures/ops/futureOps.scala

You can play with your implementation by running runMain futuresTest in the sbt console.

GitHub contributors ⭐️

In this exercise, you will implement a function that retrieves the list of contributors of a GitHub repository, and display their names and location:

sbt:futures> runMain showContributorsMain nasa marsapi 3
[info] running (fork) showContributorsMain nasa marsapi 3
[info] Getting contributors of nasa/marsapi
[info] John from Lausanne, Switzerland
[info] Mary from somewhere
[info] James from Bangalore, India
[info] Patricia from Sydney
[info] Someone from Tokyo

The arguments to the showContributorsMain function are:

  1. The name of a GitHub organization.
  2. The name of a GitHub repository.
  3. The number of contributors to get per page when querying the GitHub API (see below).

GitHub API

In order to get the list of contributors of a GitHub repository, we will use the GitHub REST API, in particular the list contributors endpoint and the get user endpoint:

  1. The first enables us to get the list of contributors of a repository. For example, we can get the list of contributors of the lampepfl/gears repository by sending a request to https://api.github.com/repos/lampepfl/gears/contributors.
  2. The second enables us to get the details of a user. For example, we can get the details of the user odersky by sending a request to https://api.github.com/users/odersky.

The GitHub API paginates the results of most endpoints that return sequences to avoid returning too much data at once.

You can control pagination using two URL parameters:

  1. per_page: the number of results to return per page. This will be the pageSize parameter of the showContributorsMain and showContributors functions.
  2. page: the page number to return.

Try it with the lampepfl/gears repository!

Scala API

To call the GitHub API, we provide you with a get function that handles the JSON parsing for you. It takes a URL and returns a Future[Response[T]]:

def get[T](
    /** The URL to call. */
    uri: Uri
)(
    /** The HTTP backend used to send requests. */
    using backend: HttpBackend
)(
    /** Used to decode the response JSON to a `T`. */
    using Reader[T]
): Future[Response[T]] =

src/main/scala/futures/rest/utils.scala

The Response contains the following 2 member that you will need to use:

  1. a body field of type T, which contains the parsed JSON response.
  2. a header(String) method, which returns the value of the given HTTP header.

You can for example use it as follows to query the contributors of the nasa/marsapi repository:

// The `uri` interpolator is used to create a `Uri` from a string.
val url = uri"$API_URL/repos/nasa/marsapi/contributors?per_page=3"

// `get[T]` returns a `Future[T]`.
get[List[Contributor]](url)
  // We `map` over the `Future` to access the response.
  .map(res =>
    // res.body` is a `List[RepoContributor]`:
    assertEquals(
      res.body,
      List(
        Contributor(f"$API_URL/users/john"),
        Contributor(f"$API_URL/users/mary"),
        Contributor(f"$API_URL/users/james")
      )
    )
    // Each `Response` contains a `Link` header with a link to the next
    // page. You can use the `getNextPageUrl` function to extract the link.
    // It returns an `Option[String]`:
    assertEquals(
      getNextPageUrl(res.header("Link")),
      Some(f"$API_URL/repos/0/contributors?per_page=3&page=2")
    )
  )

src/test/scala/futures/rest/GithubStubBackendTest.scala

Contributor is defined as:

case class Contributor(
    /** API URL of the corresponding user. */
    url: String
) derives ReadWriter

src/main/scala/futures/rest/contributors.scala

To query the details of a user, you can use the get function as follows:

val url = uri"$API_URL/users/jennifer"
get[User](url)
  .map(res =>
    assertEquals(
      res.body,
      User(Some("Jennifer"), Some("Nairobi"))
    )
  )

src/test/scala/futures/rest/GithubStubBackendTest.scala

User is defined as:

case class User(
    /** Full name of the user. */
    name: Option[String],
    /** Location of the user. */
    location: Option[String]
) derives ReadWriter:
  override def toString(): String =
    val n = name.getOrElse("Someone")
    val l = location.getOrElse("somewhere")
    f"$n from $l"

src/main/scala/futures/rest/contributors.scala

Stub API

The Github API is rate limited to 60 unauthenticated requests per minute.

To avoid hitting this limit and simplify testing, we provide you with a stub API that you can use instead of the real GitHub API. You can use it by providing githubStubBackend as the HttpBackend parameter to the get function.

The examples of the previous section use the stub API.

See also how we use it in the showContributorsMain implementation:

@main def showContributorsMain(
    org: String,
    repo: String,
    perPage: Int
): Unit =
  // given HttpBackend = sttp.client4.HttpClientFutureBackend
  given HttpBackend = githubStubBackend
  given Window = ConsoleWindow
  println(f"Getting contributors of $org/$repo")
  Await.result(showContributorsDirect(org, repo, perPage), 2.seconds)

src/main/scala/main.scala

Our stub API only handles requests to the user and contributors endpoints, and only for two fake repositories: nasa/marsapi and sleepysloths/lazyloader which respectively have 5 and 2 stub contributors. All other requests will fail with a 404 Not Found error.

If you want to use the real GitHub API, you can set the given HttpBackend to HttpClientFutureBackend().

Order of execution

The crucial part of this exercise is the order in which the requests are executed and the results printed.

The following diagram illustrates the order in which the requests should be executed. If there is an arrow from A to B, it means that B should only be executed after A has completed.

When two requests are independent, they can and must be executed in parallel. For example, the requests to get the details of users $1, $2 and $3, and the request to get the second page of contributors should all be executed in parallel.

Note also that the details of users should be printed per page, as soon as the details of all users on that page have been retrieved.

Implementation

Your task is to write a showContributors method that takes 5 parameters:

  1. org: the name of the GitHub organization.
  2. repo: the name of the GitHub repository.
  3. pageSize: the number of contributors to get per page when querying the GitHub API.
  4. an implicit HttpBackend to use to query the GitHub API.
  5. an implicit Window object which contains a single println method, that you should use to print the results.

Write two versions:

  1. A version in monadic style, where you use the zip, flatMap and map methods of Future, and Future.sequence to combine multiple futures.
  2. A version in direct style: where you use the provided await extension method on Future to block until the result of a future is available.
Hint: scaffold suggestion for the monadic style version 
def showContributors(
    /** The organization that owns the repository. */
    org: String,
    /** The name of the repository. */
    repo: String,
    /** The number of contributors per page. */
    perPage: Int
)(using
    /** The HTTP backend used to send requests, used by `get`. */
    HttpBackend,
    /** Used to print the results. */
    Window
): Future[Unit] =
  def showNextPage(
      pageUri: Uri,
      prevPagePrinted: Future[Unit]
  ): Future[Unit] =
    ???
  val url = uri"$API_URL/repos/$org/$repo/contributors?per_page=$perPage"
  showNextPage(url, Future.successful(()))

src/main/scala/futures/rest/contributors.scala

Hint: scaffold suggestion for the direct style version 
def showContributorsDirect(
    org: String,
    repo: String,
    perPage: Int
)(using HttpBackend, Window): Future[Unit] =
  def showNextPageDirect(
      pageUri: Uri,
      prevPagePrinted: Future[Unit]
  ): Future[Unit] =
    Future:
      ???

  val url = uri"$API_URL/repos/$org/$repo/contributors?per_page=$perPage"
  showNextPageDirect(url, Future.successful(()))

src/main/scala/futures/rest/contributors.scala

Exam Questions

© 2023, 2024 EPFL. All rights reserved.