Initial benchmarks, and some performance improvements.

c++/Makefile.ekam

+.PHONY: all once continuous continuous-opt clean
+
 all:
 	echo "You probably accidentally told Eclipse to build.  Stopping."
 
@@ -7,6 +9,9 @@ once:
 continuous:
 	CXX=g++-4.7 CXXFLAGS='-std=gnu++0x -g -Wall' LIBS='-lz -pthread' ekam -j6 -c -n :51315
 
+continuous-opt:
+	CXX=g++-4.7 CXXFLAGS='-std=gnu++0x -O2 -Wall' LIBS='-lz -pthread' ekam -j6 -c -n :51315
+
 clean:
 	rm -rf bin lib tmp
 

c++/src/capnproto/arena.c++

@@ -36,12 +36,22 @@ Arena::~Arena() {}
 
 ReaderArena::ReaderArena(MessageReader* message)
     : message(message),
-      readLimiter(this->message->getOptions().traversalLimitInWords * WORDS),
+      readLimiter(message->getOptions().traversalLimitInWords * WORDS),
       ignoreErrors(false),
-      segment0(this, SegmentId(0), this->message->getSegment(0), &readLimiter) {}
+      segment0(this, SegmentId(0), message->getSegment(0), &readLimiter) {}
 
 ReaderArena::~ReaderArena() {}
 
