<commit_before>//! Definition of the Shared combinator, a future that is cloneable,
//! and can be polled in multiple threads.
//!
//! # Examples
//!
//! ```
//! use futures::future::*;
//!
//! let future = ok::<_, bool>(6);
//! let shared1 = future.shared();
//! let shared2 = shared1.clone();
//! assert_eq!(6, *shared1.wait().unwrap());
//! assert_eq!(6, *shared2.wait().unwrap());
//! ```

use {Future, Poll, Async};
use executor::{self, Spawn, Unpark};
use task::{self, Task};

use std::{fmt, mem, ops};
use std::cell::UnsafeCell;
use std::sync::{Arc, Mutex};
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::collections::HashMap;

/// A future that is cloneable and can be polled in multiple threads.
/// Use Future::shared() method to convert any future into a `Shared` future.
#[must_use = "futures do nothing unless polled"]
pub struct Shared<F: Future> {
    inner: Arc<Inner<F>>,
    waiter: usize,
}

impl<F> fmt::Debug for Shared<F>
    where F: Future + fmt::Debug,
          F::Item: fmt::Debug,
          F::Error: fmt::Debug,
{
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Shared")
            .field("inner", &self.inner)
            .field("waiter", &self.waiter)
            .finish()
    }
}

struct Inner<F: Future> {
    next_clone_id: AtomicUsize,
    future: UnsafeCell<Spawn<F>>,
    result: UnsafeCell<Option<Result<SharedItem<F::Item>, SharedError<F::Error>>>>,
    unparker: Arc<Unparker>,
}

struct Unparker {
    state: AtomicUsize,
    waiters: Mutex<HashMap<usize, Task>>,
}

const IDLE: usize = 0;
const POLLING: usize = 1;
const REPOLL: usize = 2;
const COMPLETE: usize = 3;
const POISONED: usize = 4;

impl<F> Shared<F> where F: Future {
    // TODO: make this private
    #[deprecated(since = "0.1.12", note = "use `Future::shared` instead")]
    #[cfg(feature = "with-deprecated")]
    #[doc(hidden)]
    pub fn new(future: F) -> Self {
        Shared {
            inner: Arc::new(Inner {
                next_clone_id: AtomicUsize::new(1),
                unparker: Arc::new(Unparker {
                    state: AtomicUsize::new(IDLE),
                    waiters: Mutex::new(HashMap::new()),
                }),
                future: UnsafeCell::new(executor::spawn(future)),
                result: UnsafeCell::new(None),
            }),
            waiter: 0,
        }
    }

    /// If any clone of this `Shared` has completed execution, returns its result immediately
    /// without blocking. Otherwise, returns None without triggering the work represented by
    /// this `Shared`.
    pub fn peek(&self) -> Option<Result<SharedItem<F::Item>, SharedError<F::Error>>> {
        match self.inner.unparker.state.load(SeqCst) {
            COMPLETE => {
                Some(unsafe { self.clone_result() })
            }
            POISONED => panic!("inner future panicked during poll"),
            _ => None,
        }
    }

    fn set_waiter(&mut self) {
        let mut waiters = self.inner.unparker.waiters.lock().unwrap();
        waiters.insert(self.waiter, task::park());
    }

    unsafe fn clone_result(&self) -> Result<SharedItem<F::Item>, SharedError<F::Error>> {
        match *self.inner.result.get() {
            Some(Ok(ref item)) => Ok(item.clone()),
            Some(Err(ref e)) => Err(e.clone()),
            _ => unreachable!(),
        }
    }

    fn complete(&self) {
        self.inner.unparker.state.store(COMPLETE, SeqCst);
        self.inner.unparker.unpark();
    }
}

