{"action":"created","issue":{"url":"https://api.github.com/repos/carllerche/mio/issues/360","repository_url":"https://api.github.com/repos/carllerche/mio","labels_url":"https://api.github.com/repos/carllerche/mio/issues/360/labels{/name}","comments_url":"https://api.github.com/repos/carllerche/mio/issues/360/comments","events_url":"https://api.github.com/repos/carllerche/mio/issues/360/events","html_url":"https://github.com/carllerche/mio/issues/360","id":135880602,"number":360,"title":"Proposal: Unify Sockets, Timers, and Channels","user":{"login":"carllerche","id":6180,"avatar_url":"https://avatars.githubusercontent.com/u/6180?v=3","gravatar_id":"","url":"https://api.github.com/users/carllerche","html_url":"https://github.com/carllerche","followers_url":"https://api.github.com/users/carllerche/followers","following_url":"https://api.github.com/users/carllerche/following{/other_user}","gists_url":"https://api.github.com/users/carllerche/gists{/gist_id}","starred_url":"https://api.github.com/users/carllerche/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/carllerche/subscriptions","organizations_url":"https://api.github.com/users/carllerche/orgs","repos_url":"https://api.github.com/users/carllerche/repos","events_url":"https://api.github.com/users/carllerche/events{/privacy}","received_events_url":"https://api.github.com/users/carllerche/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/carllerche/mio/labels/api","name":"api","color":"c7def8"},{"url":"https://api.github.com/repos/carllerche/mio/labels/behavior","name":"behavior","color":"bfdadc"},{"url":"https://api.github.com/repos/carllerche/mio/labels/windows","name":"windows","color":"006b75"}],"state":"open","locked":false,"assignee":{"login":"carllerche","id":6180,"avatar_url":"https://avatars.githubusercontent.com/u/6180?v=3","gravatar_id":"","url":"https://api.github.com/users/carllerche","html_url":"https://github.com/carllerche","followers_url":"https://api.github.com/users/carllerche/followers","following_url":"https://api.github.com/users/carllerche/following{/other_user}","gists_url":"https://api.github.com/users/carllerche/gists{/gist_id}","starred_url":"https://api.github.com/users/carllerche/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/carllerche/subscriptions","organizations_url":"https://api.github.com/users/carllerche/orgs","repos_url":"https://api.github.com/users/carllerche/repos","events_url":"https://api.github.com/users/carllerche/events{/privacy}","received_events_url":"https://api.github.com/users/carllerche/received_events","type":"User","site_admin":false},"milestone":{"url":"https://api.github.com/repos/carllerche/mio/milestones/4","html_url":"https://github.com/carllerche/mio/milestones/v1.0","labels_url":"https://api.github.com/repos/carllerche/mio/milestones/4/labels","id":1425739,"number":4,"title":"v1.0","description":"API stability","creator":{"login":"carllerche","id":6180,"avatar_url":"https://avatars.githubusercontent.com/u/6180?v=3","gravatar_id":"","url":"https://api.github.com/users/carllerche","html_url":"https://github.com/carllerche","followers_url":"https://api.github.com/users/carllerche/followers","following_url":"https://api.github.com/users/carllerche/following{/other_user}","gists_url":"https://api.github.com/users/carllerche/gists{/gist_id}","starred_url":"https://api.github.com/users/carllerche/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/carllerche/subscriptions","organizations_url":"https://api.github.com/users/carllerche/orgs","repos_url":"https://api.github.com/users/carllerche/repos","events_url":"https://api.github.com/users/carllerche/events{/privacy}","received_events_url":"https://api.github.com/users/carllerche/received_events","type":"User","site_admin":false},"open_issues":19,"closed_issues":2,"state":"open","created_at":"2015-11-23T18:25:48Z","updated_at":"2016-02-26T21:58:44Z","due_on":"2016-02-01T05:00:00Z","closed_at":null},"comments":21,"created_at":"2016-02-23T22:02:34Z","updated_at":"2016-02-27T01:08:30Z","closed_at":null,"body":"# Unify Sockets, Timers, and Channels\r\n\r\nCurrently, there are two runtime APIs: `Poll` and `EventLoop`. `Poll` is\r\nthe abstraction handling readiness for sockets and waiting for events on\r\nthese sockets. `EventLoop` is a wrapper around `Poll` providing a\r\ntimeout API and a cross thread notification API as well as the loop /\r\nreactive (via `Handler`) abstraction.\r\n\r\nI am proposing to extract the timer and the notification channel\r\nfeatures into standalone types that implement `Evented` and thus can be\r\nused directly with `Poll`. For example:\r\n\r\n```rust\r\nlet timer = mio::Timer::new();\r\nlet poll = mio::Poll::new();\r\n\r\npoll.register(&timer, Token(0), EventSet::readable(), PollOpt::edge());\r\n\r\ntimer.timeout(\"hello\", Duration::from_millis(1_000));\r\n\r\npoll.poll();\r\n```\r\n\r\nThe insight is that a lot of the code that is currently windows specific\r\nis useful to all platforms. Stabilizing the API and providing it to all\r\nplatforms allows implementing `Evented` for arbitrary types.\r\n\r\n## Advantages\r\n\r\n- Mio would have a unified API.\r\n- No change in overhead for any platform.\r\n- Move (and improve) some of the currently windows-only code to all platforms.\r\n- The notification channel backend could be configurable:\r\n\r\n```rust\r\nlet (tx, rx) = mio::Channel::new(mpsc::channel());\r\npoll.register(&rx, Token(0), EventSet::readable(), PollOpt::edge());\r\n```\r\n\r\n## Disadvantages\r\n\r\n* Unsafe code\r\n* More code (lock-free algorithms)\r\n\r\nThe primary disadvantage that I can think of is that the code path\r\naround timers & the notification channel become slightly more\r\ncomplicated. I don't believe that the change would have a meaningful\r\nperformance impact.\r\n\r\nThere is also additional code complexity for all platforms. However,\r\nthis code complexity already exists for Windows.\r\n\r\n## Behavior\r\n\r\nAn `Evented` would mirror the behavior of a socket registered with\r\nepoll. Specifically, in a single threaded environment:\r\n\r\n* A value registered will trigger at most one notification per call to `poll`.\r\n* A value registered with readable interest & edge triggers a notification once when it becomes readable.\r\n* A value registered with readable interest & level triggers a notification every call to poll as long as the value is still readable.\r\n* A value registered (or reregistered) with readable interest triggers a notification immediately if it is currently readable.\r\n* If a value is registered with readable interest only and already has a pending writable notification, the event is discarded\r\n* If a value has any pending notifications and is deregistered, the pending notifications are cleared.\r\n* When a value is dropped, it will no longer trigger any further notifications.\r\n* `Poll` is permitted to fire of spurious readiness events *except* if the value has been dropped.\r\n\r\nIn the presence of concurrency, specifically readiness being modified on\r\na different thread than `Poll`, a best effort is made to preserve these\r\nsemantics.\r\n\r\n## Implementation\r\n\r\nThis section will describe how to implement a custom `Evented` type as\r\nwell as Mio's internals to handle it. For simplicity and performance,\r\ncustom `Evented` types will only be able to be registered with a single\r\n`Poll`.\r\n\r\nIt is important to note that the implementation is **not** intended to\r\nreplace FD polling on epoll & kqueue. It is meant to work in conjunction\r\nwith the OS's event queue to support types that cannot be implemented\r\nusing a socket or other system type that is compatible with the system's\r\nevent queue.\r\n\r\n### Readiness Queue\r\n\r\n`Poll` will maintain an internal readiness queue, represented as a\r\nlinked list. The linked list head pointer is an `AtomicPtr`. All of the\r\nnodes in the linked list are owned by the `Poll` instance.\r\n\r\nThe type declarations are for illustration only. The actual\r\nimplementations will have some additional memory safety requirements.\r\n\r\n```rust\r\nstruct Poll {\r\n\treadiness_queue: Arc<PollReadinessQueue>,\r\n}\r\n\r\nstruct PollReadinessQueue {\r\n\t// All readiness nodes owned by the `Poll` instance. When the `Poll`\r\n\t// instance is freed, the list is walked and each Arc ref count is\r\n\t// decremented.\r\n\thead_all_nodes: Box<ReadinessNode>,\r\n\r\n\t// linked list of nodes that are pending some processing\r\n\thead_readiness: AtomicPtr<ReadinessNode>,\r\n\r\n    // Hashed wheel timer for delayed readiness notifications\r\n    readiness_wheel: Vec<AtomicPtr<ReadinessNode>>,\r\n}\r\n\r\nstruct ReadinessNode {\r\n\t// Next node in ownership tracking queue\r\n\tnext_all_nodes: Box<ReadinessNode>,\r\n    // Used when the node is queued in the readiness linked list OR the\r\n    // linked list for a hashed wheel slot.\r\n\tnext_readiness: *mut ReadinessNode,\r\n\t// The Token used to register the `Evented` with `Poll`. This can change,\r\n\t// but only by calling `Poll` functions, so there will be no concurrency.\r\n\ttoken: Token,\r\n\t// The set of events to include in the notification on next poll\r\n\tevents: AtomicUsize,\r\n    // Tracks if the node is queued for readiness using the MSB, the\r\n    // rest of the usize is the readiness delay.\r\n\tqueued: AtomicUsize,\r\n\t// Both interest and opts can be mutated\r\n\tinterest: Cell<EventSet>,\r\n    // Poll opts\r\n\topts: Cell<PollOpt>,\r\n}\r\n\r\n// Implements `Sync`, aka all functions are safe to call concurrently\r\nstruct Registration {\r\n\tnode: *mut ReadinessNode,\r\n\tqueue: Arc<PollReadinessQueue>,\r\n}\r\n\r\nstruct MyEvented {\r\n\tregistration: Option<Registration>,\r\n}\r\n```\r\n\r\n#### Registration\r\n\r\nWhen a `MyEvented` value is registered with the event loop, a new `Registration` value is obtained:\r\n\r\n```rust\r\nmy_evented.registration = Some(Registration::new(poll, token, interest));\r\n```\r\n\r\n`Registration` will include the internal `EventSet::dropped()` event to\r\nthe interest.\r\n\r\n#### Re-registration\r\n\r\nA `Registration`'s `interest` & `PollOpt` can be changed by calling `Registration::update`:\r\n\r\n```rust\r\n// poll: &Poll\r\nmy_evented.registration.as_ref().unwrap()\r\n\t.update(poll, interest, opts);\r\n```\r\n\r\nThe `Poll` reference will not be used but will ensure that `update` is\r\nonly called from a single thread (the thread that owns the `Poll`\r\nreference). This allows safe mutation of `interest` and `opts` without\r\nsynchronization primitives.\r\n\r\n`Registration` will include the internal `EventSet::dropped()` event to\r\nthe interest.\r\n\r\n#### Triggering readiness notifications\r\n\r\nReadiness can be updated using `Registration::set_readiness` and\r\n`Registration::unset_readiness`. These can be called concurrently.\r\n`set_readiness` adds the given events with the existing `Registration`\r\nreadiness. `unset_readiness` subtracts the given events from the\r\nexisting `Registration`.\r\n\r\n```rust\r\nmy_evented.registration.as_ref().unwrap().set_readiness(EventSet::readable());\r\nmy_evented.registration.as_ref().unwrap().unset_readiness(EventSet::readable());\r\n```\r\n\r\n`Registration::set_readiness` ensures that the registration node is queued for processing. \r\n\r\n#### Delaying readiness\r\n\r\nIn order to support timeouts, `Registration` has the ability to schedule\r\nreadiness notifications using `Registration::delay_readiness(events,\r\ntimeout)`.\r\n\r\nThere is a big caveat. There is precise timing guarantee. A delayed\r\nreadiness event could be triggered much earlier than requested.  Also,\r\nthe readiness timer is coarse grained, so by default will be rounded to\r\n100ms or so. The one guarantee is that the event will be triggered no\r\nlater than the requested timeout + the duration of a timer tick (100ms\r\nby default).\r\n\r\n#### Queuing `Registration` for processing\r\n\r\nFirst, atomically update `Registration.queued`. Attempt to set the MSB.\r\nCheck the current delay value. If the requested delay is less than the\r\ncurrent, update the delayed portion of `queued`.\r\n\r\nIf the MSB was successfully set, then the current thread is responsible\r\nfor queuing the registration node (pseudocode):\r\n\r\n```\r\nloop {\r\n\tlet ptr = PollReadinessQueue.readiness_head.get();\r\n    ReadinessNode.next_readiness = ptr;\r\n\r\n    if PollReadinessQueue.readiness_head.compare_and_swap(ptr, &ReadinessNode) {\r\n        return;\r\n    }\r\n}\r\n```\r\n\r\n#### Dropping `Registration`\r\n\r\nProcessing a drop is handled by setting readiness to an internal\r\n`Dropped` event:\r\n\r\n```rust\r\nfn drop(&mut self) {\r\n    self.registration.as_ref().unwrap()\r\n        .set_readiness(EventSet::dropped());\r\n}\r\n```\r\n\r\nThe `Registration` value itself does not own any data, so there is\r\nnothing else to do.\r\n\r\n#### Polling\r\n\r\nOn `Poll::poll()` the following happens:\r\n\r\nReset the events on self\r\n\r\n```rust\r\nself.events.clear();\r\n```\r\n\r\nAtomically take ownership of the readiness queue:\r\n\r\n```rust\r\nlet ready_nodes = PollReadinessQueue.readiness_head.swap(ptr::null());\r\n```\r\n\r\nThe dequeued nodes are processed.\r\n\r\n```rust\r\nfor node in ready_nodes {\r\n    // Mask the readiness info by the node's interest. This is needed to\r\n    // support concurrent setting of readiness. Another thread may not\r\n    // be aware of the latest interest value.\r\n    let mut events = node.events.get() & node.interest;\r\n\r\n    // Used to read the delay component of `Registration::queued`.\r\n    let delay;\r\n\r\n    if opts.is_edge() || events.is_empty() {\r\n        // If the registration is edge, the node is always dequeued. If\r\n        // it is level, we only dequeue the event when there are no\r\n        // events (aka, no readiness). By not dequeing the event it will\r\n        // be processed again next call to `poll`\r\n        delay = unset_msb_and_read_delay_component(&node.queued);\r\n\r\n        // Reload the events to ensure that we don't \"lose\" any\r\n        // readiness notifications. Remember, it's ok to have\r\n        // spurious notifications. \r\n        events = node.events.get() | node.interest;\r\n    } else if !events.is_drop() {\r\n        // Push the node back into the queue. This is done via a compare\r\n        // and swap on `readiness_head`, pushing the node back to the\r\n        // front.\r\n        prepend(&ready_nodes, node);\r\n\r\n        delay = read_delay_component(&node.queued);\r\n    }\r\n\r\n    if delay > 0 {\r\n        node.update_delay_in_hashed_wheel(delay);\r\n    } else {\r\n        // The event will be fired immediately, if the node is currently\r\n        // being delayed, remove it from the hashed wheel.\r\n        if node.is_currently_in_hashed_wheel() {\r\n            node.remove_from_hashed_wheel();\r\n        }\r\n\r\n        if events.is_drop() {\r\n            // The registration is dropped. Unlink the node, freeing the\r\n            // memory.\r\n            node.unlink();\r\n            continue;\r\n        }\r\n\r\n        if !events.is_empty() {\r\n            // Track the event\r\n            self.events.push_event(node.token, events);\r\n        }\r\n    }\r\n\r\n}\r\n```\r\n\r\nThe next step is to process all delayed readiness nodes that have\r\nreached their timeout. The code for this is similar to the current timer\r\ncode.\r\n\r\n### Integrating with `Selector`\r\n\r\nThe readiness queue described above is *not* to replace socket\r\nnotifications on epoll / kqueue / etc... It is to be used in conjuction.\r\n\r\nTo handle this, `PollReadinessQueue` will be able to wakup the selector.\r\nThis will be implemented in a similar fashion as the current channel\r\nimplementation. A pipe will be used to force the selector to wakeup.\r\n\r\nThe full logic of `poll` will look something like:\r\n\r\n```rust\r\nlet has_pending = !readiness_queue.is_empty();\r\n\r\nif has_pending {\r\n    // Original timeout value is passed to the function...\r\n    timeout = 0;\r\n}\r\n\r\n// Poll selector\r\nselector.poll(&mut self.events, timeout);\r\n\r\n// Process custom evented readiness queue as specified above.\r\n```\r\n\r\n### Implementing `mio::Channel`\r\n\r\n`Channel` is a mpsc queue such that when messages are pushed onto the\r\nchannel, `Poll` is woken up and returns a readable readiness event for\r\nthe `Channel`. The specific queue will be supplied on creation of\r\n`Channel`, allowing the user to choose the behavior around allocation\r\nand capacity.\r\n\r\n`Channel` will look something like:\r\n\r\n```rust\r\nstruct Channel<Q> {\r\n    queue: Q,\r\n\r\n    // Poll registration\r\n    registration: Option<Registration>,\r\n\r\n    // Tracks the number of pending messages.\r\n    pending: AtomicUsize,\r\n}\r\n```\r\n\r\nWhen a new message is sent over the channel:\r\n\r\n```rust\r\nself.queue.push(msg);\r\n\r\nlet prev = self.pending.fetch_add(1);\r\n\r\nif prev == 0 {\r\n    // set readiness\r\n    self.registration.as_ref().unwrap()\r\n        .set_readiness(EventSet::readable());\r\n}\r\n```\r\n\r\nWhen readiness is set, `Poll` will wake up with a readiness\r\nnotification. The user can now \"poll\" off of the channel. The\r\nimplementation of poll is something like:\r\n\r\n```rust\r\nself.queue.poll().map(|msg| {\r\n    let first = self.pending.get();\r\n\r\n    if first == 1 {\r\n        self.registration.as_ref().unwrap()\r\n            .unset_readiness(EventSet::readable());\r\n    }\r\n\r\n    let second = self.pending.fetch_sub(1);\r\n\r\n    if first == 1 && second > 0 {\r\n        // There still are pending messages, reset readiness\r\n        self.registration.as_ref().unwrap()\r\n            .set_readiness(EventSet::readable());\r\n    }\r\n\r\n    msg\r\n})\r\n```\r\n\r\n### Implemented `Timer`\r\n\r\n`Timer` is a delay queue. Messages are pushed onto it with a delay after\r\nwhich the message can be \"popped\" from the queue. It is implemented\r\nusing a hashed wheel timer strategy which is ideal in situations where\r\nlarge number of timeouts are required and the timer can use coarse\r\nprecision (by default, 100ms ticks).\r\n\r\nThe implementation is fairly straight forward. When a timeout is\r\nrequested, the message is stored in the `Timer` implementation and\r\n`Registration::delay_readiness` is called with the timeout. There are\r\nsome potential optimizations, but those are out of scope for this\r\nproposal.\r\n\r\n### Windows\r\n\r\nThe readiness queue described in this proposal would replace the [current\r\nwindows specific\r\nimplementation](https://github.com/carllerche/mio/blob/master/src/sys/windows/selector.rs).\r\nThe proposal implementation would be more efficient as it avoids locking\r\nas well as uses lighter weight data structures (mostly, linked lists vs.\r\nvecs).\r\n\r\n## Outstanding questions\r\n\r\nThe biggest outstanding question would be what to do about `EventLoop`.\r\nIf this proposal lands, then `EventLoop` becomes entirely a very light\r\nshim around `Poll` that dispatches events to the appropriate handler\r\nfunction.\r\n\r\nThe entire implementation would look something like:\r\n\r\n```rust\r\npub fn run(&mut self, handler: &mut H) -> io::Result<()> {\r\n    self.run = true;\r\n\r\n    while self.run {\r\n        self.poll.poll();\r\n\r\n        for event in self.poll.events() {\r\n            handler.ready(self, event.token(), event.kind());\r\n        }\r\n\r\n        handler.tick(self);\r\n    }\r\n}\r\n```\r\n\r\nIt will also not be possible to maintain API compatibility.\r\n`Handler::notify` and `Handler::timeout` will no longer exist as\r\n`EventLoop` does not know the difference between those two types and\r\nother `Evented` types that have notifications called through `ready`.\r\n\r\nThe options are:\r\n\r\n- Update `EventLoop` to follow the new API and keep the minimal\r\n  impelmentation.\r\n- Get rid of `EventLoop` and make `Poll` the primary API\r\n- Provide a hire level API via `EventLoop` that accepts allocations\r\n  (though this would be post 1.0).\r\n\r\n## Alternatives\r\n\r\nIt is possible to implement `Timer` and `Channel` as standalone types\r\nwithout having to implement the readiness queue. For `Timer`, it would\r\nrequire using timerfd on linux and a timer thread on other platforms.\r\nThe disadvanage here is minor for linux as syscalls can be reduced\r\nsignificantly by only using `timerfd` to track the next timeout in the\r\n`Timer` vs. every timeout in `Timer`.\r\n\r\nHowever, on platforms that don't have `timerfd` available, a polyfill\r\nwill be needed. This can be done by creating a pipe and spawning a\r\nthread. When a timeout is needed, send a request to the thread. The\r\nthread writes a byte to the pipe after the timeout has expired. This has\r\noverhead, but again it can be amortized by only using the thread/pipe\r\ncombo for the next timeout in `Timer` vs. every timeout. Though, there\r\nmay be some complication with this amoritization when registering the\r\n`Timer` using level triggered notifications.\r\n\r\nOn the other hand. For `Channel`, a syscall would be needed for each\r\nmessage enqueued and dequeued. The implementation would be to have a\r\npipe associated with the `Chanenl`. Each time a message is enqueued,\r\nwrite a byte to the pipe. Whenever a message is dequeued, read a byte.\r\n"},"comment":{"url":"https://api.github.com/repos/carllerche/mio/issues/comments/189544769","html_url":"https://github.com/carllerche/mio/issues/360#issuecomment-189544769","issue_url":"https://api.github.com/repos/carllerche/mio/issues/360","id":189544769,"user":{"login":"dpc","id":9209,"avatar_url":"https://avatars.githubusercontent.com/u/9209?v=3","gravatar_id":"","url":"https://api.github.com/users/dpc","html_url":"https://github.com/dpc","followers_url":"https://api.github.com/users/dpc/followers","following_url":"https://api.github.com/users/dpc/following{/other_user}","gists_url":"https://api.github.com/users/dpc/gists{/gist_id}","starred_url":"https://api.github.com/users/dpc/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dpc/subscriptions","organizations_url":"https://api.github.com/users/dpc/orgs","repos_url":"https://api.github.com/users/dpc/repos","events_url":"https://api.github.com/users/dpc/events{/privacy}","received_events_url":"https://api.github.com/users/dpc/received_events","type":"User","site_admin":false},"created_at":"2016-02-27T01:08:30Z","updated_at":"2016-02-27T01:08:30Z","body":"I don't know if \"one timer per event loop\" is really 99% of use cases. In mioco each coroutine can use one or more timers. Mioco has a test running million coroutines (each sleeping for a while on it's timer) and it works just fine, so I hope in new design such scenario won't suffer. I guess it's the same as @tailhook 's point 3.\r\n\r\nAlso, I'd like to make sure there are no fixed size only designs and growable options can still be supported. Eg. in the above tests the wheel size needs to be adjusted, and I was going to fill issue to ask for \"just grow the time wheel if needed\" setting. I understand the point of \"no allocations for performance\", but IMO growing instead of crashing is much better proposition for anything except embedded applications."}}