Managing state

A program can be pided into several parts which can execute concurrently. It is often necessary to share data or state among these concurrently running tasks. Thus, we arrive at the notion of having multiple observers for some data. If the data gets modified, we must ensure that the changes are visible to all observers. For example, suppose there are two threads that read data from a common variable. This data gets modified by one thread, and the change must be propagated to the other thread as soon as possible to avoid inconsistencies.

Programming languages that support mutability handle this problem by locking over a monitor, as we demonstrated with the locking form, and maintaining local copies of the data. In such languages, a variable is just a container for data. Whenever a concurrent task accesses a variable that is shared with other tasks, it copies the data from the variable. This is done in order to prevent unwanted overwriting of the variable by other tasks while a task is performing a computation on it. In case the variable is actually modified, a given task will still have its own copy of the shared data. If there are two concurrent tasks that access a given variable, they could simultaneously modify the variable and thus both of the tasks would have an inconsistent view of the data in the given variable. This problem is termed as a race condition, and must be avoided when dealing with concurrent tasks. For this reason, monitors are used to synchronize access to shared data. However, this methodology is not really deterministic, in the sense that we cannot easily reason about the actual data contained in a variable at a given point in time. This makes developing concurrent programs quite cumbersome in programming languages that use mutability.

Like other functional programming languages, Clojure tackles this problem using immutability—all values are immutable by default and cannot be changed. To model mutable state, there is the notion of identity, state, and time:

• An identity is anything that is associated with a changing state. At a given point in time, an identity has a single state.
• State is the value associated with an identity at a given point in time.
• Time defines an ordering between the states of an identity.

Programs that actually use state can thus be pided into two layers. One layer is purely functional and has nothing to do with state. The other layer constitutes parts of the program that actually require the use of mutable state. This decomposition allows us to isolate the parts of a program that actually require the use of mutable state.

There are several ways to define mutable state in Clojure, and the data structures used for this purpose are termed as reference types. A reference type is essentially a mutable reference to an immutable value. Hence, the reference has to be changed explicitly, and the actual value contained in a reference type cannot be modified in any way. Reference types can be characterized in the following ways:

• The change of state in some reference types can either be synchronous or asynchronous. For example, suppose we are writing data to a file. A synchronous write operation would block the caller until all data is written to the file. On the other hand, an asynchronous write operation would start off a background task to write all data to the file and return to the caller immediately.
• Mutation of a reference type can be performed in either a coordinated or an independent manner. By coordinated, we mean that state can only be modified within transactions that are managed by some underlying system, which is quite similar to the way a database works. A reference type that mutates independently, however, can be changed without the explicit use of a transaction.
• Changes in some state can be visible to only the thread in which the change occurs, or they could be visible to all threads in the current process.

We will now explore the various reference types that can be used to represent mutable state in Clojure.

Using vars

Vars are used to manage state that is changed within the scope of a thread. We essentially define vars that can have state, and then bind them to different values. The modified value of a var is only visible to the current thread of execution. Hence, vars are a form of the thread-local state.

Note

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

Dynamic vars are defined using the def form with the :dynamic meta keyword. If we omit the :dynamic metadata, it would be the same as defining an ordinary variable, or a static var, using a def form. It's a convention that all dynamic var names must start and end with the asterisk character (*), but this is not mandatory. For example, let's define a dynamic variable shown as follows:

(def ^:dynamic *thread-local-state* [1 2 3])

The *thread-local-state* variable defined in Example 2.5 represents a thread-local var that can change dynamically. We have initialized the var *thread-local-state* with the vector [1 2 3], but it's not really required. In case an initial value is not supplied to a def form, then the resulting variable is termed as an unbound var. While the state of a var is confined to the current thread, its declaration is global to the current namespace. In other words, a var defined with the def form will be visible to all threads invoked from the current namespace, but the state of the variable is local to the thread in which it is changed. Thus, vars using the def form are also termed as global vars.

