Tasks and channels

A Vx program can do several things at the same time on the CPU. It starts each one as a task. This chapter shows how to start tasks, how they share data, and how they send values to each other.

Each program in this chapter is also a test of the compiler. You can find it under tests/backend/pass/ with the name that its section gives.

A tasks block and spawn

A tasks block holds tasks. spawn { .. } starts a task, and the task runs on a thread of its own. The block ends only after each of its tasks ends. So a task can borrow the locals of the code around it: the locals live longer than the task.

spawn gives a handle to the task. join() waits for the task and gives the value of its last expression. A return in the task ends the task early, and its value is then the value of the task. The value of the block is the value of its last expression too.

This program adds the numbers 1 to 100 with two tasks. Each task borrows the vector v from main and adds one half of it (book_tasks_split_the_work.vx):

import std::vec;

fn sum_range(v : &Vec<i64>, from : i32, to : i32) -> i64 {
  let mut sum : i64 = 0;
  for i in from..to {
    sum = sum + *v.get_ref(i);
  }
  return sum;
}

fn main() -> i32 {
  let mut v = Vec<i64>::new();
  for i in 1..101 {
    v.push(i as i64);
  }
  let total = tasks {
    let low = spawn {
      sum_range(&v, 0, 50)
    };
    let high = spawn {
      sum_range(&v, 50, 100)
    };
    low.join() + high.join()
  };
  println!("total {}", total);
  return 0;
}

It prints total 5050.

A task that fails

A task fails when its value is the Err of a Result. Then its join() leaves the block, and that Err is the value of the block. A join() of a task that did not fail gives the value inside the Ok. So the code of the block works with plain values, and the block gives the first failure (book_tasks_join_a_failure.vx). A ? in a task ends the task with the Err:

import core::result;

// A read that fails for an even number above 2.
fn read(n : i32) -> Result<i32, i32> {
  if n > 2 && n % 2 == 0 {
    return Result<i32, i32>::Err(n);
  }
  return Result<i32, i32>::Ok(n * 10);
}

fn read_two(a : i32, b : i32) -> Result<i32, i32> {
  let r = tasks {
    let x = spawn {
      read(a)
    };
    let y = spawn {
      read(b)
    };
    // Each `join` gives the `Ok` value. An `Err` leaves the block, and is its value.
    let a = x.join();
    let b = y.join();
    Result<i32, i32>::Ok(a + b)
  };
  return r;
}

fn show(a : i32, b : i32) -> void {
  match read_two(a, b) {
    Result<i32, i32>::Ok(v) => {
      println!("read {} and {}: Ok {}", a, b, v);
    }
    Result<i32, i32>::Err(e) => {
      println!("read {} and {}: Err {}", a, b, e);
    }
  }
}

fn main() -> i32 {
  show(1, 2);
  show(1, 4);
  return 0;
}

It prints:

read 1 and 2: Ok 30
read 1 and 4: Err 4

The block must join each task that can fail, in a statement of its own that always runs. Otherwise the compiler gives the error E3060. A task that fails also cancels the other tasks of its block, so that they do not wait for nothing. The section on channels below says what a cancel does.

What a task can take

A task can take a value from the code around it in two ways:

• It moves the value. Then the type of the value must be Send.
• It borrows the value, while the code around it can use the value too. Then the type must be Sync, because two threads can read it at the same time.

The compiler finds both traits from the fields of a type. A number, a bool and a struct of them are Send and Sync. A raw pointer is neither, so a type with a raw pointer field must say what it allows with unsafe impl. The library does this for Vec<T>, the atomics and the types below.

For example, two tasks cannot borrow one Receiver of a channel, because two recv calls at the same time would take the same value. The compiler says:

Error[E3055]: the task reads 'rx', which the code around the task can read at the same time, so
its type 'Receiver<i64>' must be `Sync`: field 'words' of 'Receiver<i64>' has type
'*mut AtomicI64', and '*mut AtomicI64' is a raw pointer, which is not `Sync`

The rule depends only on the types, never on the body of a function. So the compiler can check each function on its own, and in parallel.

Atomics that tasks share

An atomic is a number that many tasks can change at the same time, from core::sync::atomic. Each task here borrows the same counter and adds 1 to it 10000 times (book_tasks_share_an_atomic.vx):

import core::sync::atomic;

fn add_ones(counter : &AtomicI64, n : i32) -> void {
  for _i in 0..n {
    counter.fetch_add(1, AtomicOrdering::Relaxed);
  }
}

fn main() -> i32 {
  let counter = AtomicI64::new(0);
  tasks {
    for _t in 0..4 {
      spawn {
        add_ones(&counter, 10000);
      };
    }
  };
  println!("count {}", counter.load(AtomicOrdering::SeqCst));
  return 0;
}

