GitHub PullRequestReviewCommentEvent at 2014-02-06 04:13:55
Repository: rust
Actor: alexcrichton
Payload: {"comment":{"url":"https://api.github.com/repos/mozilla/rust/pulls/comments/9489757","id":9489757,"diff_hunk":"@@ -0,0 +1,382 @@\n+// Copyright 2014 The Rust Project Developers. See the COPYRIGHT\n+// file at the top-level directory of this distribution and at\n+// http://rust-lang.org/COPYRIGHT.\n+//\n+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or\n+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license\n+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your\n+// option. This file may not be copied, modified, or distributed\n+// except according to those terms.\n+\n+/// Oneshot channels/ports\n+///\n+/// This is the initial flavor of channels/ports used for comm module. This is\n+/// an optimization for the one-use case of a channel. The major optimization of\n+/// this type is to have one and exactly one allocation when the chan/port pair\n+/// is created.\n+///\n+/// Another possible optimization would be to not use an UnsafeArc box because\n+/// in theory we know when the shared packet can be deallocated (no real need\n+/// for the atomic reference counting), but I was having trouble how to destroy\n+/// the data early in a drop of a Port.\n+///\n+/// # Implementation\n+///\n+/// Oneshots are implemented around one atomic uint variable. This variable\n+/// indicates both the state of the port/chan but also contains any tasks\n+/// blocked on the port. All atomic operations happen on this one word.\n+///\n+/// In order to upgrade a oneshot channel, an upgrade is considered a disconnect\n+/// on behalf of the channel side of things (it can be mentally thought of as\n+/// consuming the port). This upgrade is then also stored in the shared packet.\n+/// The one caveat to consider is that when a port sees a disconnected channel\n+/// it must check for data because there is no \"data plus upgrade\" state.\n+\n+use comm::Port;\n+use kinds::Send;\n+use ops::Drop;\n+use option::{Some, None, Option};\n+use result::{Result, Ok, Err};\n+use rt::local::Local;\n+use rt::task::{Task, BlockedTask};\n+use sync::atomics;\n+use util;\n+\n+// Various states you can find a port in.\n+static EMPTY: uint = 0;\n+static DATA: uint = 1;\n+static DISCONNECTED: uint = 2;\n+\n+pub struct Packet<T> {\n+    // Internal state of the chan/port pair (stores the blocked task as well)\n+    state: atomics::AtomicUint,\n+    // One-shot data slot location\n+    data: Option<T>,\n+    // when used for the second time, a oneshot channel must be upgraded, and\n+    // this contains the slot for the upgrade\n+    upgrade: MyUpgrade<T>,\n+}\n+\n+pub enum Failure<T> {\n+    Empty,\n+    Disconnected,\n+    Upgraded(Port<T>),\n+}\n+\n+pub enum UpgradeResult {\n+    UpSuccess,\n+    UpDisconnected,\n+    UpWoke(BlockedTask),\n+}\n+\n+pub enum SelectionResult<T> {\n+    SelCanceled(BlockedTask),\n+    SelUpgraded(BlockedTask, Port<T>),\n+    SelSuccess,\n+}\n+\n+enum MyUpgrade<T> {\n+    NothingSent,\n+    SendUsed,\n+    GoUp(Port<T>),\n+}\n+\n+impl<T: Send> Packet<T> {\n+    pub fn new() -> Packet<T> {\n+        Packet {\n+            data: None,\n+            upgrade: NothingSent,\n+            state: atomics::AtomicUint::new(EMPTY),\n+        }\n+    }\n+\n+    pub fn send(&mut self, t: T) -> bool {\n+        // Sanity check\n+        match self.upgrade {\n+            NothingSent => {}\n+            _ => fail!(\"sending on a oneshot that's already sent on \"),\n+        }\n+        assert!(self.data.is_none());\n+        self.data = Some(t);\n+        self.upgrade = SendUsed;\n+\n+        // This atomic swap uses a \"Release\" memory ordering to ensure that all\n+        // our previous memory writes are visible to the other thread (notably\n+        // the write of data/upgrade)\n+        match self.state.swap(DATA, atomics::Release) {\n+            // Sent the data, no one was waiting\n+            EMPTY => true,\n+\n+            // Couldn't send the data, the port hung up first. We need to be\n+            // sure to deallocate the sent data (to not leave it stuck in the\n+            // queue)\n+            DISCONNECTED => {\n+                self.data.take_unwrap();\n+                false\n+            }\n+\n+            // Not possible, these are one-use channels\n+            DATA => unreachable!(),\n+\n+            // Anything else means that there was a task waiting on the other\n+            // end. We leave the 'DATA' state inside so it'll pick it up on the\n+            // other end.\n+            n => unsafe {\n+                let t = BlockedTask::cast_from_uint(n);\n+                t.wake().map(|t| t.reawaken());\n+                true\n+            }\n+        }\n+    }\n+\n+    // Just tests whether this channel has been sent on or not, this is only\n+    // safe to use from the sender.\n+    pub fn sent(&self) -> bool {\n+        match self.upgrade {\n+            NothingSent => false,\n+            _ => true,\n+        }\n+    }\n+\n+    pub fn recv(&mut self) -> Result<T, Failure<T>> {\n+        // Attempt to not block the task (it's a little expensive). If it looks\n+        // like we're not empty, then immediately go through to `try_recv`.\n+        //\n+        // These atomics use an Acquire memory ordering in order to have all the\n+        // previous writes of the releasing thread visible to us.\n+        if self.state.load(atomics::Acquire) == EMPTY {\n+            let t: ~Task = Local::take();\n+            t.deschedule(1, |task| {\n+                let n = unsafe { task.cast_to_uint() };\n+                match self.state.compare_and_swap(EMPTY, n, atomics::Acquire) {\n+                    // Nothing on the channel, we legitimately block\n+                    EMPTY => Ok(()),\n+\n+                    // If there's data or it's a disconnected channel, then we\n+                    // failed the cmpxchg, so we just wake ourselves back up\n+                    DATA | DISCONNECTED => {\n+                        unsafe { Err(BlockedTask::cast_from_uint(n)) }\n+                    }\n+\n+                    // Only one thread is allowed to sleep on this port\n+                    _ => unreachable!()\n+                }\n+            });\n+        }\n+\n+        self.try_recv()\n+    }\n+\n+    pub fn try_recv(&mut self) -> Result<T, Failure<T>> {\n+        // see above for why Acquire is used.\n+        match self.state.load(atomics::Acquire) {\n+            EMPTY => Err(Empty),\n+\n+            // We saw some data on the channel, but the channel can be used\n+            // again to send us an upgrade. As a result, we need to re-insert\n+            // into the channel that there's no data available (otherwise we'll\n+            // just see DATA next time). This is done as a cmpxchg because if\n+            // the state changes under our feet we'd rather just see that state\n+            // change.\n+            DATA => {\n+                self.state.compare_and_swap(DATA, EMPTY, atomics::Acquire);\n+                match self.data.take() {\n+                    Some(data) => Ok(data),\n+                    None => unreachable!(),\n+                }\n+            }\n+\n+            // There's no guarantee that we receive before an upgrade happens,\n+            // and an upgrade flags the channel as disconnected, so when we see\n+            // this we first need to check if there's data available and *then*\n+            // we go through and process the upgrade.\n+            DISCONNECTED => {\n+                match self.data.take() {\n+                    Some(data) => Ok(data),\n+                    None => {\n+                        match util::replace(&mut self.upgrade, SendUsed) {\n+                            SendUsed | NothingSent => Err(Disconnected),\n+                            GoUp(upgrade) => Err(Upgraded(upgrade))\n+                        }\n+                    }\n+                }\n+            }\n+            _ => unreachable!()\n+        }\n+    }\n+\n+    // Returns whether the upgrade was completed. If the upgrade wasn't\n+    // completed, then the port couldn't get sent to the other half (it will\n+    // never receive it).\n+    pub fn upgrade(&mut self, up: Port<T>) -> UpgradeResult {\n+        let prev = match self.upgrade {\n+            NothingSent => NothingSent,\n+            SendUsed => SendUsed,\n+            _ => fail!(\"upgrading again\"),\n+        };\n+        self.upgrade = GoUp(up);\n+\n+        // Use a Release memory ordering in order to make sure that our write to\n+        // `upgrade` is visible to the other thread.\n+        match self.state.swap(DISCONNECTED, atomics::Release) {\n+            // If the channel is empty or has data on it, then we're good to go.\n+            // Senders will check the data before the upgrade (in case we\n+            // plastered over the DATA state).\n+            DATA | EMPTY => UpSuccess,\n+\n+            // If the other end is already disconnected, then we failed the\n+            // upgrade. Be sure to trash the port we were given.\n+            DISCONNECTED => { self.upgrade = prev; UpDisconnected }\n+\n+            // If someone's waiting, we gotta wake them up\n+            n => UpWoke(unsafe { BlockedTask::cast_from_uint(n) })\n+        }\n+    }\n+\n+    pub fn drop_chan(&mut self) {\n+        match self.state.swap(DISCONNECTED, atomics::SeqCst) {\n+            DATA | DISCONNECTED | EMPTY => {}\n+\n+            // If someone's waiting, we gotta wake them up\n+            n => unsafe {\n+                let t = BlockedTask::cast_from_uint(n);\n+                t.wake().map(|t| t.reawaken());\n+            }\n+        }\n+    }\n+\n+    pub fn drop_port(&mut self) {\n+        // Use an Acquire memory ordering in order to see the data that the\n+        // senders are sending.\n+        match self.state.swap(DISCONNECTED, atomics::Acquire) {\n+            // An empty channel has nothing to do, and a remotely disconnected\n+            // channel also has nothing to do b/c we're about to run the drop\n+            // glue\n+            DISCONNECTED | EMPTY => {}\n+\n+            // There's data on the channel, so make sure we destroy it promptly.\n+            // This is why not using an arc is a little difficult (need the box\n+            // to stay valid while we take the data).\n+            DATA => { self.data.take_unwrap(); }\n+\n+            // We're the only ones that can block on this port\n+            _ => unreachable!()\n+        }\n+    }\n+\n+    ////////////////////////////////////////////////////////////////////////////\n+    // select implementation\n+    ////////////////////////////////////////////////////////////////////////////\n+\n+    // If Ok, the value is whether this port has data, if Err, then the upgraded\n+    // port needs to be checked instead of this one.\n+    pub fn can_recv(&mut self) -> Result<bool, Port<T>> {\n+        // Use Acquire so we can see all previous memory writes\n+        match self.state.load(atomics::Acquire) {\n+            EMPTY => Ok(false), // Welp, we tried\n+            DATA => Ok(true),   // we have some un-acquired data\n+            DISCONNECTED if self.data.is_some() => Ok(true), // we have data\n+            DISCONNECTED => {\n+                match util::replace(&mut self.upgrade, SendUsed) {\n+                    // The other end sent us an upgrade, so we need to\n+                    // propagate upwards whether the upgrade can receive\n+                    // data\n+                    GoUp(upgrade) => Err(upgrade),\n+\n+                    // If the other end disconnected without sending an\n+                    // upgrade, then we have data to receive (the channel is\n+                    // disconnected).\n+                    up => { self.upgrade = up; Ok(true) }\n+                }\n+            }\n+            _ => unreachable!(), // we're the \"one blocker\"\n+        }\n+    }\n+\n+    // Attempts to start selection on this port. This can either succeed, fail\n+    // because there is data, or fail because there is an upgrade pending.\n+    pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {\n+        let n = unsafe { task.cast_to_uint() };\n+        match self.state.compare_and_swap(EMPTY, n, atomics::SeqCst) {\n+            EMPTY => SelSuccess,\n+            DATA => SelCanceled(unsafe { BlockedTask::cast_from_uint(n) }),\n+            DISCONNECTED if self.data.is_some() => {\n+                SelCanceled(unsafe { BlockedTask::cast_from_uint(n) })\n+            }\n+            DISCONNECTED => {\n+                match util::replace(&mut self.upgrade, SendUsed) {\n+                    // The other end sent us an upgrade, so we need to\n+                    // propagate upwards whether the upgrade can receive\n+                    // data\n+                    GoUp(upgrade) => {\n+                        SelUpgraded(unsafe { BlockedTask::cast_from_uint(n) },\n+                                    upgrade)\n+                    }\n+\n+                    // If the other end disconnected without sending an\n+                    // upgrade, then we have data to receive (the channel is\n+                    // disconnected).\n+                    up => {\n+                        self.upgrade = up;\n+                        SelCanceled(unsafe { BlockedTask::cast_from_uint(n) })\n+                    }\n+                }\n+            }\n+            _ => unreachable!(), // we're the \"one blocker\"\n+        }\n+    }\n+\n+    // Remove a previous selecting task from this port. This ensures that the\n+    // blocked task will no longer be visible to any other threads.\n+    //\n+    // The return value indicates whether there's data on this port.\n+    pub fn abort_selection(&mut self) -> Result<bool, Port<T>> {\n+        // use Acquire to make sure we see all previous memory writes\n+        let state = match self.state.load(atomics::Acquire) {\n+            // Each of these states means that no further activity will happen\n+            // with regard to abortion selection\n+            s @ EMPTY |\n+            s @ DATA |\n+            s @ DISCONNECTED => s,\n+\n+            // If we've got a blocked task, then use an atomic to gain ownership\n+            // of it (may fail)\n+            n => self.state.compare_and_swap(n, EMPTY, atomics::SeqCst)\n+        };\n+\n+        // Now that we've got ownership of our state, figure out what to do\n+        // about it.\n+        match state {\n+            EMPTY => unreachable!(),\n+            // our task used for select was stolen\n+            DATA => Ok(true),\n+\n+            // If the other end has hung up, then we have complete ownership\n+            // of the port. We need to check to see if there was an upgrade\n+            // requested, and if so, the other end needs to have its selection\n+            // aborted.\n+            DISCONNECTED => {\n+                assert!(self.data.is_none());\n+                match util::replace(&mut self.upgrade, SendUsed) {\n+                    GoUp(port) => Err(port),\n+                    _ => Ok(true),\n+                }\n+            }","path":"src/libstd/comm/oneshot.rs","position":364,"original_position":364,"commit_id":"abee44e10ab65192d6c808b98d6af722a8f32d1b","original_commit_id":"abee44e10ab65192d6c808b98d6af722a8f32d1b","user":{"login":"alexcrichton","id":64996,"avatar_url":"https://gravatar.com/avatar/477e2098d9864b482d28e5ef8e657f43?d=https%3A%2F%2Fidenticons.github.com%2F621df6a7a3377f22b1ae3afc962c3f68.png&r=x","gravatar_id":"477e2098d9864b482d28e5ef8e657f43","url":"https://api.github.com/users/alexcrichton","html_url":"https://github.com/alexcrichton","followers_url":"https://api.github.com/users/alexcrichton/followers","following_url":"https://api.github.com/users/alexcrichton/following{/other_user}","gists_url":"https://api.github.com/users/alexcrichton/gists{/gist_id}","starred_url":"https://api.github.com/users/alexcrichton/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/alexcrichton/subscriptions","organizations_url":"https://api.github.com/users/alexcrichton/orgs","repos_url":"https://api.github.com/users/alexcrichton/repos","events_url":"https://api.github.com/users/alexcrichton/events{/privacy}","received_events_url":"https://api.github.com/users/alexcrichton/received_events","type":"User","site_admin":false},"body":"This case changed in order to transmit that an upgrade is waiting.","created_at":"2014-02-06T04:13:55Z","updated_at":"2014-02-06T04:13:55Z","html_url":"https://github.com/mozilla/rust/pull/11578#discussion_r9489757","pull_request_url":"https://api.github.com/repos/mozilla/rust/pulls/11578","_links":{"self":{"href":"https://api.github.com/repos/mozilla/rust/pulls/comments/9489757"},"html":{"href":"https://github.com/mozilla/rust/pull/11578#discussion_r9489757"},"pull_request":{"href":"https://api.github.com/repos/mozilla/rust/pulls/11578"}}}}