Managing concurrent tasks

Clojure has a couple of handy constructs that allow us to define concurrent tasks. A thread is the most elementary abstraction of a task that runs in the background. In the formal sense, a thread is simply a sequence of instructions that can be scheduled for execution. A task that runs in the background of a program is said to execute on a separate thread. Threads will be scheduled for execution on a specific processor by the underlying operating system. Most modern operating systems allow a process to have several threads of execution. The technique of managing multiple threads in a single process is termed as multithreading.

While Clojure does support the use of threads, concurrent tasks can be modeled in more elegant ways using other constructs. Let's explore the different ways in which we can define concurrent tasks.

Note

The following examples can be found in src/m_clj/c2/concurrent.clj of the book's source code.

Using delays

A delay can be used to define a task whose execution is delayed, or deferred, until it is necessary. A delay is only run once, and its result is cached. We simply need to wrap the instructions of a given task in a delay form to define a delay, as shown in Example 2.1:

(def delayed-1
  (delay
   (Thread/sleep 3000)
   (println "3 seconds later ...")
   1))

Example 2.1: A delayed value

Note

The static Thread/sleep method suspends execution of the current thread of execution for a given number of milliseconds, which is passed as the first argument to this method. We can optionally specify the number of nanoseconds by which the current thread must be suspended as the second argument to the Thread/sleep method.

The delay form in Example 2.1 simply sleeps for 3000 milliseconds, prints a string and returns the value 1. However, it is not yet realized, in the sense that, it is has not been executed yet. The realized? predicate can be used to check whether a delay has been executed, as shown here:

user> (realized? delayed-1)
false
user> (realized? delayed-1) ; after 3 seconds
false

Note

We can check whether a value is a delay using the delay? predicate.

The body expressions in a delay form will not be executed until the value returned by it is actually used. We can obtain the value contained in a delay by dereferencing it using the at-the-rate symbol (@):

user> @delayed-1
3 seconds later ...
1
user> (realized? delayed-1)
true

Note

Using the at-the-rate symbol (@) to dereference a value is the same as using the deref function. For example, the expression @x is equivalent to (deref x).

The deref function also has a variant form that accepts three arguments—a value to dereference, the number of milliseconds to wait before timing out, and a value that will be returned in case of a timeout.

As shown previously, the expression @delayed-1 returns the value 1, after a pause of 3 seconds. Now, the call to realized? returns true. Also, the value returned by the expression @delayed-1 will be cached, as shown here:

user> @delayed-1
1

It is thus evident that the expression @delayed-1 will be blocked for 3 seconds, will print a string, and return a value only once.

Note

Another way to execute a delay is by using the force function, which takes a delay as an argument. This function executes a given delay if needed, and returns the value of the delay's inner expression.

Delays are quite handy for representing values or tasks that need not be executed until required. However, a delay will always be executed in the same thread in which it is dereferenced. In other words, delays are synchronous. Hence, delays aren't really a solution for representing tasks that run in the background.

Using futures and promises

As we mentioned earlier, threads are the most elementary way of dealing with background tasks. In Clojure, all functions implement the clojure.lang.IFn interface, which in turn extends the java.lang.Runnable interface. This means that any Clojure function can be invoked in a separate thread of execution. For example, consider the function in Example 2.2:

(defn wait-3-seconds []
  (Thread/sleep 3000)
  (println)
  (println "3 seconds later ..."))

Example 2.2: A function that waits for 3 seconds

The wait-3-seconds function in Example 2.2 waits for 3000 milliseconds and prints a new line and a string. We can execute this function on a separate thread by constructing a java.lang.Thread object from it using the Thread. constructor. The resulting object can then be scheduled for execution in the background by invoking its .start method, as shown here:

user> (.start (Thread. wait-3-seconds))
nil
user>
3 seconds later ...

user>

The call to the .start method returns immediately to the REPL prompt. The wait-3-seconds function gets executed in the background, and prints to standard output in the REPL after 3 seconds. While using threads does indeed allow execution of tasks in the background, they have a couple shortcomings:

• There is no obvious way to obtain a return value from a function that is executed on a separate thread.
• Also, using the Thread. and .start functions is essentially interop with the underlying JVM. Thus, using these functions in a program's code would mean that the program could be run only on the JVM. We essentially lock our program into a single platform, and the program can't be run on any of the other platforms that Clojure supports.

