Refactor buf_channel

[allada]

Major rewrite of how DropCloserReadHalf and DropCloserWriteHalf.

• Simplified DropCloserReadHalf::consume()
• DropCloserWriteHalf will no longer wait for eof to be received
• Added utility func DropCloserWriteHalf::bind()
• DropCloserReadHalf can now be configured to hold a queue of recent data to aid with places that might need retry.
• EOF signal is now an AtomicBool instead of a tokio::once channel.
• DropCloserReadHalf::try_reset_stream added to put queue back onto head of stream if able.

---

This change is 

[allada]

+@adam-singer

Reviewable status: 0 of 1 LGTMs obtained, and pending CI: Analyze (javascript-typescript), Analyze (python), Bazel Dev / ubuntu-22.04, Cargo Dev / macos-13, Cargo Dev / ubuntu-22.04, Local / ubuntu-22.04, Publish image, Publish nativelink-worker-lre-cc, Remote / large-ubuntu-22.04, asan / ubuntu-22.04, docker-compose-compiles-nativelink (20.04), docker-compose-compiles-nativelink (22.04), integration-tests (20.04), integration-tests (22.04), macos-13, pre-commit-checks, ubuntu-20.04 / stable, ubuntu-22.04, ubuntu-22.04 / stable, vale, windows-2022 / stable, zig-cc ubuntu-20.04, zig-cc ubuntu-22.04 (waiting on @adam-singer)

[adam-singer]

Reviewed 13 of 16 files at r1, 1 of 2 files at r2, all commit messages.
Reviewable status: 0 of 1 LGTMs obtained

[allada]

Reviewable status: 0 of 1 LGTMs obtained (waiting on @adam-singer)

---

nativelink-store/tests/filesystem_store_test.rs line 231 at r1 (raw file):

    const HASH1: &str = "0123456789abcdef000000000000000000010000000000000123456789abcdef";
    const HASH2: &str = "0123456789abcdef000000000000000000020000000000000123456789abcdef";
    const VALUE1: &str = "1234";

fyi: The buffer queue got slightly increased, so I needed slightly more data.

[adam-singer]

Reviewed 1 of 16 files at r1, 1 of 2 files at r2.
Reviewable status: 0 of 1 LGTMs obtained, and 3 discussions need to be resolved

---

nativelink-util/src/buf_channel.rs line 67 at r2 (raw file):

impl DropCloserWriteHalf {
    /// Sends data over the channel to the receiver.
    pub fn send(&mut self, buf: Bytes) -> impl Future<Output = Result<(), Error>> + '_ {

Whats the reasoning or errors that made switching from async keyword to full type signature of impl Future<..> + '_?

---

nativelink-util/src/buf_channel.rs line 145 at r2 (raw file):

        // Now close our stream.
        self.tx = None;

Should drop be called on the contained type of tx, self.tx.map(|tx| drop(tx)); before applying None?

---

nativelink-util/src/buf_channel.rs line 223 at r2 (raw file):

                    ));
                }
                const ZERO_DATA: Bytes = Bytes::new();

Can this value be hoisted to global to this module?

[allada]

Reviewable status: 0 of 1 LGTMs obtained, and 3 discussions need to be resolved (waiting on @adam-singer)

---

nativelink-util/src/buf_channel.rs line 67 at r2 (raw file):

Previously, adam-singer (Adam Singer) wrote…

Whats the reasoning or errors that made switching from async keyword to full type signature of impl Future<..> + '_?

It's probably me over thinking things, but in theory it makes inlining easier/more-possible for the compiler.

When it's async fn it forces the user to .await another future, but if you are just forwarding on the future, you could let the compiler infer it using impl Future<>.

The compiler can probably inline some of it, but I've learnt with rust to not trust rust at inlining async functions because it needs to track stack traces on panic (or something). When we forward it, in theory there is no stack trace, so it's pure inlinable.

---

nativelink-util/src/buf_channel.rs line 145 at r2 (raw file):

Previously, adam-singer (Adam Singer) wrote…

Should drop be called on the contained type of tx, self.tx.map(|tx| drop(tx)); before applying None?

By spec (RAII), drop must be called when it's overridden.

---

nativelink-util/src/buf_channel.rs line 223 at r2 (raw file):

Previously, adam-singer (Adam Singer) wrote…

Can this value be hoisted to global to this module?

Done.

[adam-singer]

Reviewed 2 of 2 files at r3, all commit messages.
Reviewable status: complete! 1 of 1 LGTMs obtained