Auto merge of #68 - antrik:optimise-buffers, r=pcwalton

Optimisations (and cleanups) of Linux backend

This is a bunch of optimisations to the Linux platform code, along with various cleanups of the related code which the optimisation patches build upon.

Most of the cleanups are on the `send()` side, as the `recv()` side is less affected by the optimisation changes, and thus there has been less reason to refactor the code -- some extra cleanup work would probably be in order here.

The optimisations are mostly about avoiding unnecessary copies by using scatter-gather buffers for send and receive; as well as avoiding unnecessary initialisation of receive buffers.

The results are impressive: with gains of at least 5x for large transfers of several MiB (a bit more on a modern system); >5x (on an old system) up to >10x (on a modern one) for small transfers of up to a few KiB; and more than 10x for most of the range in between -- peaking at about 12x - 13x on the old system and 20x - 21x on the modern system for medium-sized transfers of about 64 KiB up to a few hundred KiB.

For another interesting data point, the CPU usage during benchmark runs (with many iterations, to amortise the setup time) was dominated by user time (more than two thirds of total time) with the original variant; whereas the optimised variant not only further reduces system time to less then half the original value (presumably because of fewer allocations?), but also almost entirely eliminates the user time, making it pretty insignificant in the total picture now -- as it should be.

On a less scientific note, Servo built with the optimised ipc-channel doesn't seem to show undue delays any more while rendering a language selection menu. (Which requires lots of fonts to be loaded, and thus triggers heavy ipc-channel activity.)

Author: bors-servo

Cargo.toml

@@ -8,10 +8,12 @@ path = "lib.rs"
 
 [dependencies]
 bincode = ">=0.4.1, <0.6"
-byteorder = "0.5"
 lazy_static = "0.2"
 libc = "0.2"
 rand = "0.3"
 serde = ">=0.6, <0.8"
 serde_macros = ">=0.6, <0.8"
 uuid = { version = "0.2", features = ["v4"] }
+
+[dev-dependencies]
+crossbeam = "0.2"

benches/bench.rs

