pipeline

A worker pool on threads and channels (lib/channel.jh), as Go programs write them: go starts several workers that take jobs from one channel until it is closed, and send their results on another.

jihoo build examples/pipeline.jh -o out/pipeline && ./out/pipeline

examples/pipeline.jh

#![native]
import channel
import libc
import sync

const WORKERS = 4
const JOBS = 100

// What the workers share: the channels, and a count of the jobs done.
struct Pool {
    jobs: channel.Chan(i64)
    results: channel.Chan(i64)
    done: i64
    running: sync.Group
}

// The number of steps the Collatz sequence takes from `n` to 1.
fn collatz(n: i64) -> i64 {
    let steps = 0
    while n != 1 {
        if n % 2 == 0 {
            n = n / 2
        } else {
            n = 3 * n + 1
        }
        steps = steps + 1
    }
    return steps
}

// Takes jobs until the channel is closed and empty.
fn worker(pool: *Pool) {
    let more = true
    while more {
        match channel.recv(i64, pool.jobs) {
            Some(n) => {
                channel.send(i64, pool.results, collatz(n))
                atomic_add(&pool.done, 1)
            }
            None => more = false
        }
    }
    sync.group_done(&pool.running)
}

fn run(jobs: channel.Chan(i64), results: channel.Chan(i64)) -> i64 {
    let pool = Pool { jobs: jobs, results: results, done: 0, running: sync.group_new() }
    sync.group_add(&pool.running, WORKERS)
    let i = 0
    while i < WORKERS {
        go worker(&pool)
        i = i + 1
    }

    // Hand out the jobs, then say there are no more.
    let n = 1
    while n <= JOBS {
        channel.send(i64, jobs, n)
        n = n + 1
    }
    channel.close(i64, jobs)

    // Every job has a result; which worker did it does not matter.
    let total = 0
    let longest = 0
    let k = 0
    while k < JOBS {
        match channel.recv(i64, results) {
            Some(steps) => {
                total = total + steps
                if steps > longest {
                    longest = steps
                }
            }
            None => return 1
        }
        k = k + 1
    }
    // The workers use `pool` until they finish: wait for them.
    sync.group_wait(&pool.running)
    libc.printf("%ld jobs done, %ld steps in all, %ld at most\n", atomic_load(&pool.done), total, longest)
    channel.free(i64, jobs)
    channel.free(i64, results)
    return 0
}

fn main() -> i64 {
    match channel.new(i64, 0) {                 // unbuffered: hands jobs over
        Some(jobs) => match channel.new(i64, JOBS) {   // room for every result
            Some(results) => return run(jobs, results)
            None => return 1
        }
        None => return 1
    }
}

View on GitHub