pub fn new(worker: &Worker<Generic>, load_balance_factor: Option<usize>, max_fuel: Option<usize>, path: String) -> Self {
        let load_balance_factor = if let Some(load_balance_factor) = load_balance_factor {
            load_balance_factor
        } else {
            1
        };

        let max_fuel = if let Some(max_fuel) = max_fuel {
            max_fuel
        } else {
            4096
        };

        let writers = if let Ok(addr) = ::std::env::var("SNAILTRAIL_ADDR") {
            info!("w{} registers logger @{:?}: lbf{}, fuel{}", worker.index(), &addr, load_balance_factor, max_fuel);
            (0 .. load_balance_factor)
                .map(|_| TcpStream::connect(&addr).expect("could not connect to logging stream"))
                .map(|stream| {
                    // SnailTrail should be able to keep up with an online computation.
                    // If batch sizes are too large, they should be buffered. Blocking the
                    // TCP connection is not an option as it slows down the main computation.
                    // stream
                    //    .set_nonblocking(true)
                    //    .expect("set_nonblocking call failed");

                    EventWriter::<Pair<u64, Duration>, _, _>::new(TcpStreamOrFile::Tcp(stream))
                })
                .collect::<Vec<_>>()
        } else {
            (0 .. load_balance_factor).map(|i| {
                let name = format!("{}/{}.dump", path, (worker.index() + i * worker.peers()));
                info!("creating {}", name);
                let path = Path::new(&name);
                let file = match File::create(&path) {
                    Err(why) => panic!("couldn't create {}: {}", path.display(), why.description()),
                    Ok(file) => file,
                };
                EventWriter::<Pair<u64, Duration>, _, _>::new(TcpStreamOrFile::File(file))
            }).collect::<Vec<_>>()
        };

        PAGLogger {
            writers,
            curr_writer: 0,
            curr_cap: Default::default(),
            next_cap: Pair::new(1, Default::default()),
            seq_no: 0,
            buffer: Vec::new(),
            tick_sys: true,
            max_fuel,
            fuel: max_fuel,
            op_addr_to_ch_target: HashMap::new(),
            op_addr_to_ch_source: HashMap::new(),
            op_id_to_op_addr: HashMap::new(),
            channel_records: HashMap::new(),
            worker_index: worker.index(),
            overall_messages: 0,
            pag_messages: 0,
            elapsed: std::time::Instant::now(),
            epoch_count: 0,
        }
    }
/// Creates a new PAGLogger. Events are logged to TCP or file.
/// Commonly called indirectly from `create_and_attach`
Creates a new PAGLogger. Events are logged to TCP or file.
Commonly called indirectly from `create_and_attach`
Creates a new PAGLogger . Events are logged to TCP or file . Commonly called indirectly from ` create_and_attach `
pub fn new(worker: &Worker<Generic>, load_balance_factor: Option<usize>, max_fuel: Option<usize>, path: String) -> Self {
        let load_balance_factor = if let Some(load_balance_factor) = load_balance_factor {
            load_balance_factor
        } else {
            1
        };
        let max_fuel = if let Some(max_fuel) = max_fuel {
            max_fuel
        } else {
            4096
        };
        let writers = if let Ok(addr) = ::std::env::var("SNAILTRAIL_ADDR") {
            info!("w{} registers logger @{:?}: lbf{}, fuel{}", worker.index(), &addr, load_balance_factor, max_fuel);
            (0 .. load_balance_factor)
                .map(|_| TcpStream::connect(&addr).expect("could not connect to logging stream"))
                .map(|stream| {
                    // SnailTrail should be able to keep up with an online computation.
                    // If batch sizes are too large, they should be buffered. Blocking the
                    // TCP connection is not an option as it slows down the main computation.
                    // stream
                    //    .set_nonblocking(true)
                    //    .expect("set_nonblocking call failed");
                    EventWriter::<Pair<u64, Duration>, _, _>::new(TcpStreamOrFile::Tcp(stream))
                })
                .collect::<Vec<_>>()
        } else {
            (0 .. load_balance_factor).map(|i| {
                let name = format!("{}/{}.dump", path, (worker.index() + i * worker.peers()));
                info!("creating {}", name);
                let path = Path::new(&name);
                let file = match File::create(&path) {
                    Err(why) => panic!("couldn't create {}: {}", path.display(), why.description()),
                    Ok(file) => file,
                };
                EventWriter::<Pair<u64, Duration>, _, _>::new(TcpStreamOrFile::File(file))
            }).collect::<Vec<_>>()
        };
        PAGLogger {
            writers,
            curr_writer: 0,
            curr_cap: Default::default(),
            next_cap: Pair::new(1, Default::default()),
            seq_no: 0,
            buffer: Vec::new(),
            tick_sys: true,
            max_fuel,
            fuel: max_fuel,
            op_addr_to_ch_target: HashMap::new(),
            op_addr_to_ch_source: HashMap::new(),
            op_id_to_op_addr: HashMap::new(),
            channel_records: HashMap::new(),
            worker_index: worker.index(),
            overall_messages: 0,
            pag_messages: 0,
            elapsed: std::time::Instant::now(),
            epoch_count: 0,
        }
    }
