Remove hyper fork

[DarrenTsung]

Use hyper 0.11 and associated libraries like tokio.

The flow of the project is largely unchanged to avoid having to refactor heavily downstream, but now we handle the threads ourselves with a RoundRobinPool and also use Executors to receive messages via futures::sync::mspc from the pool thread.

---

This change is 

[coder543]

if no one is listening, this will return an Error. Is that unimportant?

[DarrenTsung]

Good point, yeah it's unimportant because it's only used for the tests. I moved it over to the test mod, we don't really need to expose the Deliverable impl for mpsc::Sender anyways..

[jwilm]

Reviewed 5 of 10 files at r1, 1 of 2 files at r2, 3 of 3 files at r3, 1 of 5 files at r4.
Review status: 5 of 9 files reviewed at latest revision, 1 unresolved discussion.

---

src/config.rs, line 18 at r3 (raw file):

    /// Number of DNS threads per worker
    pub dns_threads_per_worker: usize,

Does hyper 0.11 still handle DNS? Seems like we should be able to plugin a tokio/futures-compatible resolver library and eliminate this option.

---

src/error.rs, line 9 at r3 (raw file):

#[derive(Debug)]
pub enum Error<D: Deliverable> {

The variants here need some explanation in doc comments. It would also be nice to expose a library Error type that buckets these into logical categories or something that is suggestive about how a consumer of this API should handle the errors.

---

src/error.rs, line 12 at r3 (raw file):

    ThreadSpawn(io::Error),
    HttpsConnector(hyper_tls::Error),
    Full(Transaction<D>),

Full? What does that mean?

---

src/error.rs, line 33 at r3 (raw file):

            Error::ThreadSpawn(err) => SpawnError::ThreadSpawn(err),
            Error::HttpsConnector(err) => SpawnError::HttpsConnector(err),
            _ => unreachable!(),

This doesn't seem unreachable

---

src/error.rs, line 43 at r3 (raw file):

            Error::Full(err) => RequestError::Full(err),
            Error::FailedSend(err) => RequestError::FailedSend(err),
            _ => unreachable!(),

Nor does this

---

src/executor.rs, line 17 at r3 (raw file):

use config::Config;
use counter::{Counter, WeakCounter};

Reviewable says counter module was removed in latest revision...

---

src/executor.rs, line 40 at r3 (raw file):

enum ExecutorState<D: Deliverable> {
    Running(FuturesMpsc::UnboundedReceiver<ExecutorMessage<D>>),

We should use a bounded channel here. The Sender for a bounded channel includes a try_send method which allows attempting a non-blocking send.

---

src/executor.rs, line 89 at r3 (raw file):