It prints count 40000. A plain i64 in place of the atomic does not compile: a task cannot write a value that another task can read at the same time.

Channels

A channel moves values from tasks that send to one task that receives. channel<T>(n) makes a channel with room for n values, and gives a Sender<T> and a Receiver<T>:

• send(v) puts v in the channel. It waits while the channel is full.
• recv() takes the next value. It waits while the channel is empty. It gives Closed when the program dropped each Sender and the channel holds no value.
• Clone the Sender to send from many tasks. A channel has one Receiver.

The values of one sender arrive in the order that it sent them. The type T must be Send, and it must not hold a reference: a value in a channel can outlive the data that a reference points to.

In this program, four tasks send squares, and the code of the block receives them (book_channel_from_workers.vx):

import std::sync::channel;

fn send_squares(tx : Sender<i64>, from : i64) -> void {
  for i in from..from + 10 {
    match tx.send(i * i) {
      Result<void, SendError<i64>>::Ok() => {}
      Result<void, SendError<i64>>::Err(_e) => {
        println!("the receiver is gone");
      }
    }
  }
}

fn main() -> i32 {
  let (tx, rx) = channel<i64>(4);
  let mut count : i64 = 0;
  let mut sum : i64 = 0;
  tasks {
    for w in 0..4 {
      let mine = tx.clone();
      spawn {
        send_squares(mine, (w as i64) * 10);
      };
    }
    // The channel closes when the last `Sender` is dropped.
    drop(tx);
    let mut open = true;
    while open {
      match rx.recv() {
        Result<i64, RecvError>::Ok(v) => {
          count = count + 1;
          sum = sum + v;
        }
        Result<i64, RecvError>::Err(_e) => {
          open = false;
        }
      }
    }
  };
  println!("got {} values, sum {}", count, sum);
  return 0;
}

It prints got 40 values, sum 20540.

Cancellation

When the code of a tasks block leaves it early, with return or break, or when a task fails, the block cancels its tasks. A wait that can be cancelled then gives an error at once: recv() gives Cancelled, and send(v) gives v back with cancelled set. The task should then end. The compiler warns about a loop that waits again after a Cancelled (W1035), because each wait in a cancelled task gives Cancelled at once, and such a loop never ends.

A join() and a lock() cannot be cancelled.

When each task waits and nothing can wake one, the program has a deadlock. The runtime then stops the program with the message vx: deadlock: ..., so that it does not hang.

A drop runs with cancellation off, so that a cleanup can finish. So a wait in a drop should have a time limit: send_for(v, seconds) and recv_for(seconds) give timed_out and TimedOut after it. The compiler warns about a send or recv in a drop that has no limit (W1036).

A drop also has a limit of its own. Its type says how long its waits can take, with a method of Drop that returns 10 seconds unless the type writes its own:

#![allow(unused)]
fn main() {
impl Drop for SlotGuard {
  fn drop(self : &mut SlotGuard) -> void {
    // The cleanup. Its waits get the limit below.
  }

  fn wait_limit(self : &SlotGuard) -> f64 {
    return 0.05;
  }
}
}

After the limit, a send or recv in the drop gives up as in a cancelled block. VX_DROP_WAIT_SECONDS can make the limit of each drop shorter, for a test or to find a drop that hangs, but never longer.

within(seconds) { .. } gives the waits in a block one time limit, in a drop or anywhere else. A flush, a close and a send in within(0.05) { .. } share the 50 ms. The limit also holds in the functions that the block calls. After the limit, a wait gives up as in a cancelled block, and cancelled() is true until the block ends. Outside a drop, a cancel still stops the waits in the block.

select

Selector::new(receivers) takes a list of receivers, and select() on it takes the first value from any of them. It gives the value and the index of its channel in the list. It gives Closed when each channel is closed and empty. Each search starts after the channel of the last value, so a channel that always has a value does not hide the others (book_select_from_two_channels.vx):

import std::sync::channel;
import std::vec;

fn send_three(tx : Sender<i64>) -> void {
  let first : i64 = 0;
  for i in first..3 {
    match tx.send(i) {
      Result<void, SendError<i64>>::Ok() => {}
      Result<void, SendError<i64>>::Err(_e) => {}
    }
  }
}

