Compared to implementing the equivalent in futures(0.1) that appears to just have a single extra statement doing the pin-projection + passing the LocalWaker around?
One thing you’re missing here is an abstraction to allow dealing with futures that may have completed in an earlier poll. futures(0.3) uses MaybeDone for this, I’m not certain if this will help in this exact case, but I’m sure it’s possible to have something that would.
You’re currently over-matching a lot, I think almost all the Ok/Err patterns can be dropped and you can just bind the result directly, e.g.
(Some(_), _) => match this.1.poll(lw) {
Ready(Ok(v2)) => Ready(EitherOr::Both(Ok(this.2.take().unwrap()), Ok(v2))),
Ready(Err(e2)) => Ready(EitherOr::Both(Ok(this.2.take().unwrap()), Err(e2))),
_ => Pending,
},
can be written
(Some(_), _) => match this.1.poll(lw) {
Ready(v2) => Ready(EitherOr::Both(Ok(this.2.take().unwrap()), v2)),
Pending => Pending,
},
and using the ready! macro
(Some(_), _) => {
let v2 = ready!(this.1.poll(lw));
Ready(EitherOr::Both(Ok(this.2.take().unwrap()), v2))
},
fully simplifying using those gives something that doesn’t seem too bad to me (definitely feels like it could be simplified a bit more with a targeted abstraction).
pub fn join_on_ok<T1, T2, E1, E2>(
f1: impl Future<Output = Result<T1, E1>>,
f2: impl Future<Output = Result<T2, E2>>,
) -> impl Future<Output = EitherOr<Result<T1,E1>,Result<T2,E2>>> {
struct JoinFuture<T1, T2, F1, F2>(F1, F2, Option<Result<T1, E1>>, Option<Result<T2, E2>>);
impl<T1, T2, E1, E2, F1, F2> Future for JoinFuture<T1, T2, F1, F2>
where
F1: Future<Output = Result<T1, E1>>,
F2: Future<Output = Result<T2, E2>>,
{
type Output = EitherOr<Result<T1,E1>,Result<T2,E2>>;
fn poll(self: Pin<&mut Self>, lw: &LocalWaker) -> Poll<Self::Output> {
let this = unsafe {
let this = self.get_unchecked_mut();
(
Pin::new_unchecked(&mut this.0),
Pin::new_unchecked(&mut this.1),
&mut this.2,
&mut this.3,
)
};
match (&this.2, &this.3) {
(Some(_), Some(_)) => unreachable!(),
(Some(_), _) => {
let v2 = ready!(this.1.poll(lw);
Ready(EitherOr::Both(this.2.take().unwrap(), v2))
},
(_, Some(_)) => {
Ready(EitherOr::Both(ready!(this.0.poll(lw), this.3.take().unwrap()))
},
_ => match (this.0.poll(lw), this.1.poll(lw)) {
(Ready(v1), Ready(v2)) => (v1, v2),
(Ready(Err(e1)), _) => Ready(EitherOr::This(Err(e1)))
(_, Ready(Err(e2))) => Ready(EitherOr::That(Err(e2)))
(Ready(v1), Pending) => {
*this.2 = Some(v1);
Pending
},
(Pending, Ready(v2)) => {
*this.3 = Some(v2);
Pending
},
(Pending, Pending) => Pending,
},
}
}
}
JoinFuture(f1, f2, None, None)
}
I think the simple criteria for something that cannot be written using async fn is just anything that involves parallelism, that’s why I think these sorts of APIs are likely to be the first to migrate to std from futures once the core APIs are stabilized. But a lot of the more specialized parallel combinators like this can be built on top of the basic select! and join! macros, e.g.
pub async fn join_on_ok<T1, T2, E1, E2>(
f1: impl Future<Output = Result<T1, E1>>,
f2: impl Future<Output = Result<T2, E2>>,
) -> EitherOr<Result<T1,E1>,Result<T2,E2>> {
let (f1, f2) = (f1.fuse(), f2.fuse());
let (mut v1, mut v2) = (None, None)
loop {
select! {
v = f1 => match v {
Ok(v) => v1.replace(v),
Err(e) => match v2.take() {
Some(v2) => break EitherOr::Both(Err(e), Ok(v2)),
None => break EitherOr::This(Err(e)),
},
},
v = f2 => match v {
Ok(v) => v2.replace(v),
Err(e) => match v1.take() {
Some(v1) => break EitherOr::Both(Ok(v1), Err(e)),
None => break EitherOr::That(Err(e)),
},
},
complete => {
break EitherOr::Both(Ok(v1.take().unwrap()), Ok(v2.take().unwrap()))
}
}
}
}
(I realise this doesn’t fully handle the case where one future completes with a value simultaneously with the other future completing with an error, but I don’t think that’s an important case to handle)
