{"action":"closed","issue":{"url":"https://api.github.com/repos/actix/actix/issues/94","repository_url":"https://api.github.com/repos/actix/actix","labels_url":"https://api.github.com/repos/actix/actix/issues/94/labels{/name}","comments_url":"https://api.github.com/repos/actix/actix/issues/94/comments","events_url":"https://api.github.com/repos/actix/actix/issues/94/events","html_url":"https://github.com/actix/actix/issues/94","id":332460658,"node_id":"MDU6SXNzdWUzMzI0NjA2NTg=","number":94,"title":"Use Self::Context::notify() instead of direct use of address","user":{"login":"daschl","id":29086,"node_id":"MDQ6VXNlcjI5MDg2","avatar_url":"https://avatars2.githubusercontent.com/u/29086?v=4","gravatar_id":"","url":"https://api.github.com/users/daschl","html_url":"https://github.com/daschl","followers_url":"https://api.github.com/users/daschl/followers","following_url":"https://api.github.com/users/daschl/following{/other_user}","gists_url":"https://api.github.com/users/daschl/gists{/gist_id}","starred_url":"https://api.github.com/users/daschl/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/daschl/subscriptions","organizations_url":"https://api.github.com/users/daschl/orgs","repos_url":"https://api.github.com/users/daschl/repos","events_url":"https://api.github.com/users/daschl/events{/privacy}","received_events_url":"https://api.github.com/users/daschl/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":2,"created_at":"2018-06-14T15:47:44Z","updated_at":"2018-06-15T07:33:53Z","closed_at":"2018-06-15T07:33:52Z","author_association":"NONE","body":"Hi,\r\n\r\nI'm new to actix so I don't know if this is a user error or not.\r\n\r\nI have an actor system right now of 3 actors, one which does a blocking loop on a file reading it line by line and then two others that need to do processing in the future but for now are just forwarding.\r\n\r\nBasic actor system setup:\r\n\r\n```rust\r\n    let system = System::new(\"oida\");\r\n\r\n    // Initialize the Parser\r\n    let parser: Addr<Unsync, _> = Parser::new(\"^%{TIMESTAMP_ISO8601:timestamp} %{NUMBER:ign} \\\\| %{LOGLEVEL:level}%{SPACE} \\\\| %{GREEDYDATA:message}$\").start();\r\n\r\n    // Initialize the Line Aggregator Actor and start it in the current event loop\r\n    let aggregator: Addr<Syn, _> = LineAggregator::new(parser.recipient()).start();\r\n    let agg_recipient: Recipient<Syn, RawLineMessage> = aggregator.recipient();\r\n\r\n    // Initialize the File Reader actor and send it to another thread since it\r\n    // is using blocking api.\r\n    SyncArbiter::start(1, move || {\r\n        FileReader::new(agg_recipient.clone(), \"tmp/input.log\")\r\n    });\r\n\r\n    system.run();\r\n```\r\n\r\nReader:\r\n\r\n```rust\r\npub struct FileReader {\r\n    recipient: Recipient<Syn, RawLineMessage>,\r\n    path: String,\r\n}\r\n\r\nimpl FileReader {\r\n    pub fn new(recipient: Recipient<Syn, RawLineMessage>, path: &str) -> Self {\r\n        FileReader {\r\n            recipient,\r\n            path: path.into(),\r\n        }\r\n    }\r\n}\r\n\r\nimpl Actor for FileReader {\r\n    type Context = SyncContext<Self>;\r\n\r\n    fn started(&mut self, _: &mut Self::Context) {\r\n        let file = File::open(&self.path).unwrap();\r\n        let reader = BufReader::new(file);\r\n        for line in reader.lines() {\r\n            match self.recipient.do_send(RawLineMessage(line.unwrap())) {\r\n                Ok(_) => (),\r\n                Err(e) => println!(\"!!!! could not send log line {:?}\", e),\r\n            }\r\n        }\r\n    }\r\n}\r\n```\r\n\r\nAggregator:\r\n\r\n```rust\r\npub struct LineAggregator {\r\n    recipient: Recipient<Unsync, RawLineMessage>,\r\n}\r\n\r\nimpl LineAggregator {\r\n    pub fn new(recipient: Recipient<Unsync, RawLineMessage>) -> Self {\r\n        LineAggregator { recipient }\r\n    }\r\n}\r\n\r\nimpl Actor for LineAggregator {\r\n    type Context = Context<Self>;\r\n}\r\n\r\nimpl Handler<RawLineMessage> for LineAggregator {\r\n    type Result = ();\r\n\r\n    fn handle(&mut self, msg: RawLineMessage, _: &mut Self::Context) {\r\n        match self.recipient.do_send(msg) {\r\n            Ok(_) => (),\r\n            Err(e) => println!(\"!!!! could not send log line {:?}\", e),\r\n        }\r\n    }\r\n}\r\n```\r\n\r\nFinally the parser:\r\n\r\n```rust\r\npub struct Parser {\r\n    pattern: Pattern,\r\n}\r\n\r\nimpl Parser {\r\n    pub fn new(pattern: &str) -> Self {\r\n        let mut grok = Grok::default();\r\n        let compiled_pattern = grok.compile(pattern, false).unwrap();\r\n        Parser {\r\n            pattern: compiled_pattern,\r\n        }\r\n    }\r\n}\r\n\r\nimpl Actor for Parser {\r\n    type Context = Context<Self>;\r\n}\r\n\r\nimpl Handler<RawLineMessage> for Parser {\r\n    type Result = ();\r\n\r\n    fn handle(&mut self, msg: RawLineMessage, _: &mut Self::Context) {\r\n        println!(\"parser received: {:?}\", msg);\r\n    }\r\n}\r\n```\r\n\r\nWhen I run this against a larger file with try_send I think it panics at the 16 item mailbox size (I know its configurable but why does it overflow?.. if I use the do_send as in the code here it fails with:\r\n\r\n```\r\nthread 'main' panicked at 'Use Self::Context::notify() instead of direct use of address',\r\n```\r\n\r\n```\r\nthread 'main' panicked at 'Use Self::Context::notify() instead of direct use of address', /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/actix-0.5.8/src/mailbox.rs:134:21\r\nstack backtrace:\r\n   0: std::sys::unix::backtrace::tracing::imp::unwind_backtrace\r\n             at libstd/sys/unix/backtrace/tracing/gcc_s.rs:49\r\n   1: std::sys_common::backtrace::print\r\n             at libstd/sys_common/backtrace.rs:71\r\n             at libstd/sys_common/backtrace.rs:59\r\n   2: std::panicking::default_hook::{{closure}}\r\n             at libstd/panicking.rs:207\r\n   3: std::panicking::default_hook\r\n             at libstd/panicking.rs:223\r\n   4: std::panicking::begin_panic\r\n             at libstd/panicking.rs:402\r\n   5: std::panicking::begin_panic\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/panicking.rs:365\r\n   6: <actix::mailbox::Mailbox<A>>::poll\r\n             at ./<panic macros>:3\r\n   7: <actix::contextimpl::ContextImpl<A>>::poll\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/actix-0.5.8/src/contextimpl.rs:291\r\n   8: <actix::context::Context<A> as futures::future::Future>::poll\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/actix-0.5.8/src/context.rs:144\r\n   9: <futures::future::map::Map<A, F> as futures::future::Future>::poll\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/future/map.rs:30\r\n  10: <futures::future::map_err::MapErr<A, F> as futures::future::Future>::poll\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/future/map_err.rs:30\r\n  11: <alloc::boxed::Box<F> as futures::future::Future>::poll\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/future/mod.rs:113\r\n  12: <futures::task_impl::Spawn<T>>::poll_future_notify::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/task_impl/mod.rs:289\r\n  13: <futures::task_impl::Spawn<T>>::enter::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/task_impl/mod.rs:363\r\n  14: futures::task_impl::std::set\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/task_impl/std/mod.rs:78\r\n  15: <futures::task_impl::Spawn<T>>::enter\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/task_impl/mod.rs:363\r\n  16: <futures::task_impl::Spawn<T>>::poll_future_notify\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.21/src/task_impl/mod.rs:289\r\n  17: <tokio::executor::current_thread::scheduler::Scheduled<'a, U>>::tick\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/scheduler.rs:352\r\n  18: <tokio::executor::current_thread::scheduler::Scheduler<U>>::tick::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/scheduler.rs:332\r\n  19: <tokio::executor::current_thread::Borrow<'a, U>>::enter::{{closure}}::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:762\r\n  20: tokio::executor::current_thread::CurrentRunner::set_spawn\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:794\r\n  21: <tokio::executor::current_thread::Borrow<'a, U>>::enter::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:761\r\n  22: <std::thread::local::LocalKey<T>>::try_with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:290\r\n  23: <std::thread::local::LocalKey<T>>::with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:244\r\n  24: <tokio::executor::current_thread::Borrow<'a, U>>::enter\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:760\r\n  25: <tokio::executor::current_thread::scheduler::Scheduler<U>>::tick\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/scheduler.rs:332\r\n  26: <tokio::executor::current_thread::Entered<'a, P>>::tick\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:611\r\n  27: <tokio::executor::current_thread::Entered<'a, P>>::turn\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.7/src/executor/current_thread/mod.rs:537\r\n  28: tokio_core::reactor::Core::poll::{{closure}}::{{closure}}::{{closure}}::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:298\r\n  29: <scoped_tls::ScopedKey<T>>::set\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/scoped-tls-0.1.2/src/lib.rs:155\r\n  30: tokio_core::reactor::Core::poll::{{closure}}::{{closure}}::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:297\r\n  31: tokio_timer::timer::handle::with_default::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-timer-0.2.4/src/timer/handle.rs:64\r\n  32: <std::thread::local::LocalKey<T>>::try_with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:290\r\n  33: <std::thread::local::LocalKey<T>>::with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:244\r\n  34: tokio_timer::timer::handle::with_default\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-timer-0.2.4/src/timer/handle.rs:56\r\n  35: tokio_core::reactor::Core::poll::{{closure}}::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:275\r\n  36: tokio_executor::global::with_default::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-executor-0.1.2/src/global.rs:176\r\n  37: <std::thread::local::LocalKey<T>>::try_with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:290\r\n  38: <std::thread::local::LocalKey<T>>::with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:244\r\n  39: tokio_executor::global::with_default\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-executor-0.1.2/src/global.rs:150\r\n  40: tokio_core::reactor::Core::poll::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:274\r\n  41: tokio_reactor::with_default::{{closure}}\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-reactor-0.1.2/src/lib.rs:231\r\n  42: <std::thread::local::LocalKey<T>>::try_with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:290\r\n  43: <std::thread::local::LocalKey<T>>::with\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/thread/local.rs:244\r\n  44: tokio_reactor::with_default\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-reactor-0.1.2/src/lib.rs:214\r\n  45: <tokio_timer::timer::Timer<T, N> as tokio_executor::park::Park>::unpark\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:273\r\n  46: tokio_core::reactor::Core::run\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.17/src/reactor/mod.rs:248\r\n  47: <std::sync::mpsc::cache_aligned::CacheAligned<T> as core::ops::deref::Deref>::deref\r\n             at /Users/daschl/.cargo/registry/src/github.com-1ecc6299db9ec823/actix-0.5.8/src/system.rs:106\r\n  48: oida::main\r\n             at src/main.rs:37\r\n  49: std::rt::lang_start::{{closure}}\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/rt.rs:74\r\n  50: std::panicking::try::do_call\r\n             at libstd/rt.rs:59\r\n             at libstd/panicking.rs:306\r\n  51: panic_unwind::dwarf::eh::read_encoded_pointer\r\n             at libpanic_unwind/lib.rs:102\r\n  52: <std::io::Write::write_fmt::Adaptor<'a, T> as core::fmt::Write>::write_str\r\n             at libstd/panicking.rs:285\r\n             at libstd/panic.rs:361\r\n             at libstd/rt.rs:58\r\n  53: std::rt::lang_start\r\n             at /Users/travis/build/rust-lang/rust/src/libstd/rt.rs:74\r\n  54: oida::main\r\n```\r\n\r\nAny idea what caused this and/or how do I maintain proper mailbox hygiene since the file reader is I assume decoupled from the rest of the actors which run in one event loop?\r\n\r\nThanks,\r\nMichael"}}
{"actor":{"display_login":"daschl"}}