write batch with capacity (#1846)

benches/writebatch/bench_writebatch.rs

@@ -91,3 +91,29 @@ fn bench_writebatch_512(b: &mut Bencher) {
 fn bench_writebatch_1024(b: &mut Bencher) {
     bench_writebatch_impl(b, 1024);
 }
+
+fn fill_writebatch(wb: &WriteBatch, target_size: usize) {
+    let (k, v) = (b"this is the key", b"this is the value");
+    loop {
+        wb.put(k, v).unwrap();
+        if wb.data_size() >= target_size {
+            break;
+        }
+    }
+}
+
+#[bench]
+fn bench_writebatch_without_capacity(b: &mut Bencher) {
+    b.iter(|| {
+        let wb = WriteBatch::new();
+        fill_writebatch(&wb, 4096);
+    });
+}
+
+#[bench]
+fn bench_writebatch_with_capacity(b: &mut Bencher) {
+    b.iter(|| {
+        let wb = WriteBatch::with_capacity(4096);
+        fill_writebatch(&wb, 4096);
+    });
+}

src/raftstore/store/peer.rs

@@ -55,6 +55,7 @@ use super::metrics::*;
 use super::local_metrics::{RaftReadyMetrics, RaftMessageMetrics, RaftProposeMetrics, RaftMetrics};
 
 const TRANSFER_LEADER_ALLOW_LOG_LAG: u64 = 10;
+const DEFAULT_APPEND_WB_SIZE: usize = 4 * 1024;
 
 struct ReadIndexRequest {
     uuid: Uuid,
@@ -146,7 +147,7 @@ pub struct ReadyContext<'a, T: 'a> {
 impl<'a, T> ReadyContext<'a, T> {
     pub fn new(metrics: &'a mut RaftMetrics, t: &'a T, cap: usize) -> ReadyContext<'a, T> {
         ReadyContext {
-            wb: WriteBatch::new(),
+            wb: WriteBatch::with_capacity(DEFAULT_APPEND_WB_SIZE),
             metrics: metrics,
             trans: t,
             ready_res: Vec::with_capacity(cap),

src/raftstore/store/worker/apply.rs

@@ -41,6 +41,7 @@ use raftstore::store::peer::{parse_data_at, check_epoch, Peer};
 use raftstore::store::metrics::*;
 
 const WRITE_BATCH_MAX_KEYS: usize = 128;
+const DEFAULT_APPLY_WB_SIZE: usize = 4 * 1024;
 
 pub struct PendingCmd {
     pub uuid: Uuid,
@@ -148,7 +149,7 @@ impl<'a> ApplyContext<'a> {
     fn new(host: &CoprocessorHost) -> ApplyContext {
         ApplyContext {
             host: host,
-            wb: Some(WriteBatch::new()),
+            wb: Some(WriteBatch::with_capacity(DEFAULT_APPLY_WB_SIZE)),
             cbs: vec![],
             wb_last_bytes: 0,
             wb_last_keys: 0,
@@ -376,7 +377,7 @@ impl ApplyDelegate {
                 for (cb, resp) in apply_ctx.cbs.drain(..) {
                     cb(resp);
                 }
-                apply_ctx.wb = Some(WriteBatch::new());
+                apply_ctx.wb = Some(WriteBatch::with_capacity(DEFAULT_APPLY_WB_SIZE));
                 apply_ctx.mark_last_bytes_and_keys();
             }