impl<F> Future for Shared<F>
    where F: Future
{
    type Item = SharedItem<F::Item>;
    type Error = SharedError<F::Error>;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        self.set_waiter();

        match self.inner.unparker.state.compare_and_swap(IDLE, POLLING, SeqCst) {
            IDLE => {
                // Lock acquired, fall through
            }
            POLLING | REPOLL => {
                // Another task is currently polling, at this point we just want
                // to ensure that our task handle is currently registered

                return Ok(Async::NotReady);
            }
            COMPLETE => {
                return unsafe { self.clone_result().map(Async::Ready) };
            }
            POISONED => panic!("inner future panicked during poll"),
            _ => unreachable!(),
        }

        loop {
            struct Reset<'a>(&'a AtomicUsize);

            impl<'a> Drop for Reset<'a> {
                fn drop(&mut self) {
                    use std::thread;

                    if thread::panicking() {
                        self.0.store(POISONED, SeqCst);
                    }
                }
            }

            let _reset = Reset(&self.inner.unparker.state);

            // Get a handle to the unparker
            let unpark: Arc<Unpark> = self.inner.unparker.clone();

            // Poll the future
            match unsafe { (*self.inner.future.get()).poll_future(unpark) } {
                Ok(Async::NotReady) => {
                    // Not ready, try to release the handle
                    match self.inner.unparker.state.compare_and_swap(POLLING, IDLE, SeqCst) {
                        POLLING => {
                            // Success
                            return Ok(Async::NotReady);
                        }
                        REPOLL => {
                            // Gotta poll again!
                            let prev = self.inner.unparker.state.swap(POLLING, SeqCst);
                            assert_eq!(prev, REPOLL);
                        }
                        _ => unreachable!(),
                    }

                }
                Ok(Async::Ready(i)) => {
                    unsafe {
                        (*self.inner.result.get()) = Some(Ok(SharedItem { item: Arc::new(i) }));
                    }

                    break;
                }
                Err(e) => {
                    unsafe {
                        (*self.inner.result.get()) = Some(Err(SharedError { error: Arc::new(e) }));
                    }

                    break;
                }
            }
        }

        self.complete();
        unsafe { self.clone_result().map(Async::Ready) }
    }
}

impl<F> Clone for Shared<F> where F: Future {
    fn clone(&self) -> Self {
        let next_clone_id = self.inner.next_clone_id.fetch_add(1, SeqCst);

        Shared {
            inner: self.inner.clone(),
            waiter: next_clone_id,
        }
    }
}

impl Unpark for Unparker {
    fn unpark(&self) {
        self.state.compare_and_swap(POLLING, REPOLL, SeqCst);

        let waiters = mem::replace(&mut *self.waiters.lock().unwrap(), HashMap::new());

        for (_, waiter) in waiters {
            waiter.unpark();
        }
    }
}

unsafe impl<F: Future> Sync for Inner<F> {}
unsafe impl<F: Future> Send for Inner<F> {}

impl<F> fmt::Debug for Inner<F>
    where F: Future + fmt::Debug,
          F::Item: fmt::Debug,
          F::Error: fmt::Debug,
{
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Inner")
            .finish()
    }
}

/// A wrapped item of the original future that is clonable and implements Deref
/// for ease of use.
#[derive(Debug)]
pub struct SharedItem<T> {
    item: Arc<T>,
}

impl<T> ops::Deref for SharedItem<T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.item.as_ref()
    }
}

impl<T> Clone for SharedItem<T> {
    fn clone(&self) -> Self {
        SharedItem { item: self.item.clone() }
    }
}

/// A wrapped error of the original future that is clonable and implements Deref
/// for ease of use.
#[derive(Debug)]
pub struct SharedError<E> {
    error: Arc<E>,
}

impl<E> ops::Deref for SharedError<E> {
    type Target = E;

    fn deref(&self) -> &E {
        &self.error.as_ref()
    }
}

impl<T> Clone for SharedError<T> {
    fn clone(&self) -> Self {
        SharedError { error: self.error.clone() }
    }
}
<commit_msg>Add Drop impl for Shared<commit_after>//! Definition of the Shared combinator, a future that is cloneable,
//! and can be polled in multiple threads.
//!
//! # Examples
//!
//! ```
//! use futures::future::*;
//!
//! let future = ok::<_, bool>(6);
//! let shared1 = future.shared();
//! let shared2 = shared1.clone();
//! assert_eq!(6, *shared1.wait().unwrap());
//! assert_eq!(6, *shared2.wait().unwrap());
//! ```

