feature: redis large payload support (#147)

- bugfix: wait for writable event in Established state before adding to ready queue
- bugfix: in edge-case buffer may revert to default 4KB size
- bugfix: requeue connection from front when there is no work to maintain order
- feature: add redis RESP serialization, required when size >64KB, make default for redis

.travis.yml

@@ -69,10 +69,17 @@ matrix:
         - sudo chown -R travis $HOME/.cargo
     - os: linux
       rust: nightly
-      env: TYPE=fuzzer_codec_redis RUST_BACKTRACE=1
+      env: TYPE=fuzzer_codec_redis_inline RUST_BACKTRACE=1
       script:
         - cargo install cargo-fuzz -f
-        - sudo -E $HOME/.cargo/bin/cargo fuzz run -s address fuzzer_codec_redis -- -max_total_time=300
+        - sudo -E $HOME/.cargo/bin/cargo fuzz run -s address fuzzer_codec_redis_inline -- -max_total_time=300
+        - sudo chown -R travis $HOME/.cargo
+    - os: linux
+      rust: nightly
+      env: TYPE=fuzzer_codec_redis_resp RUST_BACKTRACE=1
+      script:
+        - cargo install cargo-fuzz -f
+        - sudo -E $HOME/.cargo/bin/cargo fuzz run -s address fuzzer_codec_redis_resp -- -max_total_time=300
         - sudo chown -R travis $HOME/.cargo
     - os: linux
       rust: nightly

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rpc-perf"
-version = "2.0.4-pre"
+version = "2.1.0-pre"
 authors = ["Brian Martin <bmartin@twitter.com>"]
 
 license = "Apache-2.0"
@@ -41,22 +41,22 @@ byteorder = "=1.0.0"
 bytes = "=0.3.0"
 crc = "=1.2.0"
 getopts = "=0.2.14"
-log = "=0.3.7"
+log = "0.3.8"
 log-panics = "=1.1.0"
-mio = "=0.6.7"
+mio = "=0.6.8"
 mpmc = "=0.1.2"
 pad = "=0.1.4"
 rand = "=0.3.15"
 ratelimit = "=0.3.1"
 shuteye = "=0.2.0"
 slab = "=0.3.0"
 simple_logger = "=0.4.0"
-tic = "=0.2.0"
+tic = "=0.2.2"
 time = "=0.1.37"
 toml = "=0.2.1"
 
 [features]
 asm = [ "tic/asm" ]
 default = []
 unstable = []
-benchcmp = []
+benchcmp = []

fuzz/Cargo.toml

@@ -71,8 +71,12 @@ name = "fuzzer_codec_ping"
 path = "fuzzers/fuzzer_codec_ping.rs"
 
 [[bin]]
-name = "fuzzer_codec_redis"
-path = "fuzzers/fuzzer_codec_redis.rs"
+name = "fuzzer_codec_redis_inline"
+path = "fuzzers/fuzzer_codec_redis_inline.rs"
+
+[[bin]]
+name = "fuzzer_codec_redis_resp"
+path = "fuzzers/fuzzer_codec_redis_resp.rs"
 
 [[bin]]
 name = "fuzzer_codec_thrift"

fuzz/fuzzers/fuzzer_codec_redis_inline.rs

@@ -0,0 +1,35 @@
+#![no_main]
+#[macro_use]
+extern crate libfuzzer_sys;
+#[macro_use]
+extern crate log;
+extern crate bytes;
+extern crate byteorder;
+extern crate crc;
+extern crate getopts;
+extern crate mio;
+extern crate mpmc;
+extern crate pad;
+extern crate time;
+extern crate rand;
+extern crate ratelimit;
+extern crate slab;
+extern crate tic;
+extern crate toml;
+
+use std::str;
+
+#[path = "../../src/cfgtypes/mod.rs"]
+mod cfgtypes;
+
+#[path = "../../src/codec/mod.rs"]
+mod codec;
+
+use cfgtypes::*;
+use codec::redis_inline::RedisParse;
+
+//ProtocolParseFactory
+fuzz_target!(|data: &[u8]| {
+                 let parser = RedisParse;
+                 let _ = parser.parse(data);
+             });

fuzz/fuzzers/fuzzer_codec_redis.rs → fuzz/fuzzers/fuzzer_codec_redis_resp.rs

@@ -26,7 +26,7 @@ mod cfgtypes;
 mod codec;
 
 use cfgtypes::*;
-use codec::redis::RedisParse;
+use codec::redis_resp::RedisParse;
 
 //ProtocolParseFactory
 fuzz_target!(|data: &[u8]| {

src/client/buffer.rs

@@ -0,0 +1,49 @@
+//  rpc-perf - RPC Performance Testing
+//  Copyright 2017 Twitter, Inc
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use bytes::{ByteBuf, MutByteBuf};
+
+#[derive(Debug)]
+pub struct Buffer {
+    rx_bytes: usize,
+    tx_bytes: usize,
+    pub rx: Option<MutByteBuf>,
+    pub tx: Option<MutByteBuf>,
+}
+
+impl Buffer {
+    pub fn new(rx: usize, tx: usize) -> Buffer {
+        Buffer {
+            rx_bytes: rx,
+            tx_bytes: tx,
+            rx: Some(ByteBuf::mut_with_capacity(rx)),
+            tx: Some(ByteBuf::mut_with_capacity(tx)),
+        }
+    }
+
+    pub fn clear(&mut self) {
+        let mut rx = self.rx.take().unwrap_or_else(
+            || ByteBuf::mut_with_capacity(self.rx_bytes),
+        );
+        rx.clear();
+        self.rx = Some(rx);
+
+        let mut tx = self.tx.take().unwrap_or_else(
+            || ByteBuf::mut_with_capacity(self.tx_bytes),
+        );
+        tx.clear();
+        self.tx = Some(tx);
+    }
+}