        info!("Spawning Executor.");
        let tls = TlsConnector::builder().and_then(|builder| builder.build()).map_err(SpawnError::HttpsConnector)?;

I find this whole TlsConnector thing pretty strange. We can't establish our own sockets?

How is hyper doing DNS? We must not use the built-in glibc getaddrinfo that it previously used via ToSocketAddrs. This is why we switched to c-ares at some point. We should switch the resolver to TrustDNS resolver if it's not already using that.

I suppose we could actually implement our own TlsConnector and not have to fork Hyper :)

---

src/executor.rs, line 159 at r3 (raw file):

                                        );
                                    },
                                    ExecutorMessage::Shutdown => {

I'm not a big fan of the Shutdown message for terminating an executor. Dropping all of the send handles guarantees no messages arrive after detecting the channel has closed. In the current implementation, it doesn't look like we handle draining the send channel once the executor enters the Draining state.

---

src/lib.rs, line 26 at r3 (raw file):

#[cfg(test)]
mod tests {

Should these be made into integration tests that live in tests/lib.rs?

---

src/lib.rs, line 156 at r3 (raw file):

    }

    const ONE_SIGNAL_IP_ADDRESSES : [&'static str; 5] = [

The addresses you have listed here are actually CloudFlare IP addresses. Rather than have a hardcoded list of IPs, we can use CloudFlare's list of IP ranges to check whether lsof contains anything matching those IPs. I actually wrote some code to do this last night as part of debugging a potential issue on our loadbalancers. This reads from stdin rather than the lsof output, but it should be easy to adapt.

extern crate ipnet;

use std::io::{self, Read, BufRead};
use std::net::IpAddr;

use ipnet::{Contains, IpNet};

static CLOUDFLARE_NETS: &[&str] = &[
    "103.21.244.0/22",
    "103.22.200.0/22",
    "103.31.4.0/22",
    "104.16.0.0/12",
    "108.162.192.0/18",
    "131.0.72.0/22",
    "141.101.64.0/18",
    "162.158.0.0/15",
    "172.64.0.0/13",
    "173.245.48.0/20",
    "188.114.96.0/20",
    "190.93.240.0/20",
    "197.234.240.0/22",
    "198.41.128.0/17",
];

fn run() -> Result<(), Box<::std::error::Error>> {
    let cloudflare_nets = CLOUDFLARE_NETS.iter()
        .map(|net| net.parse::<IpNet>())
        .collect::<Result<Vec<IpNet>, _>>()?;

    let stdin = io::stdin();
    let mut stdin = stdin.lock();

    for line in stdin.lines() {
        let line = line?;
        match line.parse::<IpAddr>() {
            Ok(addr) => {
                let is_cf = cloudflare_nets
                    .iter()
                    .any(|net| net.contains(&addr));

                if !is_cf {
                    println!("{}", line);
                }
            },
            _ => {
                eprintln!("Failed to parse {:?} as ip", line);
            },
        }
    }


    Ok(())
}

fn main() {
    run().unwrap();
}

---

src/pool.rs, line 60 at r3 (raw file):

            match handle.send(transaction) {
                Err(RequestError::Full(transaction)) => {
                    ActResult::ValidWithError(Error::Full(transaction))

Please document the conditions that would lead to these ActResult or RequestError variants. Given that this is a new lib, readers are unlikely to be familiar with it, and this code should attempt to explain what's going on in each of these cases.

---

src/transaction.rs, line 58 at r3 (raw file):

}

struct SpawnedTransaction<D: Deliverable, W: Future, R: Future>

Bounds are not necessary here

---

src/transaction.rs, line 69 at r3 (raw file):

}

impl<D: Deliverable, W: Future, R: Future> Drop for SpawnedTransaction<D, W, R> {

Bounds are not necessary here

---

src/transaction.rs, line 84 at r3 (raw file):

    D: Deliverable,
    W: Future<
        Item=Either<((Response, Vec<u8>), Timeout), ((), R)>,

I believe we can remove the PhantomData since W is only constrained in a where clause. We would need to remove the generic bound of R from the SpawnedTransaction type as well.

---

src/transaction.rs, line 92 at r3 (raw file):

    >,
{
    type Item = ();

It's surprising to me that this is not modeled as a Future<Item=Response+Body, Error=TransactionError>.

---

src/transaction.rs, line 146 at r3 (raw file):

            .map(|deliverable| {
                deliverable.complete(delivery_result);
                self.task.notify();

What is the point of notifying the task here? Please add a comment. Maybe also a doc comment where the task field is defined.

---

src/transaction.rs, line 164 at r3 (raw file):

    }

    pub(crate) fn spawn_request(self, client: &Client<HttpsConnector<HttpConnector>>, handle: &Handle, timeout: Duration, counter: Counter) {

Break long line

---

src/transaction.rs, line 172 at r3 (raw file):

            .and_then(|response| {
                let status = response.status();
                let headers = response.headers().clone();

Is there any way to avoid the header clone?

---

src/transaction.rs, line 186 at r3 (raw file):

        match Timeout::new(timeout, handle) {
            Err(error) => {
                deliverable.complete(DeliveryResult::TimeoutError {

Perhaps TimeoutError should be rolled into a SpawnRequestError?

---

src/transaction.rs, line 208 at r3 (raw file):

#[cfg(test)]
mod tests {

Please add a test for transaction timeouts. This could be achieved by starting a local server which doesn't send a response.

---

Comments from Reviewable

[jwilm]

Reviewed 4 of 5 files at r4.
Review status: all files reviewed at latest revision, 21 unresolved discussions.

---

src/executor.rs, line 17 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Reviewable says counter module was removed in latest revision...

You pushed a fix for this during my review :P

---

src/lib.rs, line 156 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

The addresses you have listed here are actually CloudFlare IP addresses. Rather than have a hardcoded list of IPs, we can use CloudFlare's list of IP ranges to check whether lsof contains anything matching those IPs. I actually wrote some code to do this last night as part of debugging a potential issue on our loadbalancers. This reads from stdin rather than the lsof output, but it should be easy to adapt.

extern crate ipnet;

use std::io::{self, Read, BufRead};
use std::net::IpAddr;

use ipnet::{Contains, IpNet};

static CLOUDFLARE_NETS: &[&str] = &[
    "103.21.244.0/22",
    "103.22.200.0/22",
    "103.31.4.0/22",
    "104.16.0.0/12",
    "108.162.192.0/18",
    "131.0.72.0/22",
    "141.101.64.0/18",
    "162.158.0.0/15",
    "172.64.0.0/13",
    "173.245.48.0/20",
    "188.114.96.0/20",
    "190.93.240.0/20",
    "197.234.240.0/22",
    "198.41.128.0/17",
];

fn run() -> Result<(), Box<::std::error::Error>> {
    let cloudflare_nets = CLOUDFLARE_NETS.iter()
        .map(|net| net.parse::<IpNet>())
        .collect::<Result<Vec<IpNet>, _>>()?;

    let stdin = io::stdin();
    let mut stdin = stdin.lock();

    for line in stdin.lines() {
        let line = line?;
        match line.parse::<IpAddr>() {
            Ok(addr) => {
                let is_cf = cloudflare_nets
                    .iter()
                    .any(|net| net.contains(&addr));

                if !is_cf {
                    println!("{}", line);
                }
            },
            _ => {
                eprintln!("Failed to parse {:?} as ip", line);
            },
        }
    }


    Ok(())
}

fn main() {
    run().unwrap();
}

Looks like you just added IPv6 addrs as well. The list of all addresses can be found at https://www.cloudflare.com/ips/

---

Comments from Reviewable

[jwilm]

Reviewed 2 of 2 files at r5.
Review status: all files reviewed at latest revision, 21 unresolved discussions, some commit checks failed.

---

Cargo.toml, line 23 at r5 (raw file):

[replace]
"tokio-core:0.1.17" = { git = "https://github.com/DarrenTsung/tokio-core", branch = "connection-reuse" }

What is this about?

---

Comments from Reviewable

[DarrenTsung]

Review status: 7 of 9 files reviewed at latest revision, 22 unresolved discussions.

---

Cargo.toml, line 23 at r5 (raw file):

Previously, jwilm (Joe Wilm) wrote…

What is this about?

You can ignore this, I was investigating connection-reuse not seeming to work on the test server, but actually realized that it was because the loader binary was adding new connections everytime. Oops :P.

---

src/executor.rs, line 89 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

I find this whole TlsConnector thing pretty strange. We can't establish our own sockets?

How is hyper doing DNS? We must not use the built-in glibc getaddrinfo that it previously used via ToSocketAddrs. This is why we switched to c-ares at some point. We should switch the resolver to TrustDNS resolver if it's not already using that.

I suppose we could actually implement our own TlsConnector and not have to fork Hyper :)

Ah okay, I followed the path down tokio-tls and it does resolve to net::ToSocketAddrs implementation. I will take a look at implementing our own TlsConnector with a

---

Comments from Reviewable

[DarrenTsung]

Review status: 7 of 9 files reviewed at latest revision, 22 unresolved discussions.

---

src/executor.rs, line 89 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

Ah okay, I followed the path down tokio-tls and it does resolve to net::ToSocketAddrs implementation. I will take a look at implementing our own TlsConnector with a

Oops, I didn't realize I published this. I opened an issue on hyper to see what their input is: hyperium/hyper#1517

---

Comments from Reviewable

[DarrenTsung]

Review status: 1 of 9 files reviewed at latest revision, 22 unresolved discussions.

---

src/config.rs, line 18 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Does hyper 0.11 still handle DNS? Seems like we should be able to plugin a tokio/futures-compatible resolver library and eliminate this option.

Yeah, hyper still handles DNS, see other comment. Will leave this open so we can remove this option.

---

src/error.rs, line 9 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

The variants here need some explanation in doc comments. It would also be nice to expose a library Error type that buckets these into logical categories or something that is suggestive about how a consumer of this API should handle the errors.

Refactored the error handling and added doc comments!

There are two types of errors that I wanted to expose because only a subset of errors can happen during new(), and I do think that logically you could group error handling between critical errors during Spawn vs the pool being full.

---

src/error.rs, line 12 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Full? What does that mean?

Renamed to PoolFull and added a doc comment. It was previously named Full, but I agree that it's not exactly clear.

---

src/error.rs, line 33 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

This doesn't seem unreachable

Removed with error refactor

---

src/error.rs, line 43 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Nor does this

Removed with error refactor

---

src/executor.rs, line 17 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

You pushed a fix for this during my review :P

:P

---

src/executor.rs, line 40 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

We should use a bounded channel here. The Sender for a bounded channel includes a try_send method which allows attempting a non-blocking send.

UnboundedSender shouldn't block though, right? I mean, it can't block waiting for room as it's unbounded, so it should always be placing a message in the queue.

We handle backpressure over the entire system with the transaction_counter as well.

---

src/executor.rs, line 159 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

I'm not a big fan of the Shutdown message for terminating an executor. Dropping all of the send handles guarantees no messages arrive after detecting the channel has closed. In the current implementation, it doesn't look like we handle draining the send channel once the executor enters the Draining state.

Yeah I don't know why I didn't use the pattern of dropping the handle as well. I put a note down that sending a shutdown would drop the handle, but I guess that doesn't account for users that cloned the Handle (if that was possible).

I switched it over to dropping the handle. Thanks!

---

src/lib.rs, line 26 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Should these be made into integration tests that live in tests/lib.rs?

Good point, moved over!

---

src/lib.rs, line 156 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Looks like you just added IPv6 addrs as well. The list of all addresses can be found at https://www.cloudflare.com/ips/

Implemented this, thanks for the reference!

---

src/pool.rs, line 60 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Please document the conditions that would lead to these ActResult or RequestError variants. Given that this is a new lib, readers are unlikely to be familiar with it, and this code should attempt to explain what's going on in each of these cases.

Refactored it with new fpool implementation, there are comments when we invalidate the thread because it failed to send now

---

src/transaction.rs, line 58 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Bounds are not necessary here

Removed R, thanks!

---

src/transaction.rs, line 69 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Bounds are not necessary here

Removed R

---

src/transaction.rs, line 84 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

I believe we can remove the PhantomData since W is only constrained in a where clause. We would need to remove the generic bound of R from the SpawnedTransaction type as well.

Removed R + PhantomData, thanks!

---

src/transaction.rs, line 92 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

It's surprising to me that this is not modeled as a Future<Item=Response+Body, Error=TransactionError>.

That would work - as well as an and_then() that reports the result to the dispatcher(?), right? But I'm not sure how that would work with the Drop handling since we need to add the impl Drop for some type.

---

src/transaction.rs, line 146 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

What is the point of notifying the task here? Please add a comment. Maybe also a doc comment where the task field is defined.

Added comments + a doc comment. Basically the executor is waiting for the transactions to finish, but is never woken up if not notified by transactions finishing.

---

src/transaction.rs, line 164 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Break long line

Done!

---

src/transaction.rs, line 172 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Is there any way to avoid the header clone?

Unfortunately not if we want to return the Response as part of the DeliveryResult.

Spoke to seanmonstar about this and possibly better handling in hyper 0.12.

---

src/transaction.rs, line 186 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Perhaps TimeoutError should be rolled into a SpawnRequestError?

I think the naming of the errors might be confusing, there's:
SpawnError -> returned if constructing a new client for the Pool fails
Error -> contains SpawnError + PoolFull, all error cases when sending a new request.

Then there's also the types in DeliveryResult, this is returning an TimeoutError which is when the Timeout type errored. There's also DeliveryResult::Timeout which is when the connection timed out.

---

src/transaction.rs, line 208 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

Please add a test for transaction timeouts. This could be achieved by starting a local server which doesn't send a response.

I have a test for timeout in the integration tests, see timeout_works_as_expected()

---

Comments from Reviewable

[DarrenTsung]

Review status: 1 of 10 files reviewed at latest revision, 17 unresolved discussions.

---

src/config.rs, line 18 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

Yeah, hyper still handles DNS, see other comment. Will leave this open so we can remove this option.

Switched to a custom HttpConnector that uses c-ares DNS resolver. I tried with trust-dns first, but load testing revealed panics in the resolution flow.

---

src/executor.rs, line 89 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

Oops, I didn't realize I published this. I opened an issue on hyper to see what their input is: hyperium/hyper#1517

Switched to a custom HttpConnector that uses c-ares DNS resolver. I tried with trust-dns first, but load testing revealed panics in the resolution flow.

---

Comments from Reviewable

[jwilm]

Reviewed 1 of 8 files at r6, 4 of 8 files at r7, 6 of 6 files at r8.
Review status: 9 of 10 files reviewed at latest revision, 6 unresolved discussions.

---

src/config.rs, line 18 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

Switched to a custom HttpConnector that uses c-ares DNS resolver. I tried with trust-dns first, but load testing revealed panics in the resolution flow.

We should work with the TrustDNS project to resolve whatever issues we are having.

---

src/error.rs, line 42 at r7 (raw file):

    }

    pub fn into_transaction(self) -> Transaction<D> {

In the standard library, the naming convention for this type of method is into_inner.

---

src/executor.rs, line 40 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

UnboundedSender shouldn't block though, right? I mean, it can't block waiting for room as it's unbounded, so it should always be placing a message in the queue.

We handle backpressure over the entire system with the transaction_counter as well.

It's more about knowing worst-case how much memory this channel will use.

---

src/executor.rs, line 65 at r8 (raw file):

    pub(crate) fn shutdown(self) -> JoinHandle<()> {
        // We explicitly drop the sender here because that is how we indicate to the

We should prefer a doc comment rather than having unnecessary lines of code.

---

src/pool.rs, line 67 at r8 (raw file):

                    Err(RequestError::PoolFull(transaction)) => transaction,
                    Err(RequestError::FailedSend(transaction)) => {
                        // invalidate the thread as it didn't send

It's a little surprising that the consumer is responsible for this. Maybe we could use an intermediate abstraction on the fpool?

---

src/transaction.rs, line 92 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

That would work - as well as an and_then() that reports the result to the dispatcher(?), right? But I'm not sure how that would work with the Drop handling since we need to add the impl Drop for some type.

How would that change where Drop is implemented?

---

src/transaction.rs, line 146 at r3 (raw file):

Previously, DarrenTsung (Darren Tsung) wrote…

Added comments + a doc comment. Basically the executor is waiting for the transactions to finish, but is never woken up if not notified by transactions finishing.

Ah, I get it. Sounds good!

---

Comments from Reviewable

[DarrenTsung]

Review status: 9 of 10 files reviewed at latest revision, 7 unresolved discussions.

---

src/config.rs, line 18 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

We should work with the TrustDNS project to resolve whatever issues we are having.

Figured out that trying to run a ResolverFuture with tokio-core was causing the issue, moved over to trust-dns!

---

src/error.rs, line 42 at r7 (raw file):

Previously, jwilm (Joe Wilm) wrote…

In the standard library, the naming convention for this type of method is into_inner.

Done

---

src/executor.rs, line 40 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

It's more about knowing worst-case how much memory this channel will use.

Right, good point. Actually in this case this channel is completely bounded by max_transactions as the Counter is cloned and sent as part of the message.

Along with that, the executor thread would immediately spawn the transaction as part of consuming it while it has the receiver. No chance it would be backlogged unless the receiver is dropped / thread doesn't get woken up, it just seems weird to add a failure case that shouldn't happen (given the in-variants of tokio / the channel type).

---

src/executor.rs, line 65 at r8 (raw file):

Previously, jwilm (Joe Wilm) wrote…

We should prefer a doc comment rather than having unnecessary lines of code.

Done!

---

src/pool.rs, line 67 at r8 (raw file):

Previously, jwilm (Joe Wilm) wrote…

It's a little surprising that the consumer is responsible for this. Maybe we could use an intermediate abstraction on the fpool?

fpool doesn't know what the objects are being used for, which is why an outer user must invalidate the object in an error case.

I guess we could wrap the fpool in some sort of type that handles it, but I'm not sure what this type would be left with?

---

src/transaction.rs, line 92 at r3 (raw file):

Previously, jwilm (Joe Wilm) wrote…

How would that change where Drop is implemented?

Because SpawnedTransaction is the type that ensures that the transaction result is always sent to the deliverable, the .work field in SpawnedTransaction is the type you're expecting (I think).

The transaction future that returns an Item=Response+Body, Error=Timeout/TransationError is the chained, generalized type W.

Basically we have to have some type that has Item=(), Error=() because it needs to also impl Drop and notify the deliverable in that case. Even if we had a nice future type which returned Response+Body it would have to be wrapped in a struct that implements Drop+ has ownership of the Deliverable. Because that struct has ownership of Deliverable, it would also need to be the future that sends the real result to Deliverable as well.

---

Comments from Reviewable

[jwilm]

---

Reviewed 1 of 1 files at r9, 2 of 2 files at r10.
Review status: all files reviewed at latest revision, 3 unresolved discussions.

---

Comments from Reviewable