Normally, the def form creates a static var, which can only be redefined by using another def form. Static vars can also be redefined within a scope or context using the with-redefs and with-redefs-fn forms. A dynamic var, however, can be set to a new value after it has been defined by using the binding form, shown as follows:

user> (binding [*thread-local-state* [10 20]]
 (map #(* % %) *thread-local-state*))
(100 400)
user> (map #(* % %) *thread-local-state*)
(1 4 9)

In this example, the binding form changes the value contained in the *thread-local-state* var to the vector [10 20]. This causes the map form in the example to return a different value when called without a binding form surrounding it. Thus, the binding form can be used to temporarily change the state of the vars supplied to it.

The Clojure namespace system will resolve free symbols, or rather variable names, to their values. This process of resolving a variable name to a namespace qualified symbol is termed as interning. Also, a def form will first look for an existing global var depending on the symbol it is passed, and will create one if it hasn't been defined yet. The var form can be used to obtain the fully qualified name of a variable, instead of its current value, as shown here:

user> *thread-local-state*
[1 2 3]
user> (var *thread-local-state*)
#'user/*thread-local-state*

Note

Using the #' symbol is the same as using the var form. For example, #'x is equivalent to (var x).

The with-bindings form is another way to rebind vars. This form accepts a map of var and value pairs as its first argument, followed by the body of the form, shown as follows:

user> (with-bindings {#'*thread-local-state* [10 20]}
 (map #(* % %) *thread-local-state*))
(100 400)
user> (with-bindings {(var *thread-local-state*) [10 20]}
 (map #(* % %) *thread-local-state*))
(100 400)

We can check if a var is bound to any value in the current thread of execution using the thread-bound? predicate, which requires a var to be passed as its only argument:

user> (def ^:dynamic *unbound-var*)
#'user/*unbound-var*
user> (thread-bound? (var *unbound-var*))
false
user> (binding [*unbound-var* 1]
 (thread-bound? (var *unbound-var*)))
true

We can also define vars that are not interned, or local vars, using the with-local-vars form. These vars will not be resolved by the namespace system, and have to be accessed manually using the var-get and var-set functions. These functions can thus be used to create and access mutable variables, as shown in Example 2.5.

Note

Using the at-the-rate symbol (@) with a non-interned var is the same as using the var-get function. For example, if x is a non-interned var, @x is equivalent to (var-get x).

(defn factorial [n]
  (with-local-vars [i n acc 1]
    (while (> @i 0)
      (var-set acc (* @acc @i))
      (var-set i (dec @i)))
    (var-get acc)))

Example 2.5: Mutable variables using the with-local-vars form

The factorial function defined in Example 2.5 calculated the factorial of n using two mutable local vars i and acc, which are initialized with the values n and 1 respectively. Note that the code in this function exhibits an imperative style of programming, in which the state of the variables i and acc is manipulated using the var-get and var-set functions.

Note

We can check whether a value has been created through a with-local-vars form using the var? predicate.

Using refs

A Software Transactional Memory (STM) system can also be used to model mutable state. STM essentially treats mutable state as a tiny database that resides in a program's memory. Clojure provides an STM implementation through refs, and they can only be changed within a transaction. Refs are a reference type that represent synchronous and coordinated state.

Note

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

We can create a ref by using the ref function, which requires a single argument to indicate the initial state of the ref. For example, we can create a ref as follows:

(def state (ref 0))

The variable state defined here represents a ref with the initial value of 0. We can dereference state using @ or deref to obtain the value contained in it.

In order to modify a ref, we must start a transaction by using the dosync form. If two concurrent tasks invoke transactions using the dosync form simultaneously, then the transaction that completes first will update the ref successfully. The transaction which completes later will be retried until it completes successfully. Thus, I/O and other side-effects must be avoided within a dosync form, as it can be retried. Within a transaction, we can modify the value of a ref using the ref-set function. This function takes two arguments—a ref and the value that represents the new state of the ref. The ref-set function can be used to modify a ref as follows:

user> @state
0
user> (dosync (ref-set state 1))
1
user> @state
1

Initially, the expression @state returns 0, which is the initial state of the ref state. The value returned by this expression changes after the call to ref-set within the dosync form.

We can obtain the latest value contained in a ref by using the ensure function. This function returns the latest value of a ref, and has to be called within a transaction. For example, the expression (ensure state), when called within a transaction initiated by a dosync form, will return the latest value of the ref state in the transaction.

A more idiomatic way to modify a given ref is by using the alter and commute functions. Both these functions require a ref and a function to be passed to it as arguments. The alter and commute functions will apply the supplied function to the value contained in a given ref, and save the resulting value into the ref. We can also specify additional arguments to pass to the supplied function. For example, we can modify the state of the ref state using alter and commute as follows:

user> @state
1
user> (dosync (alter state + 2))
3
user> (dosync (commute state + 2))
5

The preceding transactions with the alter and commute forms will save the value (+ @state 2) into the ref state. The main difference between alter and commute is that a commute form must be preferred when the supplied function is commutative. This means two successive calls of the function supplied to a commute form must produce the same result regardless of the ordering among the two calls. Using the commute form is considered an optimization over the alter form in which we are not concerned with the ordering among concurrent transactions on a given ref.

Note

The ref-set, alter, and commute functions all return the new value contained in the supplied ref. Also, these functions will throw an error if they are not called within a dosync form.

A mutation performed by the alter and commute forms can also be validated. This is achieved using the :validator key option when creating a ref, as shown here:

user> (def r (ref 1 :validator pos?))
#'user/r
user> (dosync (alter r (fn [_] -1)))
IllegalStateException Invalid reference state  clojure.lang.ARef.validate (ARef.java:33)
user> (dosync (alter r (fn [_] 2)))
2

As shown previously, the ref r throws an exception when we try to change its state to a negative value. This is because the pos? function is used to validate the new state of the ref. Note that the :validator key option can be used with other reference types as well. We can also set the validation function of a ref that was created without a :validator key option using the set-validator! function.

Note

The :validator key option and the set-validator! function can be used with all reference types. The supplied validation function must return false or throw an exception to indicate a validation error.

The dining philosophers problem depicts the use of synchronization primitives to share resources. The problem can be defined as follows: five philosophers are seated on a round table to eat spaghetti, and each philosopher requires two forks to eat from his plate of spaghetti. There are five forks on the table, placed in between the five philosophers. A philosopher will first have to pick up a fork from his left side as well as one from his right side before he can eat. When a philosopher cannot obtain the two forks to his left and right side, he must wait until both the forks are available. After a philosopher is done eating his spaghetti, he will think for some time, thereby allowing the other philosophers to use the forks that he used. The solution to this problem requires that all philosophers share the forks among them, and none of the philosophers starve due to being unable to get two forks. The five philosophers' plates and forks are placed on the table as illustrated in the following diagram:

A philosopher must obtain exclusive access to the forks on his left and right side before he starts eating. If both the forks are unavailable, the philosopher must wait for some time for either one of the forks to be free, and retry obtaining the forks. This way, each philosopher can access the forks in tandem with the other philosophers and avoid starvation.

Generally, this solution can be implemented by using synchronization primitives to access the available forks. Refs allow us to implement a solution to the dining philosophers problem without the use of any synchronization primitives. We will now demonstrate how we can implement and simulate a solution to this problem in Clojure. Firstly, we will have to define the states of a fork and a philosopher as refs, as shown in Example 2.6:

(defn make-fork []
  (ref true))

(defn make-philosopher [name forks food]
  (ref {:name name
        :forks forks
        :eating? false
        :food food}))

Example 2.6: The dining philosophers problem using refs

The make-fork and make-philosopher functions create refs to represent the states of a fork and a philosopher, respectively. A fork is simply the state of a Boolean value, indicating whether it is available or not. And a philosopher, created by the make-philosopher function, is a map encapsulated as a state, which has the following keys:

• The :name key contains the name of a philosopher that is a string value.
• The :forks key points to the forks on the left and the right side of a philosopher. Each fork will be a ref created by the make-fork function.
• The :eating? key indicates whether a philosopher is eating at the moment. It is a Boolean value.
• The :food key represents the amount of food available to a philosopher. For simplicity, we will treat this value as an integer.

Now, let's define some primitive operations to help in handling forks, as shown in Example 2.7:

(defn has-forks? [p]
  (every? true? (map ensure (:forks @p))))

(defn update-forks [p]
  (doseq [f (:forks @p)]
    (commute f not))
  p)

Example 2.7: The dining philosophers problem using refs (continued)

The has-forks? function defined previously checks whether both the forks that are placed to the left and right of a given philosopher ref p are available. The update-forks function will modify the state of both the associated forks of a philosopher ref p using a commute form, and returns the ref p. Obviously, these functions can only be called within a transaction created by the dosync form, since they use the ensure and commute functions. Next, we will have to define some functions to initiate transactions and invoke the has-forks? and update-forks functions for a given philosopher, as shown in Example 2.8:

(defn start-eating [p]
  (dosync
   (when (has-forks? p)
     (update-forks p)
     (commute p assoc :eating? true)
     (commute p update-in [:food] dec))))

(defn stop-eating [p]
  (dosync
   (when (:eating? @p)
     (commute p assoc :eating? false)
     (update-forks p))))

(defn dine [p retry-ms max-eat-ms max-think-ms]
  (while (pos? (:food @p))
    (if (start-eating p)
      (do
        (Thread/sleep (rand-int max-eat-ms))
        (stop-eating p)
        (Thread/sleep (rand-int max-think-ms)))
      (Thread/sleep retry-ms))))

Example 2.8: The dining philosophers problem using refs (continued)

The heart of the solution to the dining philosophers problem is the start-eating function in Example 2.8. This function will check whether both the forks on either side of a philosopher are available, using the has-forks? function. The start-eating function will then proceed to update the states of these forks by calling the update-forks function. The start-eating function will also change the state of the philosopher ref p by invoking commute with the assoc and update-in functions, which both return a new map. Since the start-eating function uses a when form, it will return nil when any of the philosophers' forks are unavailable. These few steps are the solution; in a nutshell, a philosopher will eat only when both his forks are available.

The stop-eating function in Example 2.8 reverses the state of a given philosopher ref after the start-eating function has been invoked on it. This function basically sets the :eating key of the map contained in the supplied philosopher ref p to false using a commute form, and then calls update-forks to reset the state of the associated forks of the philosopher ref p.

The start-eating and stop-eating function can be called repeatedly in a loop using a while form, as long as the :food key of a philosopher ref p, or rather the amount of available food, is a positive value. This is performed by the dine function in Example 2.8. This function will call the start-eating function on a philosopher ref p, and will wait for some time if the philosopher's forks are being used by any other philosophers. The amount of time that a philosopher waits for is indicated by the retry-ms argument that is passed to the dine function. If a philosopher's forks are available, he eats for a random amount of time, as indicated by the expression (rand-int max-eat-ms). Then, the stop-eating function is called to reset the state of the philosopher ref p and the forks that it contains. Finally, the dine function waits for a random amount of time, which is represented by the (rand-int max-think-ms) expression, to indicate that a philosopher is thinking.

Let's now define some function and actually create some refs representing philosophers and associated forks, as shown in Example 2.9:

(defn init-forks [nf]
  (repeatedly nf #(make-fork)))

(defn init-philosophers [np food forks init-fn]
  (let [p-range (range np)
        p-names (map #(str "Philosopher " (inc %))
                     p-range)
        p-forks (map #(vector (nth forks %)
                              (nth forks (-> % inc (mod np))))
                     p-range)
        p-food (cycle [food])]
    (map init-fn p-names p-forks p-food)))

Example 2.9: The dining philosophers problem using refs (continued)

The init-forks function from Example 2.9 will simply invoke the make-fork function a number of times, as indicated by its argument nf. The init-philosophers function will create np number of philosophers and associate each of them with a vector of two forks and a certain amount of food. This is done by mapping the function init-fn, which is a function that matches the arity of the make-philosopher function in Example 2.6, over a range of philosopher names p-names and forks p-forks, and an infinite range p-food of the value food.

We will now define a function to print the collective state of a sequence of philosophers. This can be done in a fairly simple manner using the doseq function, as shown in Example 2.10:

(defn check-philosophers [philosophers forks]
  (doseq [i (range (count philosophers))]
    (println (str "Fork:\t\t\t available=" @(nth forks i)))
    (if-let [p @(nth philosophers i)]
      (println (str (:name p)
                    ":\t\t eating=" (:eating? p)
                    " food=" (:food p))))))

Example 2.10: The dining philosophers problem using refs (continued)

The check-philosophers function in Example 2.10 iterates through all of its supplied philosopher refs, represented by philosophers, and associated forks, represented by forks, and prints their state. The if-let form is used here to check if a dereferenced ref from the collection philosophers is not nil.

Now, let's define a function to concurrently invoke the dine function over a collection of philosopher. This function could also pass in values for the retry-ms, max-eat-ms, and max-think-ms arguments of the dine function. This is implemented in the dine-philosophers function in Example 2.11:

(defn dine-philosophers [philosophers]
  (doall (for [p philosophers]
           (future (dine p 10 100 100))))

Example 2.11: The dining philosophers problem using refs (continued)

Finally, let's define five instances of philosophers and five associated forks for our simulation, using the init-forks, init-philosophers, and make-philosopher functions, as shown in Example 2.12 as follows:

(def all-forks (init-forks 5))

(def all-philosophers
  (init-philosophers 5 1000 all-forks make-philosopher))

Example 2.12: The dining philosophers problem using refs (continued)

We can now use the check-philosopher function to print the state of the philosopher and fork refs created in Example 2.12, as shown here:

user> (check-philosophers all-philosophers all-forks)
Fork:                         available=true
Philosopher 1:                eating=false food=1000
Fork:                         available=true
Philosopher 2:                eating=false food=1000
Fork:                         available=true
Philosopher 3:                eating=false food=1000
Fork:                         available=true
Philosopher 4:                eating=false food=1000
Fork:                         available=true
Philosopher 5:                eating=false food=1000
nil

Initially, all of the forks are available and none of the philosophers are eating. To start the simulation, we must call the dine-philosophers function on the philosopher refs all-philosophers and the fork refs all-forks, as shown here:

user> (def philosophers-futures (dine-philosophers all-philosophers))
#'user/philosophers-futures
user> (check-philosophers all-philosophers all-forks)
Fork:                         available=false
Philosopher 1:                eating=true food=978
Fork:                         available=false
Philosopher 2:                eating=false food=979
Fork:                         available=false
Philosopher 3:                eating=true food=977
Fork:                         available=false
Philosopher 4:                eating=false food=980
Fork:                         available=true
Philosopher 5:                eating=false food=980
nil

After invoking the dine-philosophers function, each philosopher is observed to consume the allocated food, as shown in the output of the previous check-philosophers function. At any given point of time, one or two philosophers are observed to be eating, and the other philosophers will wait until they complete using the available forks. Subsequent calls to the check-philosophers function also indicate the same output, and the philosophers will eventually consume all of the allocated food:

user> (check-philosophers all-philosophers all-forks)
Fork:                         available=true
Philosopher 1:                eating=false food=932
Fork:                         available=true
Philosopher 2:                eating=false food=935
Fork:                         available=true
Philosopher 3:                eating=false food=933
Fork:                         available=true
Philosopher 4:                eating=false food=942
Fork:                         available=true
Philosopher 5:                eating=false food=935
nil

We can pause the simulation by calling the future-cancel function, as shown here. Once the simulation is paused, it can be resumed by calling the dine-philosophers function again, as (dine-philosophers all-philosophers):

user> (map future-cancel philosophers-futures)
(true true true true true)

To summarize, the preceding example is a concise and working implementation of a solution to the dining philosophers problem using Clojure futures and refs.

Using atoms

Atoms are used to handle state that changes atomically. Once an atom is modified, its new value is reflected in all concurrent threads. In this way, atoms represent synchronous and independent state. Let's quickly explore the functions that can be used to handle atoms.

Note

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

We can define an atom using the atom function, which requires the initial state of the atom to be passed to it as the first argument, as shown here:

(def state (atom 0))

The reset! and swap! functions can be used to modify the state of an atom. The reset! function is used to directly set the state of an atom. This function takes two arguments—an atom and the value that represents the new state of the atom, as shown here:

user> @state
0
user> (reset! state 1)
1
user> @state
1

The swap! function requires a function and additional arguments to pass to the supplied function as arguments. The supplied function is applied to the value contained in the atom along with the other additional arguments specified to the swap! function. This function can thus be used to mutate an atom using a supplied function, as shown here:

user> @state
1
user> (swap! state + 2)
3

The call to the preceding swap! function sets the state of the atom to the result of the expression (+ @state 2). The swap! function may call the function + multiple times due to concurrent calls to the swap! function on the atom state. Hence, functions that are passed to the swap! function must be free of I/O and other side effects.

Note

The reset! and swap! functions both return the new value contained in the supplied atom.

We can watch for any change in an atom, and other reference types as well, using the add-watch function. This function will call a given function whenever the state of an atom is changed. The add-watch function takes three arguments—a reference, a key and a watch function, that is, a function that must be called whenever the state of the supplied reference type is changed. The function that is supplied to the add-watch function must accept four arguments—a key, the reference that was changed, the old value of the reference, and the new value of the reference. The value of the key argument that is passed to the add-watch function gets passed to the watch function as its first argument. A watch function can also be unlinked from a given reference type using the remove-watch function. The remove-watch function accepts two arguments—a reference and a key that was specified while adding a watch function to the reference. Example 2.13 depicts how we can track the state of an atom using a watch function:

(defn make-state-with-watch []
  (let [state (atom 0)
        state-is-changed? (atom false)
        watch-fn (fn [key r old-value new-value]
                   (swap! state-is-changed? (fn [_] true))]
    (add-watch state nil watch-fn)
    [state
     state-is-changed?]))

Example 2.13: Using the add-watch function

The make-state-with-watch function defined in Example 2.13 returns a vector of two atoms. The second atom in this vector initially contains the value false. Whenever the state of the first atom in the vector returned by the make-state-with-watch function is changed, the state of the second atom in this vector is changed to the value true. This can be verified in the REPL, as shown here:

user> (def s (make-state-with-watch))
#'user/s
user> @(nth s 1)
false
user> (swap! (nth s 0) inc)
1
user> @(nth s 1)
true

Thus, watch functions can be used with the add-watch function to track the state of atoms and other reference types.

Note

The add-watch function can be used with all reference types.

Using agents

An agent is used to represent state that is associated with a queue of actions and a pool of worker threads. Any action that modifies the state of an agent must be sent to its queue, and the supplied function will be called by a thread selected from the agent's pool of worker threads. We can send actions asynchronously to agents as well. Thus, agents represent asynchronous and independent state.

Note

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

An agent is created using the agent function. For example, we can create an agent with an empty map as its initial value as follows:

(def state (agent {}))

We can modify the state of an agent by using the send and send-off functions. The send and send-off functions will send a supplied action and its additional arguments to an agent's queue in an asynchronous manner. Both these functions return the agent they are passed immediately.

The primary difference between the send and send-off functions is that the send function assigns actions to a thread selection from a pool of worker threads, whereas the send-off function creates a new dedicated thread to execute each action. Blocking actions that are sent to an agent using the send function could exhaust the agent's pool of threads. Thus, the send-off function is preferred for sending blocking actions to an agent.

To demonstrate the send and send-off functions, let's first define a function that returns a closure that sleeps for a certain amount of time, and then, call the assoc function, as shown in Example 2.14:

(defn set-value-in-ms [n ms]
  (fn [a]
    (Thread/sleep ms)
    (assoc a :value n)))

Example 2.14: A function that returns a closure which sleeps and calls assoc

A closure returned by the set-value-in-ms function, in Example 2.14, can be passed as an action to the send and send-off functions, as shown here:

user> (send state (set-value-in-ms 5 5000))
#<Agent@7fce18: {}>
user> (send-off state (set-value-in-ms 10 5000))
#<Agent@7fce18: {}>
user> @state
{}
user> @state ; after 5 seconds
{:value 5}
user> @state ; after another 5 seconds
{:value 10}

The calls to the preceding send and send-off functions will call the closures returned by the set-value-in-ms function, from Example 2.14, asynchronously over the agent state. The agent's state changes over a period of 10 seconds, which is required to execute the closures returned by the set-value-in-ms function. The new key-value pair {:value 5} is observed to be saved into the agent state after five seconds, and the state of the agent again changes to {:value 10} after another five seconds.

Any action that is passed to the send and send-off functions can use the *agent* var to access the agent through which the action will be executed.

The await function can be used to wait for all actions in an agent's queue to be completed, as shown here:

user> (send-off state (set-value-in-ms 100 3000))
#<Agent@af9ac: {:value 10}>
user> (await state) ; will block
nil
user> @state
{:value 100}

The expression (await state) is observed to be blocked until the previous action that was sent to the agent state using the send-off function is completed. The await-for function is a variant of await, which waits for a certain number of milliseconds, indicated by its first argument, for all the actions on an agent, its second argument, to complete.

An agent also saves any error it encounters while performing the actions in its queue. An agent will throw the error it has encountered on any subsequent calls to the send and send-off functions. The error saved by an agent can be accessed using the agent-error function, and can be cleared using the clear-agent-errors function, as shown here:

user> (def a (agent 1))
#'user/a
user> (send a / 0)
#<Agent@5d29f1: 1>
user> (agent-error a)
#<ArithmeticException java.lang.ArithmeticException: Divide by zero>
user> (clear-agent-errors a)
1
user> (agent-error a)
nil
user> @a
1

An agent that has encountered an error can also be restarted using the restart-agent function. This function takes an agent as its first argument and the new state of the agent as its second argument. All actions that were sent to an agent while it was failed will be executed once the restart-agent is called on the agent. We can avoid this behavior by passing the :clear-actions true optional argument to the restart-agent function. In this case, any actions held in an agent's queue are discarded before it is restarted.

To create a pool of threads, or a threadpool, to use with an agent, we must call the static newFixedThreadPool method of the java.util.concurrent.Executors class by passing the desired number of threads in the pool as an argument, as follows:

(def pool (java.util.concurrent.Executors/newFixedThreadPool 10))

The pool of threads defined previously can be used to execute the actions of an agent by using the send-via function. This function is a variant of the send function that accepts a pool of threads, such as the pool defined previously, as its first argument, as shown here:

user> (send-via pool state assoc :value 1000)
#<Agent@8efada: {:value 100}>
user> @state
{:value 1000}

We can also specify the thread pools to be used by all agents to execute actions sent to them using the send and send-off functions using the set-agent-send-executor! and set-agent-send-off-executor! functions respectively. Both of these functions accept a single argument representing a pool of threads.

All agents in the current process can be stopped by invoking the (shutdown-agents). The shutdown-agents function should only be called before exiting a process, as there is no way to restart the agents in a process after calling this function.

Now, let's try implementing the dining philosophers problem using agents. We can reuse most of the functions from the previous implementation of the dining philosophers problem that was based on refs. Let's define some functions to model this problem using agents, as shown in Example 2.15:

(defn make-philosopher-agent [name forks food]
  (agent {:name name
          :forks forks
          :eating? false
          :food food}))

(defn start-eating [max-eat-ms]
  (dosync (if (has-forks? *agent*)
            (do
              (-> *agent*
                  update-forks
                  (send assoc :eating? true)
                  (send update-in [:food] dec))
              (Thread/sleep (rand-int max-eat-ms))))))

(defn stop-eating [max-think-ms]
  (dosync (-> *agent*
              (send assoc :eating? false)
              update-forks))
  (Thread/sleep (rand-int max-think-ms)))

(def running? (atom true))

(defn dine [p max-eat-ms max-think-ms]
  (when (and p (pos? (:food p)))
    (if-not (:eating? p)
      (start-eating max-eat-ms)
      (stop-eating max-think-ms))
    (if-not @running?
      @*agent*
      @(send-off *agent* dine max-eat-ms max-think-ms))))

(defn dine-philosophers [philosophers]
  (swap! running? (fn [_] true))
  (doall (for [p philosophers]
           (send-off p dine 100 100))))

(defn stop-philosophers []
  (swap! running? (fn [_] false)))

Example 2.15: The dining philosophers problem using agents

In Example 2.15, the make-philosopher-agent function will create an agent representing a philosopher. The initial state of the resulting agent is a map of the keys :name, :forks, :eating?, and :food, as described in the previous implementation of the dining philosophers problem. Note that the forks in this implementation are still represented by refs.

The start-eating function in Example 2.15 will start a transaction, check whether the forks placed to the left and right sides of a philosopher are available, changes the state of the forks and philosopher agent accordingly, and then suspends the current thread for some time to indicate that a philosopher is eating. The stop-eating function in Example 2.15 will similarly update the state of a philosopher and the forks he had used, and then suspend the current thread for some time to indicate that a philosopher is thinking. Note that both the start-eating and stop-eating functions reuse the has-forks? and update-forks functions from Example 2.7 of the previous implementation of the dining philosophers problem.

The start-eating and stop-eating functions are called by the dine function in Example 2.15. We can assume that this function will be passed as an action to a philosopher agent. This function checks the value of the :eating? key contained in a philosopher agent to decide whether it must invoke the start-eating or stop-eating function in the current call. Next, the dine function invokes itself again using the send-off function and dereferencing the agent returned by the send-off function. The dine function also checks the state of the atom running? and does not invoke itself through the send-off function in case the expression @running returns false.

The dine-philosophers function in Example 2.15 starts the simulation by setting the value of the running? atom to true and then invoking the dine function asynchronously through the send-off function for all the philosopher agents passed to it, represented by philosophers. The function stop-philosophers simply sets the value of the running? atom to false, thereby stopping the simulation.

Finally, let's define five instances of forks and philosophers using the init-forks and init-philosophers functions from Example 2.9, shown in Example 2.16 as follows:

(def all-forks (init-forks 5))

(def all-philosophers
  (init-philosophers 5 1000 all-forks make-philosopher-agent))

Example 2.16: The dining philosophers problem using agents (continued)

We can now start the simulation by calling the dine-philosophers function. Also, we can print the collective state of the fork and philosopher instances in the simulation using the check-philosophers function defined in Example 2.10, as follows:

user> (def philosophers-agents (dine-philosophers all-philosophers))
#'user/philosophers-agents
user> (check-philosophers all-philosophers all-forks)
Fork:                    available=false
Philosopher 1:           eating=false food=936
Fork:                    available=false
Philosopher 2:           eating=false food=942
Fork:                    available=true
Philosopher 3:           eating=true food=942
Fork:                    available=true
Philosopher 4:           eating=false food=935
Fork:                    available=true
Philosopher 5:           eating=true food=943
nil
user> (check-philosophers all-philosophers all-forks)
Fork:                    available=false
Philosopher 1:           eating=true food=743
Fork:                    available=false
Philosopher 2:           eating=false food=747
Fork:                    available=true
Philosopher 3:           eating=false food=751
Fork:                    available=true
Philosopher 4:           eating=false food=741
Fork:                    available=true
Philosopher 5:           eating=false food=760
nil

As shown in the preceding output, all philosopher agents share the fork instances among themselves. In effect, they work in tandem to ensure that each philosopher eventually consumes all of their allocated food.

In summary, vars, refs, atoms, and agents can be used to represent mutable state that is shared among concurrently executing tasks.