@@ -0,0 +1,163 @@
+#![feature(test)]
+
+extern crate crossbeam;
+extern crate ipc_channel;
+extern crate test;
+
+use ipc_channel::platform;
+
+use std::sync::{mpsc, Mutex};
+
+/// Allows doing multiple inner iterations per bench.iter() run.
+///
+/// This is mostly to amortise the overhead of spawning a thread in the benchmark
+/// when sending larger messages (that might be fragmented).
+///
+/// Note that you need to compensate the displayed results
+/// for the proportionally longer runs yourself,
+/// as the benchmark framework doesn't know about the inner iterations...
+const ITERATIONS: usize = 1;
+
+fn bench_size(b: &mut test::Bencher, size: usize) {
+    let data: Vec<u8> = (0..size).map(|i| (i % 251) as u8).collect();
+    let (tx, rx) = platform::channel().unwrap();
+
+    let (wait_tx, wait_rx) = mpsc::channel();
+    let wait_rx = Mutex::new(wait_rx);
+
+    if size > platform::OsIpcSender::get_max_fragment_size() {
+        b.iter(|| {
+            crossbeam::scope(|scope| {
+                scope.spawn(|| {
+                    let wait_rx = wait_rx.lock().unwrap();
+                    for _ in 0..ITERATIONS {
+                        tx.send(&data, vec![], vec![]).unwrap();
+                        if ITERATIONS > 1 {
+                            // Prevent beginning of the next send
+                            // from overlapping with receive of last fragment,
+                            // as otherwise results of runs with a large tail fragment
+                            // are significantly skewed.
+                            wait_rx.recv().unwrap();
+                        }
+                    }
+                });
+                for _ in 0..ITERATIONS {
+                    rx.recv().unwrap();
+                    if ITERATIONS > 1 {
+                        wait_tx.send(()).unwrap();
+                    }
+                }
+                // For reasons mysterious to me,
+                // not returning a value *from every branch*
+                // adds some 100 ns or so of overhead to all results --
+                // which is quite significant for very short tests...
+                0
+            })
+        });
+    } else {
+        b.iter(|| {
+            for _ in 0..ITERATIONS {
+                tx.send(&data, vec![], vec![]).unwrap();
+                rx.recv().unwrap();
+            }
+            0
+        });
+    }
+}
+
+#[bench]
+fn size_00_1(b: &mut test::Bencher) {
+    bench_size(b, 1);
+}
+#[bench]
+fn size_01_2(b: &mut test::Bencher) {
+    bench_size(b, 2);
+}
+#[bench]
+fn size_02_4(b: &mut test::Bencher) {
+    bench_size(b, 4);
+}
+#[bench]
+fn size_03_8(b: &mut test::Bencher) {
+    bench_size(b, 8);
+}
+#[bench]
+fn size_04_16(b: &mut test::Bencher) {
+    bench_size(b, 16);
+}
+#[bench]
+fn size_05_32(b: &mut test::Bencher) {
+    bench_size(b, 32);
+}
+#[bench]
+fn size_06_64(b: &mut test::Bencher) {
+    bench_size(b, 64);
+}
+#[bench]
+fn size_07_128(b: &mut test::Bencher) {
+    bench_size(b, 128);
+}
+#[bench]
+fn size_08_256(b: &mut test::Bencher) {
+    bench_size(b, 256);
+}
+#[bench]
+fn size_09_512(b: &mut test::Bencher) {
+    bench_size(b, 512);
+}
+#[bench]
+fn size_10_1k(b: &mut test::Bencher) {
+    bench_size(b, 1 * 1024);
+}
+#[bench]
+fn size_11_2k(b: &mut test::Bencher) {
+    bench_size(b, 2 * 1024);
+}
+#[bench]
+fn size_12_4k(b: &mut test::Bencher) {
+    bench_size(b, 4 * 1024);
+}
+#[bench]
+fn size_13_8k(b: &mut test::Bencher) {
+    bench_size(b, 8 * 1024);
+}
+#[bench]
+fn size_14_16k(b: &mut test::Bencher) {
+    bench_size(b, 16 * 1024);
+}
+#[bench]
+fn size_15_32k(b: &mut test::Bencher) {
+    bench_size(b, 32 * 1024);
+}
+#[bench]
+fn size_16_64k(b: &mut test::Bencher) {
+    bench_size(b, 64 * 1024);
+}
+#[bench]
+fn size_17_128k(b: &mut test::Bencher) {
+    bench_size(b, 128 * 1024);
+}
+#[bench]
+fn size_18_256k(b: &mut test::Bencher) {
+    bench_size(b, 256 * 1024);
+}
+#[bench]
+fn size_19_512k(b: &mut test::Bencher) {
+    bench_size(b, 512 * 1024);
+}
+#[bench]
+fn size_20_1m(b: &mut test::Bencher) {
+    bench_size(b, 1 * 1024 * 1024);
+}
+#[bench]
+fn size_21_2m(b: &mut test::Bencher) {
+    bench_size(b, 2 * 1024 * 1024);
+}
+#[bench]
+fn size_22_4m(b: &mut test::Bencher) {
+    bench_size(b, 4 * 1024 * 1024);
+}
+#[bench]
+fn size_23_8m(b: &mut test::Bencher) {
+    bench_size(b, 8 * 1024 * 1024);
+}

lib.rs

@@ -16,7 +16,6 @@
 extern crate lazy_static;
 
 extern crate bincode;
-extern crate byteorder;
 extern crate libc;
 extern crate rand;
 extern crate serde;

platform/inprocess/mod.rs

@@ -18,6 +18,7 @@ use std::fmt::{self, Debug, Formatter};
 use std::cmp::{PartialEq};
 use std::ops::Deref;
 use std::mem;
+use std::usize;
 
 use uuid::Uuid;
 
@@ -148,6 +149,10 @@ impl MpscSender {
         Ok(record.sender)
     }
 
+    pub fn get_max_fragment_size() -> usize {
+        usize::MAX
+    }
+
     pub fn send(&self,
                 data: &[u8],
                 ports: Vec<MpscChannel>,