futures-rs: Shared can interact badly with futures that don't always poll their subfutures

(I’m opening this issue to continue the discussion from https://github.com/alexcrichton/futures-rs/pull/305.)

The problem is that Shared, as it is currently designed, can interact poorly with certain futures, such as the ModedFuture sketched below. We should either formalize some reason why ModedFuture is an invalid implementation of Future, or we should redesign Shared to accommodate this kind of situation.

extern crate futures;
extern crate tokio_core;

use futures::{Future, Poll};
use futures::sync::oneshot;
use std::rc::Rc;
use std::cell::RefCell;

enum Mode { Left, Right }

struct ModedFutureInner<F> where F: Future {
    mode: Mode,
    left: F,
    right: F,
    task: Option<::futures::task::Task>,
}

struct ModedFuture<F> where F: Future {
    inner: Rc<RefCell<ModedFutureInner<F>>>,
}

struct ModedFutureHandle<F> where F: Future {
    inner: Rc<RefCell<ModedFutureInner<F>>>,
}

impl <F> ModedFuture<F> where F: Future {
    pub fn new(left: F, right: F, mode: Mode) -> (ModedFutureHandle<F>, ModedFuture<F>) {
        let inner = Rc::new(RefCell::new(ModedFutureInner {
            left: left, right: right, mode: mode, task: None,
         }));
        (ModedFutureHandle { inner: inner.clone() }, ModedFuture { inner: inner })
    }
}

impl <F> ModedFutureHandle<F> where F: Future {
    pub fn switch_mode(&mut self, mode: Mode) {
        self.inner.borrow_mut().mode = mode;
        if let Some(t) = self.inner.borrow_mut().task.take() {
            // The other future may have become ready while we were ignoring it.
            t.unpark();
        }
    }
}

impl <F> Future for ModedFuture<F> where F: Future {
    type Item = F::Item;
    type Error = F::Error;
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        let ModedFutureInner { ref mut mode, ref mut left, ref mut right, ref mut task } =
            *self.inner.borrow_mut();
        *task = Some(::futures::task::park());
        match *mode {
            Mode::Left => left.poll(),
            Mode::Right => right.poll(),
        }
    }
}

pub fn main() {
    let mut core = ::tokio_core::reactor::Core::new().unwrap();
    let handle = core.handle();

    let (tx, rx) = oneshot::channel::<u32>();
    let f1 = rx.shared();
    let f2 = f1.clone();

    let (mut mfh, mf) = ModedFuture::new(
        Box::new(f1.map_err(|_| ()).map(|v| *v)) as Box<Future<Item=u32, Error=()>>,
        Box::new(::futures::future::empty()) as Box<Future<Item=u32, Error=()>>,
        Mode::Left);

    let (tx3, rx3) = oneshot::channel::<u32>();
    handle.spawn(f2.map(|x| tx3.complete(*x)).map_err(|_| ()));

    core.turn(Some(::std::time::Duration::from_millis(1)));

    handle.spawn(mf.map(|_| ()));

    core.turn(Some(::std::time::Duration::from_millis(1)));

    mfh.switch_mode(Mode::Right);

    tx.complete(11); // It seems like this ought to cause f2 and then rx3 to get resolved.

    // This hangs forever.
    core.run(rx3).unwrap();
}