Trait Map 

pub trait Map<S: Scope, D: 'static>: Sized {
    // Required methods
    fn map_in_place<L: FnMut(&mut D) + 'static>(
        self,
        logic: L,
    ) -> StreamVec<S, D>;
    fn flat_map<I: IntoIterator, L: FnMut(D) -> I + 'static>(
        self,
        logic: L,
    ) -> StreamVec<S, I::Item>
       where I::Item: 'static;

    // Provided method
    fn map<D2: 'static, L: FnMut(D) -> D2 + 'static>(
        self,
        logic: L,
    ) -> StreamVec<S, D2> { ... }
}

Extension trait for StreamVec.

Required Methods

fn map_in_place<L: FnMut(&mut D) + 'static>(self, logic: L) -> StreamVec<S, D>

Updates each element of the stream and yields the element, re-using memory where possible.

Examples

use timely::dataflow::operators::{ToStream, Inspect, vec::Map};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .map_in_place(|x| *x += 1)
           .inspect(|x| println!("seen: {:?}", x));
});

fn flat_map<I: IntoIterator, L: FnMut(D) -> I + 'static>( self, logic: L, ) -> StreamVec<S, I::Item>

where I::Item: 'static,

Consumes each element of the stream and yields some number of new elements.

Examples

use timely::dataflow::operators::{ToStream, Inspect, vec::Map};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .flat_map(|x| (0..x))
           .inspect(|x| println!("seen: {:?}", x));
});

Provided Methods

fn map<D2: 'static, L: FnMut(D) -> D2 + 'static>( self, logic: L, ) -> StreamVec<S, D2>

Consumes each element of the stream and yields a new element.

Examples

use timely::dataflow::operators::{ToStream, Inspect, vec::Map};

timely::example(|scope| {
    (0..10).to_stream(scope)
           .map(|x| x + 1)
           .inspect(|x| println!("seen: {:?}", x));
});

Examples found in repository

examples/event_driven.rs (line 28)

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let timer = std::time::Instant::now();

        let mut args = std::env::args();
        args.next();

        let dataflows = args.next().unwrap().parse::<usize>().unwrap();
        let length = args.next().unwrap().parse::<usize>().unwrap();
        let record = args.next() == Some("record".to_string());

        let mut inputs = Vec::new();
        let mut probes = Vec::new();

        // create a new input, exchange data, and inspect its output
        for _dataflow in 0 .. dataflows {
            worker.dataflow(|scope| {
                let (input, stream) = scope.new_input();
                let stream = scope.region(|inner| {
                    let mut stream = stream.enter(inner);
                    for _step in 0 .. length {
                        stream = stream.map(|x: ()| x);
                    }
                    stream.leave()
                });
                let (probe, _stream) = stream.probe();
                inputs.push(input);
                probes.push(probe);
            });
        }

        println!("{:?}\tdataflows built ({} x {})", timer.elapsed(), dataflows, length);

        for round in 0 .. {
            let dataflow = round % dataflows;
            if record {
                inputs[dataflow].send(());
            }
            inputs[dataflow].advance_to(round);
            let mut steps = 0;
            while probes[dataflow].less_than(&round) {
                worker.step();
                steps += 1;
            }
            if round % 1000 == 0 { println!("{:?}\tround {} complete in {} steps", timer.elapsed(), round, steps); }
        }

    }).unwrap();
}

examples/loopdemo.rs (line 32)

fn main() {

    let mut args = std::env::args();
    args.next();
    let rate: usize = args.next().expect("Must specify rate").parse().expect("Rate must be an usize");
    let duration_s: usize = args.next().expect("Must specify duration_s").parse().expect("duration_s must be an usize");

    timely::execute_from_args(args, move |worker| {

        let index = worker.index();
        let peers = worker.peers();

        let timer = std::time::Instant::now();

        let mut input = InputHandle::new();
        let probe = ProbeHandle::new();

        // Create a dataflow that discards input data (just synchronizes).
        worker.dataflow(|scope| {

            let stream = scope.input_from(&mut input);

            let (loop_handle, loop_stream) = scope.feedback(1);

            let step =
            stream
                .concat(loop_stream)
                .map(|x| if x % 2 == 0 { x / 2 } else { 3 * x + 1 })
                .filter(|x| x > &1);

            step.probe_with(&probe)
                .connect_loop(loop_handle);
        });

        let ns_per_request = 1_000_000_000 / rate;
        let mut insert_counter = index;           // counts up as we insert records.
        let mut retire_counter = index;           // counts up as we retire records.

        let mut inserted_ns = 0;

        // We repeatedly consult the elapsed time, and introduce any data now considered available.
        // At the same time, we observe the output and record which inputs are considered retired.

        let mut counts = vec![[0u64; 16]; 64];

        let counter_limit = rate * duration_s;
        while retire_counter < counter_limit {

            // Open-loop latency-throughput test, parameterized by offered rate `ns_per_request`.
            let elapsed = timer.elapsed();
            let elapsed_ns = elapsed.as_secs() * 1_000_000_000 + (elapsed.subsec_nanos() as u64);

            // Determine completed ns.
            let acknowledged_ns: u64 = probe.with_frontier(|frontier| frontier[0]);

            // Notice any newly-retired records.
            while ((retire_counter * ns_per_request) as u64) < acknowledged_ns && retire_counter < counter_limit {
                let requested_at = (retire_counter * ns_per_request) as u64;
                let latency_ns = elapsed_ns - requested_at;

                let count_index = latency_ns.next_power_of_two().trailing_zeros() as usize;
                let low_bits = ((elapsed_ns - requested_at) >> (count_index - 5)) & 0xF;
                counts[count_index][low_bits as usize] += 1;

                retire_counter += peers;
            }

            // Now, should we introduce more records before stepping the worker?
            // Three choices here:
            //
            //   1. Wait until previous batch acknowledged.
            //   2. Tick at most once every millisecond-ish.
            //   3. Geometrically increasing outstanding batches.

            // Technique 1:
            // let target_ns = if acknowledged_ns >= inserted_ns { elapsed_ns } else { inserted_ns };

            // Technique 2:
            // let target_ns = elapsed_ns & !((1 << 20) - 1);

            // Technique 3:
            let scale = (inserted_ns - acknowledged_ns).next_power_of_two();
            let target_ns = elapsed_ns & !(scale - 1);

            if inserted_ns < target_ns {

                while ((insert_counter * ns_per_request) as u64) < target_ns {
                    input.send(insert_counter);
                    insert_counter += peers;
                }
                input.advance_to(target_ns);
                inserted_ns = target_ns;
            }

            worker.step();
        }

        // Report observed latency measurements.
        if index == 0 {

            let mut results = Vec::new();
            let total = counts.iter().map(|x| x.iter().sum::<u64>()).sum();
            let mut sum = 0;
            for index in (10 .. counts.len()).rev() {
                for sub in (0 .. 16).rev() {
                    if sum > 0 && sum < total {
                        let latency = (1 << (index-1)) + (sub << (index-5));
                        let fraction = (sum as f64) / (total as f64);
                        results.push((latency, fraction));
                    }
                    sum += counts[index][sub];
                }
            }
            for (latency, fraction) in results.drain(..).rev() {
                println!("{}\t{}", latency, fraction);
            }
        }

    }).unwrap();
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl<S: Scope, D: 'static> Map<S, D> for StreamVec<S, D>