A future is a more idiomatic way to represent a task that is executed in a separate thread. Futures can be concisely defined as values that will be realized in the future. A future represents a task that performs a certain computation and returns the result of the computation. We can create a future using the future form, as shown in Example 2.3:

(defn val-as-future [n secs]
  (future
    (Thread/sleep (* secs 1000))
    (println)
    (println (str secs " seconds later ..."))
    n))

Example 2.3: A future that sleeps for some time and returns a value

The val-as-future function defined in Example 2.3 invokes a future that waits for the number of seconds specified by the argument secs, prints a new line and a string, and finally returns the supplied value n. A call to the val-as-future function will return a future immediately, and a string will be printed after the specified number of seconds, as shown here:

user> (def future-1 (val-as-future 1 3))
#'user/future-1
user>
3 seconds later ...

user>

The realized? and future-done? predicates can be used to check whether a future has completed, as shown here:

user> (realized? future-1)
true
user> (future-done? future-1)
true

Note

We can check whether a value is a future using the future? predicate.

A future that is being executed can be stopped by using the future-cancel function, which takes a future as its only argument and returns a Boolean value indicating whether the supplied future was cancelled, as depicted here:

user> (def future-10 (val-as-future 10 10))
#'user/future-10
user> (future-cancel future-10)
true

We can check whether a future has been cancelled using the future-cancelled? function. Also, dereferencing a future after it has been cancelled will cause an exception, as shown here:

user> (future-cancelled? future-10)
true
user> @future-10
CancellationException   java.util.concurrent.FutureTask.report (FutureTask.java:121)

Now that we are familiar with the notion of representing tasks as futures, let's talk about how multiple futures can be synchronized. Firstly, we can use promises to synchronize two or more futures. A promise, created using the promise function, is simply a value that can be set only once. A promise is set, or delivered, using the deliver form. Subsequent calls to the deliver form on a promise that has been delivered will not have any effect, and will return nil. When a promise is not delivered, dereferencing it using the @ symbol or the deref form will block the current thread of execution. Hence, a promise can be used with a future in order to pause the execution of the future until a certain value is available. The promise and deliver forms can be quickly demonstrated as follows:

user> (def p (promise))
#'user/p
user> (deliver p 100)
#<core$promise$reify__6363@1792b00: 100>
user> (deliver p 200)
nil
user> @p
100

As shown in the preceding output, the first call to the deliver form using the promise p sets the value of the promise to 100, and the second call to the deliver form has no effect.

Note

The realized? predicate can be used to check whether a promise instance has been delivered.

Another way to synchronize concurrent tasks is by using the locking form. The locking form allows only a single task to hold a lock variable, or a monitor, at any given point in time. Any value can be treated as a monitor. When a monitor is held, or locked, by a certain task, any other concurrent tasks that try to acquire the monitor are blocked until the monitor is available. We can thus use the locking form to synchronize two or more concurrent futures, as shown in Example 2.4:

(defn lock-for-2-seconds []
  (let [lock (Object.)
        task-1 (fn []
                 (future
                   (locking lock
                     (Thread/sleep 2000)
                     (println "Task 1 completed"))))
        task-2 (fn []
                 (future
                   (locking lock
                     (Thread/sleep 1000)
                     (println "Task 2 completed"))))]
    (task-1)
    (task-2)))

Example 2.4: Using the locking form

The lock-for-2-seconds function in Example 2.4 creates two functions, task-1 and task-2, which both invoke futures that try to acquire a monitor, represented by the variable lock. In this example, we use a boring java.lang.Object instance as a monitor for synchronizing two futures. The future invoked by the task-1 function sleeps for two seconds, whereas the future called by the task-2 function sleeps for a single second. The future called by the task-1 function is observed to complete first as the future invoked by the task-2 function will not be executed until the locking form in the future obtains the monitor lock, as shown in the following output:

user> (lock-for-2-seconds)
[#<core$future_call$reify__6320@19ed4e9: :pending>
 #<core$future_call$reify__6320@ac35d5: :pending>]
user>
Task 1 completed
Task 2 completed

We can thus use the locking form to synchronize multiple futures. However, the locking form must be used sparingly as careless use of it could result in a deadlock among concurrent tasks. Concurrent tasks are generally synchronized to pass around a shared state. Clojure allows us to avoid using the locking form and any possible deadlocks through the use of reference types to represent shared state, as we will examine in the following section.