tikv_util: introduce future channel (tikv#13407)

ref tikv#12842

It's a super set of batch channel. It can be used as a batch channel
or just a future channel.

This is the first PR to use unified thread pool in apply system for v2.

Signed-off-by: Jay Lee <BusyJayLee@gmail.com>

Co-authored-by: Ti Chi Robot <ti-community-prow-bot@tidb.io>

components/tikv_util/Cargo.toml

@@ -73,3 +73,8 @@ regex = "1.0"
 tempfile = "3.0"
 toml = "0.5"
 utime = "0.2"
+
+[[bench]]
+name = "channel"
+path = "benches/channel/mod.rs"
+test = true

tests/benches/channel/bench_channel.rs → components/tikv_util/benches/channel/bench_channel.rs

@@ -113,8 +113,8 @@ fn bench_crossbeam_channel(b: &mut Bencher) {
 }
 
 #[bench]
-fn bench_receiver_stream_batch(b: &mut Bencher) {
-    let (tx, rx) = mpsc::batch::bounded::<i32>(128, 8);
+fn bench_receiver_stream_unbounded_batch(b: &mut Bencher) {
+    let (tx, rx) = mpsc::future::unbounded::<i32>(mpsc::future::WakePolicy::TillReach(8));
     for _ in 0..1 {
         let tx1 = tx.clone();
         thread::spawn(move || {
@@ -124,12 +124,9 @@ fn bench_receiver_stream_batch(b: &mut Bencher) {
         });
     }
 
-    let mut rx = Some(mpsc::batch::BatchReceiver::new(
-        rx,
-        32,
-        Vec::new,
-        mpsc::batch::VecCollector,
-    ));
+    let rx = mpsc::future::BatchReceiver::new(rx, 32, Vec::new, Vec::push);
+
+    let mut rx = Some(block_on(rx.into_future()).1);
 
     b.iter(|| {
         let mut count = 0;
@@ -150,8 +147,8 @@ fn bench_receiver_stream_batch(b: &mut Bencher) {
 }
 
 #[bench]
-fn bench_receiver_stream(b: &mut Bencher) {
-    let (tx, rx) = mpsc::batch::bounded::<i32>(128, 1);
+fn bench_receiver_stream_unbounded_nobatch(b: &mut Bencher) {
+    let (tx, rx) = mpsc::future::unbounded::<i32>(mpsc::future::WakePolicy::Immediately);
     for _ in 0..1 {
         let tx1 = tx.clone();
         thread::spawn(move || {
@@ -161,7 +158,7 @@ fn bench_receiver_stream(b: &mut Bencher) {
         });
     }
 
-    let mut rx = Some(rx);
+    let mut rx = Some(block_on(rx.into_future()).1);
     b.iter(|| {
         let mut count = 0;
         let mut rx1 = rx.take().unwrap();