store and reuse logging streams
@@ -57,19 +57,35 @@ impl LoggerConfig {
/// TIMELY_WORKER_LOG_ADDR or TIMELY_COMM_LOG_ADDR are environment
/// variables set to the log destinations.
pub fn default_with_env() -> LoggerConfig {
- use std::sync::Arc;
+
+ use std::sync::{Arc, Mutex};
use std::rc::Rc;
use std::cell::RefCell;
+ use std::net::TcpStream;
+ use std::collections::HashMap;
+
+ let timely_stream = Mutex::new(HashMap::<usize, TcpStream>::new());
+ let comm_stream = Mutex::new(HashMap::<::timely_communication::logging::CommsSetup, TcpStream>::new());
::logging::LoggerConfig {
timely_logging: match ::std::env::var("TIMELY_WORKER_LOG_ADDR") {
Ok(addr) => {
eprintln!("enabled WORKER logging to {}", addr);
Arc::new(move |events_setup: ::logging::TimelySetup| {
+
+ let send =
+ timely_stream
+ .lock()
+ .expect("unable to lock timely logging streams")
+ .entry(events_setup.index)
+ .or_insert_with(|| {
let addr = addr.clone();
+ TcpStream::connect(addr).unwrap()
+ })
+ .try_clone()
+ .expect("unable to clone logging stream");
+
let logger = RefCell::new(::logging::BatchLogger::new((move |_setup| {
- use std::net::TcpStream;
- let send = TcpStream::connect(addr).unwrap();
::dataflow::operators::capture::EventWriter::new(send)
})(events_setup)));
Rc::new(::timely_communication::logging::BufferingLogger::new(
@@ -82,10 +98,23 @@ impl LoggerConfig {
Ok(addr) => {
eprintln!("enabled COMM logging to {}", addr);
Arc::new(move |events_setup: ::timely_communication::logging::CommsSetup| {
+
+ let send =
+ comm_stream
+ .lock()
+ .expect("unable to lock comms logging streams")
+ .entry(events_setup)
+ .or_insert_with(|| {
let addr = addr.clone();
+ TcpStream::connect(addr).unwrap()
+ })
+ .try_clone()
+ .expect("unable to clone logging stream");
+
+ // let addr = addr.clone();
let logger = RefCell::new(::logging::BatchLogger::new((move |_setup| {
- use std::net::TcpStream;
- let send = TcpStream::connect(addr).unwrap();
+ // use std::net::TcpStream;
+ // let send = TcpStream::connect(addr).unwrap();
::dataflow::operators::capture::EventWriter::new(send)
})(events_setup)));
Rc::new(::timely_communication::logging::BufferingLogger::new(