use {Future, Poll, Async};
use executor::{self, Spawn, Unpark};
use task::{self, Task};

use std::{fmt, mem, ops};
use std::cell::UnsafeCell;
use std::sync::{Arc, Mutex};
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::collections::HashMap;

/// A future that is cloneable and can be polled in multiple threads.
/// Use Future::shared() method to convert any future into a `Shared` future.
#[must_use = "futures do nothing unless polled"]
pub struct Shared<F: Future> {
    inner: Arc<Inner<F>>,
    waiter: usize,
}

impl<F> fmt::Debug for Shared<F>
    where F: Future + fmt::Debug,
          F::Item: fmt::Debug,
          F::Error: fmt::Debug,
{
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Shared")
            .field("inner", &self.inner)
            .field("waiter", &self.waiter)
            .finish()
    }
}

struct Inner<F: Future> {
    next_clone_id: AtomicUsize,
    future: UnsafeCell<Spawn<F>>,
    result: UnsafeCell<Option<Result<SharedItem<F::Item>, SharedError<F::Error>>>>,
    unparker: Arc<Unparker>,
}

struct Unparker {
    state: AtomicUsize,
    waiters: Mutex<HashMap<usize, Task>>,
}

const IDLE: usize = 0;
const POLLING: usize = 1;
const REPOLL: usize = 2;
const COMPLETE: usize = 3;
const POISONED: usize = 4;

impl<F> Shared<F> where F: Future {
    // TODO: make this private
    #[deprecated(since = "0.1.12", note = "use `Future::shared` instead")]
    #[cfg(feature = "with-deprecated")]
    #[doc(hidden)]
    pub fn new(future: F) -> Self {
        Shared {
            inner: Arc::new(Inner {
                next_clone_id: AtomicUsize::new(1),
                unparker: Arc::new(Unparker {
                    state: AtomicUsize::new(IDLE),
                    waiters: Mutex::new(HashMap::new()),
                }),
                future: UnsafeCell::new(executor::spawn(future)),
                result: UnsafeCell::new(None),
            }),
            waiter: 0,
        }
    }

    /// If any clone of this `Shared` has completed execution, returns its result immediately
    /// without blocking. Otherwise, returns None without triggering the work represented by
    /// this `Shared`.
    pub fn peek(&self) -> Option<Result<SharedItem<F::Item>, SharedError<F::Error>>> {
        match self.inner.unparker.state.load(SeqCst) {
            COMPLETE => {
                Some(unsafe { self.clone_result() })
            }
            POISONED => panic!("inner future panicked during poll"),
            _ => None,
        }
    }

    fn set_waiter(&mut self) {
        let mut waiters = self.inner.unparker.waiters.lock().unwrap();
        waiters.insert(self.waiter, task::park());
    }

    unsafe fn clone_result(&self) -> Result<SharedItem<F::Item>, SharedError<F::Error>> {
        match *self.inner.result.get() {
            Some(Ok(ref item)) => Ok(item.clone()),
            Some(Err(ref e)) => Err(e.clone()),
            _ => unreachable!(),
        }
    }

    fn complete(&self) {
        self.inner.unparker.state.store(COMPLETE, SeqCst);
        self.inner.unparker.unpark();
    }
}

