Concurrency

danger

Concurrency in Fuyu is an active area of research.

Constructs on this page are not part of the language. This page is a draft of how concurrency may be implemented with first-class language support.

A concurrent process is just a function that can be evaluated in a special way. The rules for defining a process are simple:

• The first argument to the process must have type Process a.
• The process can have any number of arguments following that.
• The process must evaluate to ().

A process is similar to a function; however, there are the following differences:

• A select expression can be used within the process body to receive a message sent to the process.
• A process can use the self keyword to get its own ProcessId.
• A process is created by calling it as a function prefixed with spawn, for example, spawn storage (hashmap.new ()) below.
• Within a process, other processes can be called as regular functions only if the message type matches. This allows processes to call themselves recursively.

Spawning a new process with spawn creates a handle to the process that is a ProcessId. For example, a process with type Process String -> () would produce a ProcessId String.

The ProcessId of a process dictates the type of messages that can be sent to it, which are the types of messages that can be received via select in the process body.

import std/assert
import std/hashmap for HashMap
import std/process for ProcessId

/// The types of messages that can be sent to and received by
/// the `storage` process.
type Store k v =
  /// Associate the `key` with the `value`.
  Write { key: k value: v },
  /// Read the value associated with the `key` and send it to the `process_id`.
  Read { key: k process_id: ProcessId v }

/// A process that implements a key-value storage system.
storage: Process (Store k v) -> HashMap k v -> ()
storage data = select
  Write { key, value } -> key_value (data | hashmap.update key value)
  Read { key, process_id } ->
    data | hashmap.try_read key match
      Some value -> process_id | process.send value
      _ -> ()
    storage Process data

/// A process that interacts with a storage process.
store_and_read: Process String -> ()
store_and_read _ = do
  let s = spawn storage (hashmap.new ())
  s | process.send Write { key: 0, value: "zero" }
  s | process.send Write { key: 1, value: "one" }
  s | process.send Write { key: 2, value: "two" }
  s | process.send Read { key: 1, process_id: self }
  select
    x -> assert.eq x "one"

main: () -> ()
main () = do
  let p = spawn store_and_read
  process.wait p