fn main() -> i32 {
  let (tx0, rx0) = channel<i64>(1);
  let (tx1, rx1) = channel<i64>(1);
  let mut rs = Vec<Receiver<i64>>::new();
  rs.push(rx0);
  rs.push(rx1);
  let mut sel = Selector<i64>::new(rs);
  let mut counts : Tensor<i64, [2]> = [0, 0];
  tasks {
    spawn {
      send_three(tx0);
    };
    spawn {
      send_three(tx1);
    };
    let mut open = true;
    while open {
      match sel.select() {
        Result<Selected<i64>, RecvError>::Ok(s) => {
          counts[s.index] = counts[s.index] + 1;
        }
        // Both channels are closed and empty.
        Result<Selected<i64>, RecvError>::Err(_e) => {
          open = false;
        }
      }
    }
  };
  println!("from 0: {} values, from 1: {} values", counts[0], counts[1]);
  return 0;
}

It prints from 0: 3 values, from 1: 3 values.

Mutex

A Mutex<T> guards a value that many tasks change. lock() waits until no other task holds the lock, and gives a guard. get() on the guard gives a pointer to the value. The drop of the guard unlocks the Mutex (book_mutex_guards_a_total.vx):

import std::sync::mutex;

fn add_to(total : &Mutex<i64>, n : i32) -> void {
  for _i in 0..n {
    // The guard unlocks the `Mutex` when it is dropped, at the end of each pass.
    let mut guard = total.lock();
    let value = guard.get();
    *value = *value + 1;
  }
}

fn main() -> i32 {
  let total = Mutex<i64>::new(0);
  tasks {
    for _t in 0..4 {
      spawn {
        add_to(&total, 5000);
      };
    }
  };
  println!("total {}", total.into_inner());
  return 0;
}

It prints total 20000. Use an atomic for one number, and a Mutex when a change touches more than one number.

Arc

A task can borrow the locals of its parent, so most programs need no Arc. Use an Arc<T> when a value must live longer than its tasks block. clone() gives another Arc that shares the value, and get() gives &T. The drop of the last Arc drops the value. It works as Rust’s Arc does. An Arc<T> can move to another task only when T is Send and Sync (book_arc_outlives_its_task.vx):

import std::sync::arc;
import std::sync::channel;

struct Settings {
  scale : i64,
}

fn load(tx : Sender<Arc<Settings>>) -> void {
  let settings = Arc<Settings>::new(Settings {
    scale : 10
  });
  // A clone shares the value and adds 1 to the count of owners.
  match tx.send(settings.clone()) {
    Result<void, SendError<Arc<Settings>>>::Ok() => {}
    Result<void, SendError<Arc<Settings>>>::Err(_e) => {}
  }
  match tx.send(settings) {
    Result<void, SendError<Arc<Settings>>>::Ok() => {}
    Result<void, SendError<Arc<Settings>>>::Err(_e) => {}
  }
}

fn receive(rx : &Receiver<Arc<Settings>>) -> Arc<Settings> {
  match rx.recv() {
    Result<Arc<Settings>, RecvError>::Ok(got) => {
      return got;
    }
    Result<Arc<Settings>, RecvError>::Err(_e) => {
      panic("the task sent nothing");
    }
  }
}

fn main() -> i32 {
  let (tx, rx) = channel<Arc<Settings>>(2);
  tasks {
    spawn {
      load(tx);
    };
  };
  let a = receive(&rx);
  let b = receive(&rx);
  println!("scale {}, owners {}", a.get().scale, a.strong_count());
  // The drop of the last owner drops the value.
  drop(b);
  println!("owners {}", a.strong_count());
  return 0;
}

It prints scale 10, owners 2, and then owners 1. To change a shared value, put a Mutex in the Arc.

Finding a data race at run time

The compiler cannot see a race through a raw pointer in unsafe code. To find one, run the program with vxc --sanitize=thread program.vx. LLVM’s ThreadSanitizer then checks each memory access of the program. It prints WARNING: ThreadSanitizer: data race for each race, and the program fails.

ThreadSanitizer sees the start and the end of a task, the atomics, and the waits of channels and Mutex. This is also true for the runtime that runs many tasks on a few OS threads (VX_TASK_RUNTIME=shared). It finds only the races that happen in that run. The option works only when vxc runs the program. On Ubuntu, the runtime of ThreadSanitizer is in the package libclang-rt-22-dev.

Why Vx has no async

In Vx, a function that waits is a normal function. recv, join, lock and a network read wait for an event, and the task continues after it. A function that calls them is a normal function too. So Vx has no async and no await.

In a language with async, a function that waits must be async, and so must each function that calls it. Such a function has a “color”, and a program must keep the two colors apart. Vx does not have that problem. A borrow that lives across a wait is a borrow across a call, so the borrow checker needs no new rule.

The runtime runs each task on an OS thread of its own today. A later runtime will run many tasks on a few OS threads. The language is the same for both.

Compared with Rust and Go