impl<F> Future for Shared<F>
    where F: Future
{
    type Item = SharedItem<F::Item>;
    type Error = SharedError<F::Error>;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        self.set_waiter();

        match self.inner.unparker.state.compare_and_swap(IDLE, POLLING, SeqCst) {
            IDLE => {
                // Lock acquired, fall through
            }
            POLLING | REPOLL => {
                // Another task is currently polling, at this point we just want
                // to ensure that our task handle is currently registered

                return Ok(Async::NotReady);
            }
            COMPLETE => {
                return unsafe { self.clone_result().map(Async::Ready) };
            }
            POISONED => panic!("inner future panicked during poll"),
            _ => unreachable!(),
        }

        loop {
            struct Reset<'a>(&'a AtomicUsize);

            impl<'a> Drop for Reset<'a> {
                fn drop(&mut self) {
                    use std::thread;

                    if thread::panicking() {
                        self.0.store(POISONED, SeqCst);
                    }
                }
            }

            let _reset = Reset(&self.inner.unparker.state);

            // Get a handle to the unparker
            let unpark: Arc<Unpark> = self.inner.unparker.clone();

            // Poll the future
            match unsafe { (*self.inner.future.get()).poll_future(unpark) } {
                Ok(Async::NotReady) => {
                    // Not ready, try to release the handle
                    match self.inner.unparker.state.compare_and_swap(POLLING, IDLE, SeqCst) {
                        POLLING => {
                            // Success
                            return Ok(Async::NotReady);
                        }
                        REPOLL => {
                            // Gotta poll again!
                            let prev = self.inner.unparker.state.swap(POLLING, SeqCst);
                            assert_eq!(prev, REPOLL);
                        }
                        _ => unreachable!(),
                    }

                }
                Ok(Async::Ready(i)) => {
                    unsafe {
                        (*self.inner.result.get()) = Some(Ok(SharedItem { item: Arc::new(i) }));
                    }

                    break;
                }
                Err(e) => {
                    unsafe {
                        (*self.inner.result.get()) = Some(Err(SharedError { error: Arc::new(e) }));
                    }

                    break;
                }
            }
        }

        self.complete();
        unsafe { self.clone_result().map(Async::Ready) }
    }
}

impl<F> Clone for Shared<F> where F: Future {
    fn clone(&self) -> Self {
        let next_clone_id = self.inner.next_clone_id.fetch_add(1, SeqCst);

        Shared {
            inner: self.inner.clone(),
            waiter: next_clone_id,
        }
    }
}

impl<F> Drop for Shared<F> where F: Future {
    fn drop(&mut self) {
        let mut waiters = self.inner.unparker.waiters.lock().unwrap();
        waiters.remove(&self.waiter);
    }
}

impl Unpark for Unparker {
    fn unpark(&self) {
        self.state.compare_and_swap(POLLING, REPOLL, SeqCst);

        let waiters = mem::replace(&mut *self.waiters.lock().unwrap(), HashMap::new());

        for (_, waiter) in waiters {
            waiter.unpark();
        }
    }
}

unsafe impl<F: Future> Sync for Inner<F> {}
unsafe impl<F: Future> Send for Inner<F> {}

impl<F> fmt::Debug for Inner<F>
    where F: Future + fmt::Debug,
          F::Item: fmt::Debug,
          F::Error: fmt::Debug,
{
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        fmt.debug_struct("Inner")
            .finish()
    }
}

/// A wrapped item of the original future that is clonable and implements Deref
/// for ease of use.
#[derive(Debug)]
pub struct SharedItem<T> {
    item: Arc<T>,
}

impl<T> ops::Deref for SharedItem<T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.item.as_ref()
    }
}

impl<T> Clone for SharedItem<T> {
    fn clone(&self) -> Self {
        SharedItem { item: self.item.clone() }
    }
}

/// A wrapped error of the original future that is clonable and implements Deref
/// for ease of use.
#[derive(Debug)]
pub struct SharedError<E> {
    error: Arc<E>,
}

impl<E> ops::Deref for SharedError<E> {
    type Target = E;

    fn deref(&self) -> &E {
        &self.error.as_ref()
    }
}

impl<T> Clone for SharedError<T> {
    fn clone(&self) -> Self {
        SharedError { error: self.error.clone() }
    }
}

{'fraction_non_alphanumeric': 0.14562388206994387, 'fraction_numerical': 0.0025905137852340714, 'mean_word_length': 1.8331294775467413, 'pattern_counts': {'":': 0, '<': 115, '<?xml version=': 0, '>': 171, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1316686', 'n_tokens_mistral': 5118, 'n_tokens_neox': 4924, 'n_words': 1167}