Futures-rs: mpsc Channel and custom retry logic make for never-closing channel

[xrl]

I am writing a AWS Kinesis client using the Rusoto AWS SDK and I have hit an issue where I cannot retry requests without cloning the futures::sync::mpsc::channel Sender and still have the right kind of drop semantics for closing the Receiver.

The code spawns a thread to handle receiving records, it batches them up, and will create inflight number of concurrent requests. This is great for throughput as Kinesis has relatively high latency (50ms) and small request size (500 records). These requests can fail for a number of reasons: the TCP connection sees an error, the kinesis stream is under-provisioned for the current message rate (bursts, etc), Kinesis has errors, etc. I need to simply retry requests when they fail.

My first stab at retrying requests has me cloning the Sender and moving that handle in to the Receiver thread. Now the issue is that when the original Sender is dropped the receiver does not automatically close -- my Receiver thread is now open forever and I am leaking threads in my code.

So my issue, is there a way to create a Sender which does NOT count for the life of the channel? Or is there a way to close() a Sender which includes closing all the other Senders? Either one of these solutions may yield data loss for requests which need to be retried. But perhaps it would be useful? Or, perhaps there is a better way to structure this retry logic? Could I do it in the AndThen to keep that initial request going?

Here is code which spawns the Receiver thread and attempts retries:

pub fn kinesis_tx(stream_name: String, chan_buf: usize, inflight: usize) -> RecordsChannel {
    info!("CREATING kinesis channel");
    use futures::sync::mpsc::channel;
    use futures::{Future, Sink, Stream};

    let (tx, rx) = channel(chan_buf);
    let client = Arc::new(KinesisClient::simple(Region::UsWest2));

    let mut retry_tx = tx.clone().wait();
    std::thread::spawn(move || {
        info!("STARTING kinesis batch put thread");
        let puts = rx.chunks(500)
            .map(|batch: Vec<PutRecordsRequestEntry>| {
                let input = PutRecordsInput {
                    records: batch,
                    stream_name: stream_name.clone(),
                };
                let chain = client.put_records(&input).then(|put_res| {
                    let retval: Result<
                        Result<PutRecordsOutput, (PutRecordsError, PutRecordsInput)>,
                        (),
                    > = match put_res {
                        Ok(res) => {
                            trace!("match put_res: it worked");
                            Ok(Ok(res))
                        }
                        Err(err) => {
                            trace!("match put_res: failed");
                            Ok(Err((err, input)))
                        }
                    };
                    retval
                });
                chain
            })
            .buffer_unordered(inflight);

        for put_res in puts.wait() {
            match put_res {
                Ok(Ok(put)) => {
                    if let Some(failed) = put.failed_record_count {
                        if failed > 0 {
                            error!("{} record(s) failed to commit to kinesis", failed);
                            let put: PutRecordsOutput = put;
                            for rec in put.records {
                                if rec.error_code.is_some() {
                                    error!("failed record: {:?}", rec);
                                }
                            }
                        }
                    }
                }
                Ok(Err((put_records_err, put_records_input))) => match put_records_err {
                    PutRecordsError::HttpDispatch(dispatch_err) => {
                        error!(
                            "http dispatch error: {:?}. retrying records...",
                            dispatch_err
                        );
                        for record in put_records_input.records {
                            match retry_tx.send(record) {
                                Ok(()) => debug!("Wait#send succeeded"),
                                Err(e) => error!("Wait#send error {:?}", e),
                            }
                        }
                    }
                    PutRecordsError::Unknown(raw_message) => {
                        error!("unknown error: '{:?}', retrying records...", raw_message);
                        for record in put_records_input.records {
                            match retry_tx.send(record) {
                                Ok(()) => debug!("Wait#send succeeded"),
                                Err(e) => error!("Wait#send error {:?}", e),
                            }
                        }
                    }
                    other => {
                        error!("unhandled kinesis error: {:?}", other);
                    }
                },
                other => {
                    error!("puts.wait() fallthrough: {:?}", other);
                }
            }
        }
        info!("STOPPING kinesis batch put thread");
    });

    return tx;
}

[seanmonstar]

I have also encountered this situation, and think the there should be a close method on senders that closes the channel for all, no matter how many clones there may be.

Since this does not exist, I've resorted to including an additional cancelable token future that can be closed from any sender, and a wrapped receiver checks this token first on polls.

[xrl]

@seanmonstar can you point to an example of how you do the cancelable token future?

[seanmonstar]

@xrl it's slightly more than a channel, since I conflated a callback concept directly into it, but here: https://github.com/hyperium/hyper/blob/v0.11.18/src/client/dispatch.rs

The cancelable token is in the same folder, cancel.rs.

However, the downside to this is that all the information is actually already inside the mpsc channel. I have to allocate a secondary slot to hold this canceled state, and do additional atomic operations on each poll to check it. It'd be more efficient to just allow the mpsc channel to do this directly.