pub fn new ( worker : & Worker < Generic > , load_balance_factor : Option < usize > , max_fuel : Option < usize > , path : String ) -> Self { let load_balance_factor = if let Some ( load_balance_factor ) = load_balance_factor { load_balance_factor } else { 1 } ; let max_fuel = if let Some ( max_fuel ) = max_fuel { max_fuel } else { 4096 } ; let writers = if let Ok ( addr ) = :: std :: env :: var ( " " ) { info ! ( " " worker index ( ) addr load_balance_factor max_fuel ) ; ( 0 .. load_balance_factor ) . map ( | _ | TcpStream :: connect ( & addr ) . expect ( " " ) ) . map ( | stream | { // SnailTrail should be able to keep up with an online computation. // If batch sizes are too large, they should be buffered. Blocking the // TCP connection is not an option as it slows down the main computation. // stream //    .set_nonblocking(true) //    .expect("set_nonblocking call failed"); EventWriter :: < Pair < u64 , Duration > , _ , _ > :: new ( TcpStreamOrFile :: Tcp ( stream ) ) } ) . collect :: < Vec < _ > > ( ) } else { ( 0 .. load_balance_factor ) . map ( | i | { let name = format ! ( " " path ( worker index ( ) i worker peers ( ) ) ) ; info ! ( " " name ) ; let path = Path :: new ( & name ) ; let file = match File :: create ( & path ) { Err ( why ) => panic ! ( " " path display ( ) why description ( ) ) , Ok ( file ) => file , } ; EventWriter :: < Pair < u64 , Duration > , _ , _ > :: new ( TcpStreamOrFile :: File ( file ) ) } ) . collect :: < Vec < _ > > ( ) } ; PAGLogger { writers , curr_writer : 0 , curr_cap : Default :: default ( ) , next_cap : Pair :: new ( 1 , Default :: default ( ) ) , seq_no : 0 , buffer : Vec :: new ( ) , tick_sys : true , max_fuel , fuel : max_fuel , op_addr_to_ch_target : HashMap :: new ( ) , op_addr_to_ch_source : HashMap :: new ( ) , op_id_to_op_addr : HashMap :: new ( ) , channel_records : HashMap :: new ( ) , worker_index : worker . index ( ) , overall_messages : 0 , pag_messages : 0 , elapsed : std :: time :: Instant :: now ( ) , epoch_count : 0 , } }
Creates a new PAGLogger.
Creates a new PAGLogger .
worker
&Worker<Generic>
load_balance_factor
Option<usize>
max_fuel
Option<usize>
path
String
worker
&Worker<Generic>
load_balance_factor
Option<usize>
max_fuel
Option<usize>
path
String