Enum Generic

pub enum Generic {
    Thread(Thread),
    Process(Process),
    ProcessBinary(ProcessAllocator),
    ZeroCopy(TcpAllocator<Process>),
}

Enumerates known implementors of Allocate. Passes trait method calls on to members.

Variants

Thread(Thread)

Intra-thread allocator.

Process(Process)

Inter-thread, intra-process allocator.

ProcessBinary(ProcessAllocator)

Inter-thread, intra-process serializing allocator.

ZeroCopy(TcpAllocator<Process>)

Inter-process allocator.

Implementations

impl Generic

pub fn index(&self) -> usize

The index of the worker out of (0..self.peers()).

Examples found in repository

examples/comm_hello.rs (line 12)

fn main() {

    // extract the configuration from user-supplied arguments, initialize the computation.
    let config = timely_communication::Config::from_args(std::env::args()).unwrap();
    let guards = timely_communication::initialize(config, |mut allocator| {

        println!("worker {} of {} started", allocator.index(), allocator.peers());

        // allocates a pair of senders list and one receiver.
        let (mut senders, mut receiver) = allocator.allocate(0);

        // send typed data along each channel
        for i in 0 .. allocator.peers() {
            senders[i].send(Message::from_typed(format!("hello, {}", i)));
            senders[i].done();
        }

        // no support for termination notification,
        // we have to count down ourselves.
        let mut received = 0;
        while received < allocator.peers() {

            allocator.receive();

            if let Some(message) = receiver.recv() {
                println!("worker {}: received: <{}>", allocator.index(), message.deref());
                received += 1;
            }

            allocator.release();
        }

        allocator.index()
    });

    // computation runs until guards are joined or dropped.
    if let Ok(guards) = guards {
        for guard in guards.join() {
            println!("result: {:?}", guard);
        }
    }
    else { println!("error in computation"); }
}

pub fn peers(&self) -> usize

The number of workers.

Examples found in repository

examples/comm_hello.rs (line 12)

fn main() {

    // extract the configuration from user-supplied arguments, initialize the computation.
    let config = timely_communication::Config::from_args(std::env::args()).unwrap();
    let guards = timely_communication::initialize(config, |mut allocator| {

        println!("worker {} of {} started", allocator.index(), allocator.peers());

        // allocates a pair of senders list and one receiver.
        let (mut senders, mut receiver) = allocator.allocate(0);

        // send typed data along each channel
        for i in 0 .. allocator.peers() {
            senders[i].send(Message::from_typed(format!("hello, {}", i)));
            senders[i].done();
        }

        // no support for termination notification,
        // we have to count down ourselves.
        let mut received = 0;
        while received < allocator.peers() {

            allocator.receive();

            if let Some(message) = receiver.recv() {
                println!("worker {}: received: <{}>", allocator.index(), message.deref());
                received += 1;
            }

            allocator.release();
        }

        allocator.index()
    });

    // computation runs until guards are joined or dropped.
    if let Ok(guards) = guards {
        for guard in guards.join() {
            println!("result: {:?}", guard);
        }
    }
    else { println!("error in computation"); }
}

pub fn release(&mut self)

Perform work after scheduling operators.

Examples found in repository

examples/comm_hello.rs (line 35)

fn main() {

    // extract the configuration from user-supplied arguments, initialize the computation.
    let config = timely_communication::Config::from_args(std::env::args()).unwrap();
    let guards = timely_communication::initialize(config, |mut allocator| {

        println!("worker {} of {} started", allocator.index(), allocator.peers());

        // allocates a pair of senders list and one receiver.
        let (mut senders, mut receiver) = allocator.allocate(0);

        // send typed data along each channel
        for i in 0 .. allocator.peers() {
            senders[i].send(Message::from_typed(format!("hello, {}", i)));
            senders[i].done();
        }

        // no support for termination notification,
        // we have to count down ourselves.
        let mut received = 0;
        while received < allocator.peers() {

            allocator.receive();

            if let Some(message) = receiver.recv() {
                println!("worker {}: received: <{}>", allocator.index(), message.deref());
                received += 1;
            }

            allocator.release();
        }

        allocator.index()
    });

    // computation runs until guards are joined or dropped.
    if let Ok(guards) = guards {
        for guard in guards.join() {
            println!("result: {:?}", guard);
        }
    }
    else { println!("error in computation"); }
}

Trait Implementations

impl Allocate for Generic

fn index(&self) -> usize

The index of the worker out of (0..self.peers()).

fn peers(&self) -> usize

The number of workers in the communication group.

fn allocate<T: Data>( &mut self, identifier: usize, ) -> (Vec<Box<dyn Push<Message<T>>>>, Box<dyn Pull<Message<T>>>)

Constructs several send endpoints and one receive endpoint.

fn receive(&mut self)

Ensure that received messages are surfaced in each channel.

fn release(&mut self)

Signal the completion of a batch of reads from channels.

fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>>

A shared queue of communication events with channel identifier.

fn await_events(&self, _duration: Option<Duration>)

Awaits communication events.

fn pipeline<T: 'static>( &mut self, identifier: usize, ) -> (ThreadPusher<Message<T>>, ThreadPuller<Message<T>>)

Constructs a pipeline channel from the worker to itself.