Gray Soft / Tags / Concurrency

6
SEP
2014
Taking Rust to Task

Now that I've reached the point where I can get some Rust code running without asking questions in IRC every five minutes, I really wanted to play with some tasks. Tasks are the way Rust handles multiprocessing code. Under the hood they can map one-to-one with operating system threads or you can use a many-to-one mapping that I'm not ready to go into yet.

Probably one of the most exciting aspect of tasks in Rust, in my opinion, is that unsafe use of shared memory is rejected outright as a compile error. That lead me to want to figure out how you communicate correctly. (Spoiler: the same was you do in Ruby: just pass messages.)

Ready to dive in, I grossly simplified a recent challenge from work and coded it up in Rust. You can get the idea with a glance at main():
```
use std::collections::HashMap;

// ...

fn string_vec(strs: &[&'static str]) -> Vec<String> {
    let mut v = Vec::new();
    for s in strs.iter() {
        v.push(s.to_string());
    }
    v
}

fn main() {
    let mut services = HashMap::new();
    services.insert("S1".to_string(), string_vec(["A", "B"]));
    services.insert("S2".to_string(), string_vec(["A", "C"]));
    services.insert("S3".to_string(), string_vec(["C", "D", "E", "F"]));
    services.insert("S4".to_string(), string_vec(["D", "B"]));
    services.insert("S5".to_string(), string_vec(["A", "Z"]));

    let work = Work(Search::new("A".to_string(), "B".to_string()));

    let mut task_manager = TaskManager::new(services);
    task_manager.run(work);
}
```
Read more…
In: Rusting | Tags: Concurrency & Rust | 1 Comment

22

AUG
2014

Sleepy Programs

When we think of real multiprocessing, our thoughts probably drift more towards languages like Erlang, Go, Clojure, or Rust. Such languages really focus on getting separate "processes" to communicate via messages. This makes it a lot easier to know when one process is waiting on another, because calls to receive messages typically block until one is available.

But what about Ruby? Can we do intelligent process coordination in Ruby?

Yes, we can. The tools for it are more awkward though. It's easy to run into tricky edge cases and hard to code your way out of them correctly.

Let's play with an example to see how good we can make things. Here's what we will do:

We will start one parent process that will fork() a single child process
The child will push three messages onto a RabbitMQ queue and exit()
The parent will listen for three messages to arrive, then exit()

Here's a somewhat sloppy first attempt at solving this:

#!/usr/bin/env ruby

require "benchmark"

require "bunny"

QUEUE_NAME = "example"
MESSAGES   = %w[first second third]

def send_messages(*messages)
  connection = Bunny.new.tap(&:start)
  exchange   = connection.create_channel.default_exchange

  messages.each do |message|
    exchange.publish(message, routing_key: QUEUE_NAME)
  end

  connection.close
end

def listen_for_messages(received_messages)
  connection = Bunny.new.tap(&:start)
  queue      = connection.create_channel.queue(QUEUE_NAME, auto_delete: true)

  queue.subscribe do |delivery_info, metadata, payload|
    received_messages << payload
  end

  time_it("Received #{MESSAGES.size} messages") do
    yield
  end

  connection.close
end

def time_it(name)
  elapsed = Benchmark.realtime do
    yield
  end
  puts "%s: %.2fs" % [name, elapsed]
end

def wait_for_messages(received_messages)
  until received_messages == MESSAGES
    sleep 0.1  # don't peg the CPU while we wait
  end
end

def send_and_receive
  pid = fork do
    sleep 3  # make sure we're receiving before they are sent
    send_messages(*MESSAGES)
  end
  Process.detach(pid)

  received_messages = [ ]
  listen_for_messages(received_messages) do
    wait_for_messages(received_messages)
  end
end

send_and_receive

Concurrency

6

Taking Rust to Task

22

Sleepy Programs

13

Erlang Message Passing

27

The Ruby VM: Episode III

Search

Categories

Tags