+void ReaderArena::reset() {
+  readLimiter.reset(message->getOptions().traversalLimitInWords * WORDS);
+  ignoreErrors = false;
+  segment0.~SegmentReader();
+  new(&segment0) SegmentReader(this, SegmentId(0), this->message->getSegment(0), &readLimiter);
+
+  // TODO:  Reuse the rest of the SegmentReaders?
+  moreSegments = nullptr;
+}
+
 SegmentReader* ReaderArena::tryGetSegment(SegmentId id) {
   if (id == SegmentId(0)) {
     if (segment0.getArray() == nullptr) {
@@ -100,6 +110,16 @@ BuilderArena::BuilderArena(MessageBuilder* message)
     : message(message), segment0(nullptr, SegmentId(0), nullptr, nullptr) {}
 BuilderArena::~BuilderArena() {}
 
+void BuilderArena::reset() {
+  segment0.reset();
+  if (moreSegments != nullptr) {
+    // TODO:  As mentioned in another TODO below, only the last segment will only be reused.
+    for (auto& segment: moreSegments->builders) {
+      segment->reset();
+    }
+  }
+}
+
 SegmentBuilder* BuilderArena::getSegment(SegmentId id) {
   // This method is allowed to crash if the segment ID is not valid.
   if (id == SegmentId(0)) {

c++/src/capnproto/arena.h

@@ -63,6 +63,8 @@ public:
   inline explicit ReadLimiter();                     // No limit.
   inline explicit ReadLimiter(WordCount64 limit);    // Limit to the given number of words.
 
+  inline void reset(WordCount64 limit);
+
   CAPNPROTO_ALWAYS_INLINE(bool canRead(WordCount amount, Arena* arena));
 
 private:
@@ -112,6 +114,8 @@ public:
 
   inline ArrayPtr<const word> currentlyAllocated();
 
+  inline void reset();
+
 private:
   word* pos;
 
@@ -160,6 +164,8 @@ public:
   ~ReaderArena();
   CAPNPROTO_DISALLOW_COPY(ReaderArena);
 
+  void reset();
+
   // implements Arena ------------------------------------------------
   SegmentReader* tryGetSegment(SegmentId id) override;
   void reportInvalidData(const char* description) override;
@@ -183,6 +189,9 @@ public:
   ~BuilderArena();
   CAPNPROTO_DISALLOW_COPY(BuilderArena);
 
+  void reset();
+  // Resets all the segments to be empty, so that a new message can be started.
+
   SegmentBuilder* getSegment(SegmentId id);
   // Get the segment with the given id.  Crashes or throws an exception if no such segment exists.
 
@@ -224,6 +233,8 @@ inline ReadLimiter::ReadLimiter()
 
 inline ReadLimiter::ReadLimiter(WordCount64 limit): limit(limit) {}
 
+inline void ReadLimiter::reset(WordCount64 limit) { this->limit = limit; }
+
 inline bool ReadLimiter::canRead(WordCount amount, Arena* arena) {
   if (CAPNPROTO_EXPECT_FALSE(amount > limit)) {
     arena->reportReadLimitReached();
@@ -293,6 +304,12 @@ inline ArrayPtr<const word> SegmentBuilder::currentlyAllocated() {
   return arrayPtr(ptr.begin(), pos - ptr.begin());
 }
 
+inline void SegmentBuilder::reset() {
+  word* start = getPtrUnchecked(0 * WORDS);
+  memset(start, 0, (pos - start) * sizeof(word));
+  pos = start;
+}
+
 }  // namespace internal
 }  // namespace capnproto
 

c++/src/capnproto/benchmark/benchmark-capnproto.c++

+// Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "benchmark.capnp.h"
+#include <capnproto/serialize.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <inttypes.h>
+#include <limits.h>
+#include <iostream>
+#include <stdlib.h>
+#include <stdexcept>
+#include <memory>
+#include <thread>
+#include <mutex>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <semaphore.h>
+
+namespace capnproto {
+namespace benchmark {
+namespace capnp {
+
+template <typename T>
+class ProducerConsumerQueue {
+public:
+  ProducerConsumerQueue() {
+    front = new Node;
+    back = front;
+    sem_init(&semaphore, 0, 0);
+  }
+
+  ~ProducerConsumerQueue() {
+    while (front != nullptr) {
+      Node* oldFront = front;
+      front = front->next;
+      delete oldFront;
+    }
+    sem_destroy(&semaphore);
+  }
+
+  void post(T t) {
+    back->next = new Node(t);
+    back = back->next;
+    sem_post(&semaphore);
+  }
+
+  T next() {
+    sem_wait(&semaphore);
+    Node* oldFront = front;
+    front = front->next;
+    delete oldFront;
+    return front->value;
+  }
+
+private:
+  struct Node {
+    T value;
+    Node* next;
+
+    Node(): next(nullptr) {}
+    Node(T value): value(value), next(nullptr) {}
+  };
+
+  Node* front;  // Last node that has been consumed.
+  Node* back;   // Last node in list.
+  sem_t semaphore;
+};
+
+class OsException: public std::exception {
+public:
+  OsException(int error): error(error) {}
+  ~OsException() noexcept {}
+
+  const char* what() const noexcept override {
+    // TODO:  Use strerror_r or whatever for thread-safety.  Ugh.
+    return strerror(error);
+  }
+
+private:
+  int error;
+};
+
+// =======================================================================================
+
+inline int32_t div(int32_t a, int32_t b) {
+  if (b == 0) return INT_MAX;
+  // INT_MIN / -1 => SIGFPE.  Who knew?
+  if (a == INT_MIN && b == -1) return INT_MAX;
+  return a / b;
+}
+
+inline int32_t mod(int32_t a, int32_t b) {
+  if (b == 0) return INT_MAX;
+  // INT_MIN % -1 => SIGFPE.  Who knew?
+  if (a == INT_MIN && b == -1) return INT_MAX;
+  return a % b;
+}
+
+int32_t makeExpression(Expression::Builder exp, int depth) {
+  if (rand() % 8 < depth) {
+    exp.setOp(Operation::VALUE);
+    exp.setValue(rand() % 128 + 1);
+    return exp.getValue();
+  } else {
+    // TODO:  Operation_MAX or something.
+    exp.setOp((Operation)(rand() % (int)Operation::MODULUS + 1));
+    int32_t left = makeExpression(exp.initLeft(), depth + 1);
+    int32_t right = makeExpression(exp.initRight(), depth + 1);
+    switch (exp.getOp()) {
+      case Operation::ADD:
+        return left + right;
+      case Operation::SUBTRACT:
+        return left - right;
+      case Operation::MULTIPLY:
+        return left * right;
+      case Operation::DIVIDE:
+        return div(left, right);
+      case Operation::MODULUS:
+        return mod(left, right);
+      case Operation::VALUE:
+        break;
+    }
+    throw std::logic_error("Can't get here.");
+  }
+}
+
+int32_t evaluateExpression(Expression::Reader exp) {
+  switch (exp.getOp()) {
+    case Operation::VALUE:
+      return exp.getValue();
+    case Operation::ADD:
+      return evaluateExpression(exp.getLeft()) + evaluateExpression(exp.getRight());
+    case Operation::SUBTRACT:
+      return evaluateExpression(exp.getLeft()) - evaluateExpression(exp.getRight());
+    case Operation::MULTIPLY:
+      return evaluateExpression(exp.getLeft()) * evaluateExpression(exp.getRight());
+    case Operation::DIVIDE:
+      return div(evaluateExpression(exp.getLeft()), evaluateExpression(exp.getRight()));
+    case Operation::MODULUS: {
+      return mod(evaluateExpression(exp.getLeft()), evaluateExpression(exp.getRight()));
+    }
+  }
+  throw std::logic_error("Can't get here.");
+}
+
+class ExpressionTestCase {
+public:
+  ~ExpressionTestCase() {}
+
+  typedef Expression Request;
+  typedef EvaluationResult Response;
+  typedef int32_t Expectation;
+
+  static inline int32_t setupRequest(Expression::Builder request) {
+    return makeExpression(request, 0);
+  }
+  static inline void handleRequest(Expression::Reader request, EvaluationResult::Builder response) {
+    response.setValue(evaluateExpression(request));
+  }
+  static inline bool checkResponse(EvaluationResult::Reader response, int32_t expected) {
+    return response.getValue() == expected;
+  }
+};
+
+// =======================================================================================
+
+template <typename TestCase>
+void syncClient(int inputFd, int outputFd, uint64_t iters) {
+  MallocMessageBuilder builder;
+//  StreamFdMessageReader reader(inputFd, ReaderOptions(), InputStrategy::EAGER_WAIT_FOR_READ_NEXT);
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = TestCase::setupRequest(
+        builder.initRoot<typename TestCase::Request>());
+    writeMessageToFd(outputFd, builder);
+
+//    reader.readNext();
+    StreamFdMessageReader reader(inputFd);
+    if (!TestCase::checkResponse(reader.getRoot<typename TestCase::Response>(), expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+  }
+}
+
+template <typename TestCase>
+void asyncClientSender(int outputFd,
+                       ProducerConsumerQueue<typename TestCase::Expectation>* expectations,
+                       uint64_t iters) {
+  MallocMessageBuilder builder;
+
+  for (; iters > 0; --iters) {
+    expectations->post(TestCase::setupRequest(builder.initRoot<typename TestCase::Request>()));
+    writeMessageToFd(outputFd, builder);
+  }
+}
+
+template <typename TestCase>
+void asyncClientReceiver(int inputFd,
+                         ProducerConsumerQueue<typename TestCase::Expectation>* expectations,
+                         uint64_t iters) {
+  StreamFdMessageReader reader(inputFd, ReaderOptions(), InputStrategy::EAGER_WAIT_FOR_READ_NEXT);
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = expectations->next();
+    reader.readNext();
+    if (!TestCase::checkResponse(reader.getRoot<typename TestCase::Response>(), expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+  }
+}
+
+template <typename TestCase>
+void asyncClient(int inputFd, int outputFd, uint64_t iters) {
+  ProducerConsumerQueue<typename TestCase::Expectation> expectations;
+  std::thread receiverThread(asyncClientReceiver<TestCase>, inputFd, &expectations, iters);
+  asyncClientSender<TestCase>(outputFd, &expectations, iters);
+  receiverThread.join();
+}
+
+template <typename TestCase>
+void server(int inputFd, int outputFd, uint64_t iters) {
+  StreamFdMessageReader reader(inputFd, ReaderOptions(), InputStrategy::EAGER_WAIT_FOR_READ_NEXT);
+  MallocMessageBuilder builder;
+
+  for (; iters > 0; --iters) {
+    reader.readNext();
+//    StreamFdMessageReader reader(inputFd);
+    TestCase::handleRequest(reader.getRoot<typename TestCase::Request>(),
+                            builder.initRoot<typename TestCase::Response>());
+    writeMessageToFd(outputFd, builder);
+  }
+}
+
+template <typename TestCase>
+void passByObject(uint64_t iters) {
+  MallocMessageBuilder requestMessage;
+  MallocMessageBuilder responseMessage;
+
+  for (; iters > 0; --iters) {
+    auto request = requestMessage.initRoot<typename TestCase::Request>();
+    typename TestCase::Expectation expected = TestCase::setupRequest(request);
+
+    auto response = responseMessage.initRoot<typename TestCase::Response>();
+    TestCase::handleRequest(request.asReader(), response);
+
+    if (!TestCase::checkResponse(response.asReader(), expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+  }
+}
+
+template <typename TestCase>
+void passByBytes(uint64_t iters) {
+  MallocMessageBuilder requestBuilder;
+  MallocMessageBuilder responseBuilder;
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = TestCase::setupRequest(
+        requestBuilder.initRoot<typename TestCase::Request>());
+
+    Array<word> requestBytes = messageToFlatArray(requestBuilder);
+    FlatArrayMessageReader requestReader(requestBytes.asPtr());
+    TestCase::handleRequest(requestReader.getRoot<typename TestCase::Request>(),
+                            responseBuilder.initRoot<typename TestCase::Response>());
+
+    Array<word> responseBytes = messageToFlatArray(responseBuilder);
+    FlatArrayMessageReader responseReader(responseBytes.asPtr());
+    if (!TestCase::checkResponse(responseReader.getRoot<typename TestCase::Response>(), expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+  }
+}
+
+template <typename TestCase, typename Func>
+void passByPipe(Func&& clientFunc, uint64_t iters) {
+  int clientToServer[2];
+  int serverToClient[2];
+  if (pipe(clientToServer) < 0) throw OsException(errno);
+  if (pipe(serverToClient) < 0) throw OsException(errno);
+
+  pid_t child = fork();
+  if (child == 0) {
+    // Client.
+    close(clientToServer[0]);
+    close(serverToClient[1]);
+
+    clientFunc(serverToClient[0], clientToServer[1], iters);
+    exit(0);
+  } else {
+    // Server.
+    close(clientToServer[1]);
+    close(serverToClient[0]);
+
+    server<TestCase>(clientToServer[0], serverToClient[1], iters);
+
+    int status;
+    if (waitpid(child, &status, 0) != child) {
+      throw OsException(errno);
+    }
+    if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
+      throw std::logic_error("Child exited abnormally.");
+    }
+  }
+}
+
+int main(int argc, char* argv[]) {
+  if (argc != 3) {
+    std::cerr << "USAGE:  " << argv[0] << " MODE ITERATION_COUNT" << std::endl;
+    return 1;
+  }
+
+  uint64_t iters = strtoull(argv[2], nullptr, 0);
+  srand(123);
+
+  std::cerr << "Doing " << iters << " iterations..." << std::endl;
+
+  std::string mode = argv[1];
+  if (mode == "client") {
+    syncClient<ExpressionTestCase>(STDIN_FILENO, STDOUT_FILENO, iters);
+  } else if (mode == "server") {
+    server<ExpressionTestCase>(STDIN_FILENO, STDOUT_FILENO, iters);
+  } else if (mode == "object") {
+    passByObject<ExpressionTestCase>(iters);
+  } else if (mode == "bytes") {
+    passByBytes<ExpressionTestCase>(iters);
+  } else if (mode == "pipe") {
+    passByPipe<ExpressionTestCase>(syncClient<ExpressionTestCase>, iters);
+  } else if (mode == "pipe-async") {
+    passByPipe<ExpressionTestCase>(asyncClient<ExpressionTestCase>, iters);
+  } else {
+    std::cerr << "Unknown mode: " << mode << std::endl;
+    return 1;
+  }
+
+  return 0;
+}
+
+}  // namespace protobuf
+}  // namespace benchmark
+}  // namespace capnproto
+
+int main(int argc, char* argv[]) {
+  return capnproto::benchmark::capnp::main(argc, argv);
+}

c++/src/capnproto/benchmark/benchmark-protobuf.c++

+// Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+#include "benchmark.pb.h"
+#include <google/protobuf/io/zero_copy_stream_impl.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <inttypes.h>
+#include <iostream>
+#include <stdlib.h>
+#include <stdexcept>
+#include <memory>
+#include <thread>
+#include <mutex>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <semaphore.h>
+
+namespace capnproto {
+namespace benchmark {
+namespace protobuf {
+
+template <typename T>
+class ProducerConsumerQueue {
+public:
+  ProducerConsumerQueue() {
+    front = new Node;
+    back = front;
+    sem_init(&semaphore, 0, 0);
+  }
+
+  ~ProducerConsumerQueue() {
+    while (front != nullptr) {
+      Node* oldFront = front;
+      front = front->next;
+      delete oldFront;
+    }
+    sem_destroy(&semaphore);
+  }
+
+  void post(T t) {
+    back->next = new Node(t);
+    back = back->next;
+    sem_post(&semaphore);
+  }
+
+  T next() {
+    sem_wait(&semaphore);
+    Node* oldFront = front;
+    front = front->next;
+    delete oldFront;
+    return front->value;
+  }
+
+private:
+  struct Node {
+    T value;
+    Node* next;
+
+    Node(): next(nullptr) {}
+    Node(T value): value(value), next(nullptr) {}
+  };
+
+  Node* front;  // Last node that has been consumed.
+  Node* back;   // Last node in list.
+  sem_t semaphore;
+};
+
+class OsException: public std::exception {
+public:
+  OsException(int error): error(error) {}
+  ~OsException() noexcept {}
+
+  const char* what() const noexcept override {
+    // TODO:  Use strerror_r or whatever for thread-safety.  Ugh.
+    return strerror(error);
+  }
+
+private:
+  int error;
+};
+
+// =======================================================================================
+
+inline int32_t div(int32_t a, int32_t b) {
+  if (b == 0) return INT_MAX;
+  // INT_MIN / -1 => SIGFPE.  Who knew?
+  if (a == INT_MIN && b == -1) return INT_MAX;
+  return a / b;
+}
+
+inline int32_t mod(int32_t a, int32_t b) {
+  if (b == 0) return INT_MAX;
+  // INT_MIN % -1 => SIGFPE.  Who knew?
+  if (a == INT_MIN && b == -1) return INT_MAX;
+  return a % b;
+}
+
+int32_t makeExpression(Expression* exp, int depth) {
+  if (rand() % 8 < depth) {
+    exp->set_op(Operation::VALUE);
+    exp->set_value(rand() % 128 + 1);
+    return exp->value();
+  } else {
+    exp->set_op((Operation)(rand() % Operation_MAX + 1));
+    int32_t left = makeExpression(exp->mutable_left(), depth + 1);
+    int32_t right = makeExpression(exp->mutable_right(), depth + 1);
+    switch (exp->op()) {
+      case Operation::ADD:
+        return left + right;
+      case Operation::SUBTRACT:
+        return left - right;
+      case Operation::MULTIPLY:
+        return left * right;
+      case Operation::DIVIDE:
+        return div(left, right);
+      case Operation::MODULUS:
+        return mod(left, right);
+      case Operation::VALUE:
+        break;
+    }
+    throw std::logic_error("Can't get here.");
+  }
+}
+
+int32_t evaluateExpression(const Expression& exp) {
+  switch (exp.op()) {
+    case Operation::VALUE:
+      return exp.value();
+    case Operation::ADD:
+      return evaluateExpression(exp.left()) + evaluateExpression(exp.right());
+    case Operation::SUBTRACT:
+      return evaluateExpression(exp.left()) - evaluateExpression(exp.right());
+    case Operation::MULTIPLY:
+      return evaluateExpression(exp.left()) * evaluateExpression(exp.right());
+    case Operation::DIVIDE:
+      return div(evaluateExpression(exp.left()), evaluateExpression(exp.right()));
+    case Operation::MODULUS: {
+      return mod(evaluateExpression(exp.left()), evaluateExpression(exp.right()));
+    }
+  }
+  throw std::logic_error("Can't get here.");
+}
+
+class ExpressionTestCase {
+public:
+  ~ExpressionTestCase() {}
+
+  typedef Expression Request;
+  typedef EvaluationResult Response;
+  typedef int32_t Expectation;
+
+  static inline int32_t setupRequest(Expression* request) {
+    return makeExpression(request, 0);
+  }
+  static inline void handleRequest(const Expression& request, EvaluationResult* response) {
+    response->set_value(evaluateExpression(request));
+  }
+  static inline bool checkResponse(const EvaluationResult& response, int32_t expected) {
+    return response.value() == expected;
+  }
+};
+
+// =======================================================================================
+
+void writeProtoFast(const google::protobuf::MessageLite& message,
+                    google::protobuf::io::FileOutputStream* rawOutput) {
+  google::protobuf::io::CodedOutputStream output(rawOutput);
+  const int size = message.ByteSize();
+  output.WriteVarint32(size);
+  uint8_t* buffer = output.GetDirectBufferForNBytesAndAdvance(size);
+  if (buffer != NULL) {
+    message.SerializeWithCachedSizesToArray(buffer);
+  } else {
+    message.SerializeWithCachedSizes(&output);
+    if (output.HadError()) {
+      throw OsException(rawOutput->GetErrno());
+    }
+  }
+}
+
+void readProtoFast(google::protobuf::io::ZeroCopyInputStream* rawInput,
+                   google::protobuf::MessageLite* message) {
+  google::protobuf::io::CodedInputStream input(rawInput);
+  uint32_t size;
+  if (!input.ReadVarint32(&size)) {
+    throw std::logic_error("Read failed.");
+  }
+
+  auto limit = input.PushLimit(size);
+
+  if (!message->MergePartialFromCodedStream(&input) ||
+      !input.ConsumedEntireMessage()) {
+    throw std::logic_error("Read failed.");
+  }
+
+  input.PopLimit(limit);
+}
+
+template <typename TestCase>
+void syncClient(int inputFd, int outputFd, uint64_t iters) {
+  google::protobuf::io::FileOutputStream output(outputFd);
+  google::protobuf::io::FileInputStream input(inputFd);
+
+  typename TestCase::Request request;
+  typename TestCase::Response response;
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = TestCase::setupRequest(&request);
+    writeProtoFast(request, &output);
+    if (!output.Flush()) throw OsException(output.GetErrno());
+    request.Clear();
+
+    // std::cerr << "client: wait" << std::endl;
+    readProtoFast(&input, &response);
+    if (!TestCase::checkResponse(response, expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+    response.Clear();
+  }
+}
+
+template <typename TestCase>
+void asyncClientSender(int outputFd,
+                       ProducerConsumerQueue<typename TestCase::Expectation>* expectations,
+                       uint64_t iters) {
+  google::protobuf::io::FileOutputStream output(outputFd);
+  typename TestCase::Request request;
+
+  for (; iters > 0; --iters) {
+    expectations->post(TestCase::setupRequest(&request));
+    writeProtoFast(request, &output);
+    request.Clear();
+  }
+
+  if (!output.Flush()) throw OsException(output.GetErrno());
+}
+
+template <typename TestCase>
+void asyncClientReceiver(int inputFd,
+                         ProducerConsumerQueue<typename TestCase::Expectation>* expectations,
+                         uint64_t iters) {
+  google::protobuf::io::FileInputStream input(inputFd);
+  typename TestCase::Response response;
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = expectations->next();
+    readProtoFast(&input, &response);
+    if (!TestCase::checkResponse(response, expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+    response.Clear();
+  }
+}
+
+template <typename TestCase>
+void asyncClient(int inputFd, int outputFd, uint64_t iters) {
+  ProducerConsumerQueue<typename TestCase::Expectation> expectations;
+  std::thread receiverThread(asyncClientReceiver<TestCase>, inputFd, &expectations, iters);
+  asyncClientSender<TestCase>(outputFd, &expectations, iters);
+  receiverThread.join();
+}
+
+template <typename TestCase>
+void server(int inputFd, int outputFd, uint64_t iters) {
+  google::protobuf::io::FileOutputStream output(outputFd);
+  google::protobuf::io::FileInputStream input(inputFd);
+
+  typename TestCase::Request request;
+  typename TestCase::Response response;
+
+  for (; iters > 0; --iters) {
+    readProtoFast(&input, &request);
+    TestCase::handleRequest(request, &response);
+    request.Clear();
+
+    writeProtoFast(response, &output);
+    if (!output.Flush()) throw std::logic_error("Write failed.");
+    response.Clear();
+  }
+}
+
+template <typename TestCase>
+void passByObject(uint64_t iters) {
+  typename TestCase::Request request;
+  typename TestCase::Response response;
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = TestCase::setupRequest(&request);
+    TestCase::handleRequest(request, &response);
+    request.Clear();
+    if (!TestCase::checkResponse(response, expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+    response.Clear();
+  }
+}
+
+template <typename TestCase>
+void passByBytes(uint64_t iters) {
+  typename TestCase::Request clientRequest;
+  typename TestCase::Request serverRequest;
+  typename TestCase::Response serverResponse;
+  typename TestCase::Response clientResponse;
+  std::string requestString, responseString;
+
+  for (; iters > 0; --iters) {
+    typename TestCase::Expectation expected = TestCase::setupRequest(&clientRequest);
+
+    clientRequest.SerializePartialToString(&requestString);
+    clientRequest.Clear();
+
+    serverRequest.ParsePartialFromString(requestString);
+    requestString.clear();
+
+    TestCase::handleRequest(serverRequest, &serverResponse);
+    serverRequest.Clear();
+
+    serverResponse.SerializePartialToString(&responseString);
+    serverResponse.Clear();
+
+    clientResponse.ParsePartialFromString(responseString);
+    responseString.clear();
+
+    if (!TestCase::checkResponse(clientResponse, expected)) {
+      throw std::logic_error("Incorrect response.");
+    }
+    clientResponse.Clear();
+  }
+}
+
+template <typename TestCase, typename Func>
+void passByPipe(Func&& clientFunc, uint64_t iters) {
+  int clientToServer[2];
+  int serverToClient[2];
+  if (pipe(clientToServer) < 0) throw OsException(errno);
+  if (pipe(serverToClient) < 0) throw OsException(errno);
+
+  pid_t child = fork();
+  if (child == 0) {
+    // Client.
+    close(clientToServer[0]);
+    close(serverToClient[1]);
+
+    clientFunc(serverToClient[0], clientToServer[1], iters);
+    exit(0);
+  } else {
+    // Server.
+    close(clientToServer[1]);
+    close(serverToClient[0]);
+
+    server<TestCase>(clientToServer[0], serverToClient[1], iters);
+
+    int status;
+    if (waitpid(child, &status, 0) != child) {
+      throw OsException(errno);
+    }
+    if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
+      throw std::logic_error("Child exited abnormally.");
+    }
+  }
+}
+
+int main(int argc, char* argv[]) {
+  if (argc != 3) {
+    std::cerr << "USAGE:  " << argv[0] << " MODE ITERATION_COUNT" << std::endl;
+    return 1;
+  }
+
+  uint64_t iters = strtoull(argv[2], nullptr, 0);
+  srand(123);
+
+  std::cerr << "Doing " << iters << " iterations..." << std::endl;
+
+  std::string mode = argv[1];
+  if (mode == "client") {
+    syncClient<ExpressionTestCase>(STDIN_FILENO, STDOUT_FILENO, iters);
+  } else if (mode == "server") {
+    server<ExpressionTestCase>(STDIN_FILENO, STDOUT_FILENO, iters);
+  } else if (mode == "object") {
+    passByObject<ExpressionTestCase>(iters);
+  } else if (mode == "bytes") {
+    passByBytes<ExpressionTestCase>(iters);
+  } else if (mode == "pipe") {
+    passByPipe<ExpressionTestCase>(syncClient<ExpressionTestCase>, iters);
+  } else if (mode == "pipe-async") {
+    passByPipe<ExpressionTestCase>(asyncClient<ExpressionTestCase>, iters);
+  } else {
+    std::cerr << "Unknown mode: " << mode << std::endl;
+    return 1;
+  }
+
+  return 0;
+}
+
+}  // namespace protobuf
+}  // namespace benchmark
+}  // namespace capnproto
+
+int main(int argc, char* argv[]) {
+  return capnproto::benchmark::protobuf::main(argc, argv);
+}

c++/src/capnproto/benchmark/benchmark.capnp

+# Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice, this
+#    list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+enum Operation {
+  value = 0;
+  add = 1;
+  subtract = 2;
+  multiply = 3;
+  divide = 4;
+  modulus = 5;
+}
+
+struct Expression {
+  op@0: Operation;
+  value@1: Int32;
+  left@2: Expression;
+  right@3: Expression;
+}
+
+struct EvaluationResult {
+  value@0: Int32;
+}

c++/src/capnproto/benchmark/benchmark.proto

+// Copyright (c) 2013, Kenton Varda <temporal@gmail.com>
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+package capnproto.benchmark.protobuf;
+
+enum Operation {
+  VALUE = 0;
+  ADD = 1;
+  SUBTRACT = 2;
+  MULTIPLY = 3;
+  DIVIDE = 4;
+  MODULUS = 5;
+}
+
+message Expression {
+  required Operation op = 1;
+  optional int32 value = 2;
+  optional Expression left = 3;
+  optional Expression right = 4;
+}
+
+message EvaluationResult {
+  required sint32 value = 1;
+}

c++/src/capnproto/message.c++

@@ -38,6 +38,10 @@ MessageReader::~MessageReader() {
   }
 }
 
+void MessageReader::reset() {
+  if (allocatedArena) arena()->reset();
+}
+
 internal::StructReader MessageReader::getRoot(const word* defaultValue) {
   if (!allocatedArena) {
     static_assert(sizeof(internal::ReaderArena) <= sizeof(arenaSpace),
@@ -67,35 +71,35 @@ MessageBuilder::~MessageBuilder() {
   }
 }
 
-internal::SegmentBuilder* MessageBuilder::getRootSegment() {
-  if (allocatedArena) {
-    return arena()->getSegment(SegmentId(0));
-  } else {
+internal::SegmentBuilder* MessageBuilder::allocateRootSegment() {
+  if (!allocatedArena) {
     static_assert(sizeof(internal::BuilderArena) <= sizeof(arenaSpace),
         "arenaSpace is too small to hold a BuilderArena.  Please increase it.  This will break "
         "ABI compatibility.");
     new(arena()) internal::BuilderArena(this);
     allocatedArena = true;
-
-    WordCount refSize = 1 * REFERENCES * WORDS_PER_REFERENCE;
-    internal::SegmentBuilder* segment = arena()->getSegmentWithAvailable(refSize);
-    CAPNPROTO_ASSERT(segment->getSegmentId() == SegmentId(0),
-        "First allocated word of new arena was not in segment ID 0.");
-    word* location = segment->allocate(refSize);
-    CAPNPROTO_ASSERT(location == segment->getPtrUnchecked(0 * WORDS),
-        "First allocated word of new arena was not the first word in its segment.");
-    return segment;
   }
+
+  WordCount refSize = 1 * REFERENCES * WORDS_PER_REFERENCE;
+  internal::SegmentBuilder* segment = arena()->getSegmentWithAvailable(refSize);
+  CAPNPROTO_ASSERT(segment->getSegmentId() == SegmentId(0),
+      "First allocated word of new arena was not in segment ID 0.");
+  word* location = segment->allocate(refSize);
+  CAPNPROTO_ASSERT(location == segment->getPtrUnchecked(0 * WORDS),
+      "First allocated word of new arena was not the first word in its segment.");
+  return segment;
 }
 
 internal::StructBuilder MessageBuilder::initRoot(const word* defaultValue) {
-  internal::SegmentBuilder* rootSegment = getRootSegment();
+  if (allocatedArena) arena()->reset();
+  internal::SegmentBuilder* rootSegment = allocateRootSegment();
   return internal::StructBuilder::initRoot(
       rootSegment, rootSegment->getPtrUnchecked(0 * WORDS), defaultValue);
 }
 
 internal::StructBuilder MessageBuilder::getRoot(const word* defaultValue) {
-  internal::SegmentBuilder* rootSegment = getRootSegment();
+  internal::SegmentBuilder* rootSegment = allocatedArena ?
+      arena()->getSegment(SegmentId(0)) : allocateRootSegment();
   return internal::StructBuilder::getRoot(
       rootSegment, rootSegment->getPtrUnchecked(0 * WORDS), defaultValue);
 }

c++/src/capnproto/message.h

@@ -123,6 +123,9 @@ public:
 
   virtual ArrayPtr<const word> getSegment(uint id) = 0;
   // Gets the segment with the given ID, or returns null if no such segment exists.
+  //
+  // Normally getSegment() will only be called once for each segment ID.  Subclasses can call
+  // reset() to clear the segment table and start over with new segments.
 
   inline const ReaderOptions& getOptions();
   // Get the options passed to the constructor.
@@ -130,6 +133,14 @@ public:
   template <typename RootType>
   typename RootType::Reader getRoot();
 
+protected:
+  void reset();
+  // Clear the cached segment table so that the reader can be reused to read another message.
+  // reset() may call getSegment() again before returning, so you must arrange for the new segment
+  // set to be active *before* calling this.
+  //
+  // This invalidates any Readers currently pointing into this message.
+
 private:
   ReaderOptions options;
 
@@ -171,7 +182,7 @@ private:
   bool allocatedArena = false;
 
   internal::BuilderArena* arena() { return reinterpret_cast<internal::BuilderArena*>(arenaSpace); }
-  internal::SegmentBuilder* getRootSegment();
+  internal::SegmentBuilder* allocateRootSegment();
   internal::StructBuilder initRoot(const word* defaultValue);
   internal::StructBuilder getRoot(const word* defaultValue);
 };

c++/src/capnproto/serialize.c++

@@ -134,7 +134,60 @@ OutputStream::~OutputStream() {}
 
 InputStreamMessageReader::InputStreamMessageReader(
     InputStream* inputStream, ReaderOptions options, InputStrategy inputStrategy)
-    : MessageReader(options), inputStream(inputStream), segmentsReadSoFar(0) {
+    : MessageReader(options), inputStream(inputStream), inputStrategy(inputStrategy),
+      segmentsReadSoFar(0) {
+  switch (inputStrategy) {
+    case InputStrategy::EAGER:
+    case InputStrategy::LAZY:
+      readNextInternal();
+      break;
+    case InputStrategy::EAGER_WAIT_FOR_READ_NEXT:
+    case InputStrategy::LAZY_WAIT_FOR_READ_NEXT:
+      break;
+  }
+}
+
+void InputStreamMessageReader::readNext() {
+  bool needReset = false;
+
+  switch (inputStrategy) {
+    case InputStrategy::LAZY:
+      if (moreSegments != nullptr || segment0.size != 0) {
+        // Make sure we've finished reading the previous message.
+        // Note that this sort of defeats the purpose of lazy parsing.  In theory we could be a
+        // little more efficient by reading into a stack-allocated scratch buffer rather than
+        // allocating space for the remaining segments, but people really shouldn't be using
+        // readNext() when lazy-parsing anyway.
+        getSegment(moreSegments.size());
+      }
+
+      // no break
+
+    case InputStrategy::EAGER:
+      needReset = true;
+
+      // TODO:  Save moreSegments for reuse?
+      moreSegments = nullptr;
+
+      segmentsReadSoFar = 0;
+      segment0.size = 0;
+      break;
+
+    case InputStrategy::EAGER_WAIT_FOR_READ_NEXT:
+      this->inputStrategy = InputStrategy::EAGER;
+      break;
+    case InputStrategy::LAZY_WAIT_FOR_READ_NEXT:
+      this->inputStrategy = InputStrategy::LAZY;
+      break;
+  }
+
+  if (inputStream != nullptr) {
+    readNextInternal();
+  }
+  if (needReset) reset();
+}
+
+void InputStreamMessageReader::readNextInternal() {
   internal::WireValue<uint32_t> firstWord[2];
 
   if (!inputStream->read(firstWord, sizeof(firstWord))) return;
@@ -160,7 +213,6 @@ InputStreamMessageReader::InputStreamMessageReader(
 
   if (inputStrategy == InputStrategy::EAGER) {
     getSegment(segmentCount - 1);
-    inputStream = nullptr;
   }
 }
 
@@ -171,20 +223,22 @@ ArrayPtr<const word> InputStreamMessageReader::getSegment(uint id) {
 
   while (segmentsReadSoFar <= id && inputStream != nullptr) {
     LazySegment& segment = segmentsReadSoFar == 0 ? segment0 : moreSegments[segmentsReadSoFar - 1];
-    Array<word> words = newArray<word>(segment.size);
+    if (segment.words.size() < segment.size) {
+      segment.words = newArray<word>(segment.size);
+    }
 
-    if (!inputStream->read(words.begin(), words.size() * sizeof(word))) {
+    if (!inputStream->read(segment.words.begin(), segment.size * sizeof(word))) {
       // There was an error but no exception was thrown, so we're supposed to plod along with
       // default values.  Discard the broken stream.
       inputStream = nullptr;
       break;
     }
 
-    segment.words = move(words);
     ++segmentsReadSoFar;
   }
 
-  return id == 0 ? segment0.words.asPtr() : moreSegments[id - 1].words.asPtr();
+  LazySegment& segment = id == 0 ? segment0 : moreSegments[id - 1];
+  return segment.words.slice(0, segment.size);
 }
 
 // -------------------------------------------------------------------

c++/src/capnproto/serialize.h

@@ -103,11 +103,17 @@ enum class InputStrategy {
   // an InputStream, the stream will then be positioned at the byte immediately after the end of
   // the message, and will not be accessed again.
 
-  LAZY
+  LAZY,
   // Read segments of the message into RAM as needed while the message structure is being traversed.
   // When reading from an InputStream, segments must be read in order, so segments up to the
   // required segment will also be read.  No guarantee is made about the position of the InputStream
   // after reading.  When using an InputFile, only the exact segments desired are read.
+
+  EAGER_WAIT_FOR_READ_NEXT,
+  // Like EAGER but don't read the first mesasge until readNext() is called the first time.
+
+  LAZY_WAIT_FOR_READ_NEXT,
+  // Like LAZY but don't read the first mesasge until readNext() is called the first time.
 };
 
 class InputStreamMessageReader: public MessageReader {
@@ -116,16 +122,23 @@ public:
                            ReaderOptions options = ReaderOptions(),
                            InputStrategy inputStrategy = InputStrategy::EAGER);
 
+  void readNext();
+  // Progress to the next message in the input stream, reusing the same memory if possible.
+  // Calling this invalidates any Readers currently pointing into this message.
+
   // implements MessageReader ----------------------------------------
   ArrayPtr<const word> getSegment(uint id) override;
 
 private:
   InputStream* inputStream;
+  InputStrategy inputStrategy;
   uint segmentsReadSoFar;
 
   struct LazySegment {
     uint size;
-    Array<word> words;   // null until actually read
+    Array<word> words;
+    // words may be larger than the desired size in the case where space is being reused from a
+    // previous read.
 
     inline LazySegment(): size(0), words(nullptr) {}
   };
@@ -133,6 +146,8 @@ private:
   // Optimize for single-segment case.
   LazySegment segment0;
   Array<LazySegment> moreSegments;
+
+  void readNextInternal();
 };
 
 class InputFileMessageReader: public MessageReader {

compiler/src/CxxGenerator.hs

@@ -157,6 +157,16 @@ elementType _ = error "Called elementType on non-list."
 repeatedlyTake _ [] = []
 repeatedlyTake n l = take n l : repeatedlyTake n (drop n l)
 
+enumValueContext parent desc = mkStrContext context where
+    context "enumValueName" = MuVariable $ toUpperCaseWithUnderscores $ enumValueName desc
+    context "enumValueNumber" = MuVariable $ enumValueNumber desc
+    context s = parent s
+
+enumContext parent desc = mkStrContext context where
+    context "enumName" = MuVariable $ enumName desc
+    context "enumValues" = MuList $ map (enumValueContext context) $ enumValues desc
+    context s = parent s
+
 defaultBytesContext :: Monad m => (String -> MuType m) -> TypeDesc -> [Word8] -> MuContext m
 defaultBytesContext parent t bytes = mkStrContext context where
     codeLines = map (delimit ", ") $ repeatedlyTake 8 $ map (printf "%3d") bytes
@@ -212,6 +222,7 @@ fileContext desc = mkStrContext context where
     context "fileIncludeGuard" = MuVariable $
         "CAPNPROTO_INCLUDED_" ++ hashString (fileName desc)
     context "fileNamespaces" = MuList []  -- TODO
+    context "fileEnums" = MuList $ map (enumContext context) $ fileEnums desc
     context "fileStructs" = MuList $ map (structContext context) $ fileStructs desc
     context s = error ("Template variable not defined: " ++ s)
 

compiler/src/c++-header.mustache

@@ -54,6 +54,14 @@ struct {{structName}} {
 {{/structFields}}
 };
 {{/fileStructs}}
+{{#fileEnums}}
+
+enum class {{enumName}}: uint16_t {
+{{#enumValues}}
+  {{enumValueName}} = {{enumValueNumber}},
+{{/enumValues}}
+};
+{{/fileEnums}}
 {{#fileStructs}}
 
 class {{structName}}::Reader {