Combining CRDTs and MPI (crdts.rs and rsmpi crates)

[DaveLanday]

I am looking to combine CRDTs and MPI using the rsmpi crate. A previous discussion (#139) mentioned that we must serialize/deserialize the operations we want to send or receive. In my "hello-world" example using crdts and mpi, I am testing the use of a CmRDT::GCounter that is incremented independently by different process ranks that own a replica of it. How would I serialize a Dot<A> or the output of local_counter.inc(&rank) to send to the process at the next_rank and receive from the process at the prev_rank?

What I have tried is to create struct that contains actor: i32 and counter: u64 fields and implements the Equivalence trait that formats rust types to MPI types (ie: i32 -> MPI_i32 etc...) and then do local_counter.apply(Dot::new(&remote_actor, remote_counter)) after receiving from the process at the pervious rank (prev_rank). But this does not seem to work. I am guessing I need serialization, but unsure what I should do. Any guidance would be appreciated

[davidrusu]

Hi Dave,

I believe the simplest way to hook this up would be to use a lightweight serialization format like bincode.

let op = local_counter.inc(&rank);
let op_bytes: Vec<u8> = bincode::serialize(&ob).unwrap();

// you can then transmit op_bytes over the MPI communication channels

// on the receiving side

let decoded_op: Dot<mpi::Rank> = bincode::deserialize(&op_bytes).unwrap();
let local_counter.apply(decoded_op)

Bincode is very lightweight so you should see minimal overhead in encoding/decoding. Plus you can avoid having to write your own Equivalence implementations and avoid conversion bugs there.

Let me know if this works for you.

David

[DaveLanday]

@davidrusu another general question I have comes from when I apply an op sent from another actor to a replica of a GCounter. Say I have the following:

// replica A
local_counter.apply(local_counter.inc(&process_rank_A));

// Receive op from remote replica B and apply it:
local_count.apply(remote_op);

And below, remote_op is the remote's copy of local_counter. From the remote process' point of view, incrementing it's count would look like

// From remote replica B's point of view:
let op = local_counter.inc(&remote_process_rank_B); 
local_counter.apply(op.clone()); 
// send op to replica A

or something like that.
Anyway, my question is: after A applies the remote op sent from B, A's counter might look something like:

{process_rank_A: n, remote_process_rank_B: 1}

where n is a u64 and local_counter.read() would return n+1 as the value of the GCounter. In what case would read return just n in the case where adding B's op caused a conflict? Should I be using a common actor name instead for each replica of my counter to catch conflicts? For example, both replicas A and B would create ops that look like the following: local_counter.inc(&common_rank) ?

Thanks again!

[davidrusu]

This is difficult to answer generically.

The GCounter is meant to be a distributed counter. The use cases are things like the following:

You have group of workers and you want an eventually consistent view of how many tasks the group did, each worker increments it's local counter using it's own identifier when it performs a task and shares the op with the others who apply it locally.

If you query any individual worker, you'll get some lower bound on number of tasks completed by the group.

---

If your use-case doesn't map cleanly onto this case, it may be that the gcounter is the wrong CRDT here.

If you can share more details about what state you are trying to represent with the gcounter then maybe I can point you towards a CRDT that fits your problem better.

[DaveLanday]

Hi @davidrusu, thank you again for the reply. The way you describe it is exactly how I intended to use it; I want leverage CRDTs to provide independent processes with a global view of the work that other collective processes have done (such as number of problems computed by each process, number of tasks each process is yet to do (i.e: local queue size) etc...). I think that they can possibly be leveraged to inform load-balancers in distributed compute (i.e collaborative computer tasks and not text-editors/ applications intended for human-2-human collaboration). I was also hoping to add custom resolution rules to some of the existing rules around registers for my eventual use case. For example, is there such a thing as a grow-only register? My thought was that a register could represent a monotonically increasing/decreasing floating point value with greater-value-wins or lower-value-wins semantics which could be useful for, say plotting a loss-function in an ML application or keeping track of some best-found result in a large distributed computation.

I am in the process of fixing my "hello world"-like example using GCounter and rsmpi (I was introducing a race condition with the way I was doing send/recv apparently), but will post a working version hopefully today.

[DaveLanday]

Ok! Here is some working code to do point-to-point MPI sending inc operations to a replicated counter. I think it would make more sense (with resepect to Ops based CRDTs) to have all-to-all type broadcast or define a topology where eventually every message/op is seen by every replica, but I hope this serves as a naive example:

use bincode;
#[allow(unused_imports)]
use crdts::{CmRDT, Dot, GCounter};
use mpi::request::WaitGuard;
use mpi::traits::*;
use rand::{thread_rng, Rng};
use std::{thread, time};

fn main() {
    #[allow(unused_variables)]
    // number of fake jobs to process (i.e: count up to this number)
    const NJOBS: u16 = 2000;

    // Represents the worst time job can take to finish
    let dur = time::Duration::from_millis(25);

    // Represents the probability of performing a large job:
    const P: f64 = 0.015; // 1.5% chance of a long running job (i.e: 25 milliseconds)
    let mut rng = thread_rng();

    // Init MPI
    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    // Get the rank and size of the communicator
    let size = world.size();
    let rank = world.rank();

    // Define the processor that sends to this rank and  that receives from this rank:
    let prev_rank = (rank + size - 1) % size;
    let next_rank = (rank + 1) % size;

    // Init local replica of the count
    let mut local_counter = GCounter::new();

    println!(
        "Prev rank: {}, This rank: {}, Next rank: {}",
        prev_rank, rank, next_rank
    );
    for _n in 0..NJOBS {
        // simulate a long or short process between incrementing the count:
        if rng.gen_bool(P) {
            thread::sleep(dur);
            // println!("Running a long job on processor {}", rank);
        }

        // Declare intent to increment by creating an `Op`. Use rank as the uid of the replica
        let inc_op = local_counter.inc(rank);

        // Apply the operation to our local counter first and then send the `Op` off to others
        local_counter.apply(inc_op.clone()); // Apply the increment:

        // Serialize the local operation for sending:
        let send_op: Vec<u8> = bincode::serialize(&inc_op).unwrap();

        // For getting the most up-to-date count of the remote counter
        let mut remote_op = vec![0u8; 24]; // the buffer must provide a size

        // Simple point-to-point messaging in a ring configuration:
        mpi::request::scope(|scope| {
            let _rreq = WaitGuard::from(
                world
                    .process_at_rank(prev_rank)
                    .immediate_receive_into(scope, &mut remote_op),
            );
            let _sreq = WaitGuard::from(
                world
                    .process_at_rank(next_rank)
                    .immediate_send(scope, &send_op),
            );
        });

        // Deserialize the remote_op:
        let recv_op: Dot<mpi::Rank> = bincode::deserialize(&remote_op).unwrap();

        // Apply the change to the local counter:
        local_counter.apply(recv_op);
    }
    println!(
        "Processor {} says the count = {:?}",
        rank,
        local_counter.read()
    );
}

The output looks like this:

❯ mpirun -np 6 target/debug/mpi_example
Prev rank: 1, This rank: 2, Next rank: 3
Prev rank: 5, This rank: 0, Next rank: 1
Prev rank: 4, This rank: 5, Next rank: 0
Prev rank: 2, This rank: 3, Next rank: 4
Prev rank: 3, This rank: 4, Next rank: 5
Prev rank: 0, This rank: 1, Next rank: 2
Processor 0 says the count = 4000
Processor 1 says the count = 4000
Processor 2 says the count = 4000
Processor 3 says the count = 4000
Processor 4 says the count = 4000
Processor 5 says the count = 4000

[davidrusu]

Great! glad you got it working

[DaveLanday]

Thank you for all of your help @davidrusu ! My next step is figuring out the best way to broadcast Ops for the application I want to build. If you are familiar with MPI and have any suggestions for how to properly broadcast Ops, I am all ears.