wip

command_line.cc

@@ -95,6 +95,17 @@ std::unique_ptr<InMessage> ParseMessage() {
   return MessageRegistry::instance()->Parse(document);
 }
 
+std::string Join(const std::vector<std::string>& elements, std::string sep) {
+  bool first = true;
+  std::string result;
+  for (const auto& element : elements) {
+    if (!first)
+      result += ", ";
+    first = false;
+    result += element;
+  }
+  return result;
+}
 
 
 
@@ -176,6 +187,32 @@ void Reflect(TVisitor& visitor, IpcMessage_OpenProject& value) {
 
 
 
+struct IpcMessage_IndexTranslationUnitRequest : public BaseIpcMessage<IpcMessage_IndexTranslationUnitRequest> {
+  static constexpr IpcId kIpcId = IpcId::IndexTranslationUnitRequest;
+  std::string path;
+  std::vector<std::string> args;
+};
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, IpcMessage_IndexTranslationUnitRequest& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(path);
+  REFLECT_MEMBER(args);
+  REFLECT_MEMBER_END();
+}
+
+struct IpcMessage_IndexTranslationUnitResponse : public BaseIpcMessage<IpcMessage_IndexTranslationUnitResponse> {
+  static constexpr IpcId kIpcId = IpcId::IndexTranslationUnitResponse;
+  IndexUpdate update;
+
+  explicit IpcMessage_IndexTranslationUnitResponse(IndexUpdate& update) : update(update) {}
+};
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, IpcMessage_IndexTranslationUnitResponse& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(update);
+  REFLECT_MEMBER_END();
+}
+
 
 
 
@@ -254,24 +291,84 @@ void Reflect(TVisitor& visitor, IpcMessage_WorkspaceSymbolsResponse& value) {
 
 
 
+struct Timer {
+  using Clock = std::chrono::high_resolution_clock;
+  std::chrono::time_point<Clock> start_;
+
+  Timer() {
+    Reset();
+  }
+
+  void Reset() {
+    start_ = Clock::now();
+  }
+
+  long long ElapsedMilliseconds() {
+    std::chrono::time_point<Clock> end = Clock::now();
+    return std::chrono::duration_cast<std::chrono::milliseconds>(end - start_).count();
+  }
+};
 
 
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
+void IndexMainLoop(IpcClient* ipc) {
   std::vector<std::unique_ptr<IpcMessage>> messages = ipc->TakeMessages();
+  for (auto& message : messages) {
+    std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
+
+    switch (message->ipc_id) {
+    case IpcId::Quit: {
+      exit(0);
+      break;
+    }
+
+    case IpcId::IndexTranslationUnitRequest: {
+      IpcMessage_IndexTranslationUnitRequest* msg = static_cast<IpcMessage_IndexTranslationUnitRequest*>(message.get());
+
+      std::cerr << "Parsing file " << msg->path << " with args " << Join(msg->args, ", ") << std::endl;
+
+      Timer time;
+      IndexedFile file = Parse(msg->path, msg->args);
+      std::cerr << "Parsing/indexing took " << time.ElapsedMilliseconds() << "ms" << std::endl;
+
+      time.Reset();
+      auto response = IpcMessage_IndexTranslationUnitResponse(IndexUpdate(file));
+      std::cerr << "Creating index update took " << time.ElapsedMilliseconds() << "ms" << std::endl;
+
+      time.Reset();
+      ipc->SendToServer(&response);
+      std::cerr << "Sending to server took " << time.ElapsedMilliseconds() << "ms" << std::endl;
+
+      break;
+    }
+    }
+  }
+}
+
+void IndexMain(int id) {
+  return;
+
+  IpcClient client_ipc("indexer", id);
+
+  while (true) {
+    IndexMainLoop(&client_ipc);
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+}
+
+
+
+
+
+
+
+
+
+
+
+
+
+void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>& indexers, QueryableDatabase* db) {
+  std::vector<std::unique_ptr<IpcMessage>> messages = client->TakeMessages();
 
   for (auto& message : messages) {
     std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
@@ -283,7 +380,7 @@ void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
 
     case IpcId::IsAlive: {
       IpcMessage_IsAlive response;
-      ipc->SendToClient(0, &response); // todo: make non-blocking
+      client->SendToClient(&response); // todo: make non-blocking
       break;
     }
 
@@ -296,15 +393,29 @@ void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
       for (int i = 0; i < entries.size(); ++i) {
         const CompilationEntry& entry = entries[i];
         std::string filepath = path + "/" + entry.filename;
-        std::cerr << "[" << i << "/" << (entries.size() - 1) << "] Parsing file " << filepath << std::endl;
-        IndexedFile file = Parse(filepath, entry.args);
-        IndexUpdate update(file);
-        db->ApplyIndexUpdate(&update);
+        std::cerr << "[" << i << "/" << (entries.size() - 1) << "] Dispatching index request for file " << filepath << std::endl;
+
+        // TODO: indexers should steal work. load balance.
+        IpcMessage_IndexTranslationUnitRequest request;
+        request.path = filepath;
+        request.args = entry.args;
+        indexers[i % indexers.size()]->SendToClient(&request);
+        //IndexedFile file = Parse(filepath, entry.args);
+        //IndexUpdate update(file);
+        //db->ApplyIndexUpdate(&update);
       }
       std::cerr << "Done" << std::endl;
       break;
     }
 
+    case IpcId::IndexTranslationUnitResponse: {
+      IpcMessage_IndexTranslationUnitResponse* msg = static_cast<IpcMessage_IndexTranslationUnitResponse*>(message.get());
+      Timer time;
+      db->ApplyIndexUpdate(&msg->update);
+      std::cerr << "Applying index update took " << time.ElapsedMilliseconds() << "ms" << std::endl;
+      break;
+    }
+
     case IpcId::DocumentSymbolsRequest: {
       auto msg = static_cast<IpcMessage_DocumentSymbolsRequest*>(message.get());
 
@@ -376,7 +487,7 @@ void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
 
 
 
-      ipc->SendToClient(0, &response);
+      client->SendToClient(&response);
 
       break;
     }
@@ -466,7 +577,7 @@ void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
       }
 
 
-      ipc->SendToClient(0, &response);
+      client->SendToClient(&response);
       break;
     }
 
@@ -515,7 +626,7 @@ void QueryDbMainLoop(IpcServer* ipc, QueryableDatabase* db) {
 // blocks.
 //
 // |ipc| is connected to a server.
-void LanguageServerStdinLoop(IpcClient* ipc) {
+void LanguageServerStdinLoop(IpcClient* server) {
   while (true) {
     std::unique_ptr<InMessage> message = ParseMessage();
 
@@ -533,7 +644,7 @@ void LanguageServerStdinLoop(IpcClient* ipc) {
         std::cerr << "Initialize in directory " << project_path << " with uri " << request->params.rootUri->raw_uri << std::endl;
         IpcMessage_OpenProject open_project;
         open_project.project_path = project_path;
-        ipc->SendToServer(&open_project);
+        server->SendToServer(&open_project);
       }
 
       auto response = Out_InitializeResponse();
@@ -554,7 +665,7 @@ void LanguageServerStdinLoop(IpcClient* ipc) {
       ipc_request.request_id = request->id.value();
       ipc_request.document = request->params.textDocument.uri.GetPath();
       std::cerr << "Request textDocument=" << ipc_request.document << std::endl;
-      ipc->SendToServer(&ipc_request);
+      server->SendToServer(&ipc_request);
       break;
     }
 
@@ -565,7 +676,7 @@ void LanguageServerStdinLoop(IpcClient* ipc) {
       ipc_request.request_id = request->id.value();
       ipc_request.query = request->params.query;
       std::cerr << "Request query=" << ipc_request.query << std::endl;
-      ipc->SendToServer(&ipc_request);
+      server->SendToServer(&ipc_request);
       break;
     }
     }
@@ -611,6 +722,8 @@ void LanguageServerMainLoop(IpcClient* ipc) {
   }
 }
 
+const int kNumIndexers = 8 - 1;
+
 void LanguageServerMain(std::string process_name) {
   IpcClient client_ipc("languageserver", 0);
 
@@ -622,7 +735,7 @@ void LanguageServerMain(std::string process_name) {
   client_ipc.SendToServer(&check_alive);
 
   // TODO: Tune this value or make it configurable.
-  std::this_thread::sleep_for(std::chrono::milliseconds(20));
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
   // Check if we got an IsAlive message back.
   std::vector<std::unique_ptr<IpcMessage>> messages = client_ipc.TakeMessages();
@@ -663,15 +776,33 @@ void LanguageServerMain(std::string process_name) {
 
   std::thread stdio_reader(&LanguageServerStdinLoop, &client_ipc);
 
+
+
   if (!has_server) {
+    std::vector<std::unique_ptr<IpcServer>> server_indexers;
+    IpcServer server_ipc("languageserver", 0);
+    QueryableDatabase db;
+
     // No server. Run it in-process.
     new std::thread([&]() {
-      IpcServer server_ipc("languageserver");
-      QueryableDatabase db;
+
+      std::cerr << "!! starting processes" << std::endl;
+      // Start indexer processes.
+      // TODO: make sure to handle this when running querydb out of process.
+      for (int i = 0; i < kNumIndexers; ++i) {
+        server_indexers.emplace_back(MakeUnique<IpcServer>("indexer", i + 1));
+        //new Process(process_name + " --indexer " + std::to_string(i + 1));
+        new std::thread([i]() {
+          IndexMain(i + 1);
+        });
+      }
+      std::cerr << "!! done processes" << std::endl;
+
+
+      
       while (true) {
-        QueryDbMainLoop(&server_ipc, &db);
-        // TODO: use a condition variable.
-        std::this_thread::sleep_for(std::chrono::microseconds(0));
+        QueryDbMainLoop(&server_ipc, server_indexers, &db);
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
       }
     });
   }
@@ -679,8 +810,7 @@ void LanguageServerMain(std::string process_name) {
   // Run language client.
   while (true) {
     LanguageServerMainLoop(&client_ipc);
-    // TODO: use a condition variable.
-    std::this_thread::sleep_for(std::chrono::microseconds(0));
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
   }
 }
 
@@ -733,7 +863,7 @@ void LanguageServerMain(std::string process_name) {
 int main(int argc, char** argv) {
   bool loop = false;
   while (loop)
-    std::this_thread::sleep_for(std::chrono::milliseconds(16));
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
   if (argc == 1) {
     RunTests();
@@ -775,6 +905,7 @@ int main(int argc, char** argv) {
     LanguageServerMain(argv[0]);
     return 0;
   }
+  /* TODO: out of process querydb -- maybe?
   else if (HasOption(options, "--querydb")) {
     std::cerr << "Running querydb" << std::endl;
     QueryableDatabase db;
@@ -782,10 +913,17 @@ int main(int argc, char** argv) {
     while (true) {
       QueryDbMainLoop(&ipc, &db);
       // TODO: use a condition variable.
-      std::this_thread::sleep_for(std::chrono::milliseconds(20));
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
     }
     return 0;
   }
+  */
+  else if (HasOption(options, "--indexer")) {
+    int index = atoi(options["--indexer"].c_str());
+    if (index == 0)
+      std::cerr << "--indexer expects an indexer id > 0" << std::endl;
+    IndexMain(index);
+  }
   else {
     std::cerr << "Running language server" << std::endl;
     LanguageServerMain(argv[0]);

compilation_database_loader.cc

@@ -39,7 +39,7 @@ std::vector<CompilationEntry> LoadFromDirectoryListing(const std::string& projec
       CompilationEntry entry;
       entry.directory = ".";
       entry.filename = file;
-      entry.args = {};
+      entry.args = args;
       result.push_back(entry);
     }
   }

indexer.cpp

@@ -630,7 +630,7 @@ clang::VisiterResult AddDeclInitializerUsagesVisitor(clang::Cursor cursor, clang
     VarId ref_id = db->ToVarId(ref_usr);
     IndexedVarDef* ref_def = db->Resolve(ref_id);
     Location loc = db->id_cache.Resolve(cursor, false /*interesting*/);
-    std::cerr << "Adding usage to id=" << ref_id.id << " usr=" << ref_usr << " at " << loc.ToString() << std::endl;
+    //std::cerr << "Adding usage to id=" << ref_id.id << " usr=" << ref_usr << " at " << loc.ToString() << std::endl;
     AddUsage(ref_def->uses, loc);
     break;
   }
@@ -1117,25 +1117,6 @@ void indexEntityReference(CXClientData client_data, const CXIdxEntityRefInfo* re
 void emptyIndexDeclaration(CXClientData client_data, const CXIdxDeclInfo* decl) {}
 void emptyIndexEntityReference(CXClientData client_data, const CXIdxEntityRefInfo* ref) {}
 
-struct Timer {
-  using Clock = std::chrono::high_resolution_clock;
-  std::chrono::time_point<Clock> start_;
-
-  Timer() {
-    Reset();
-  }
-
-  void Reset() {
-    start_ = Clock::now();
-  }
-
-  void PrintElapsed() {
-    std::chrono::time_point<Clock> end = Clock::now();
-
-    std::cerr << "Indexing took " << std::chrono::duration_cast<std::chrono::milliseconds>(end - start_).count() << "ms" << std::endl;
-  }
-};
-
 IndexedFile Parse(std::string filename, std::vector<std::string> args, bool dump_ast) {
   args.push_back("-std=c++11");
   args.push_back("-fms-compatibility");
@@ -1170,10 +1151,8 @@ IndexedFile Parse(std::string filename, std::vector<std::string> args, bool dump
   NamespaceHelper ns;
   IndexParam param(&db, &ns);
 
-  Timer time;
   clang_indexTranslationUnit(index_action, &param, callbacks, sizeof(callbacks),
     CXIndexOpt_IndexFunctionLocalSymbols | CXIndexOpt_SkipParsedBodiesInSession, tu.cx_tu);
-  time.PrintElapsed();
 
   clang_IndexAction_dispose(index_action);
 

indexer.h

@@ -13,6 +13,7 @@
 #include "bitfield.h"
 #include "utils.h"
 #include "optional.h"
+#include "serializer.h"
 
 #include <rapidjson/writer.h>
 #include <rapidjson/prettywriter.h>
@@ -297,6 +298,7 @@ struct TypeDefDefinitionData {
   std::vector<FuncId> funcs;
   std::vector<VarId> vars;
 
+  TypeDefDefinitionData() {} // For reflection.
   TypeDefDefinitionData(const std::string& usr) : usr(usr) {}
 
   bool operator==(const TypeDefDefinitionData<TypeId, FuncId, VarId, Location>& other) const {
@@ -314,6 +316,21 @@ struct TypeDefDefinitionData {
 
   bool operator!=(const TypeDefDefinitionData<TypeId, FuncId, VarId, Location>& other) const { return !(*this == other); }
 };
+template<typename TVisitor, typename TypeId, typename FuncId, typename VarId, typename Location>
+void Reflect(TVisitor& visitor, TypeDefDefinitionData<TypeId, FuncId, VarId, Location>& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(usr);
+  REFLECT_MEMBER(short_name);
+  REFLECT_MEMBER(qualified_name);
+  REFLECT_MEMBER(definition);
+  REFLECT_MEMBER(alias_of);
+  REFLECT_MEMBER(parents);
+  REFLECT_MEMBER(types);
+  REFLECT_MEMBER(funcs);
+  REFLECT_MEMBER(vars);
+  REFLECT_MEMBER_END();
+}
+
 
 struct IndexedTypeDef {
   TypeDefDefinitionData<> def;
@@ -367,6 +384,7 @@ struct FuncDefDefinitionData {
   // Functions that this function calls.
   std::vector<FuncRef> callees;
 
+  FuncDefDefinitionData() {} // For reflection.
   FuncDefDefinitionData(const std::string& usr) : usr(usr) {
     //assert(usr.size() > 0);
   }
@@ -385,6 +403,20 @@ struct FuncDefDefinitionData {
   bool operator!=(const FuncDefDefinitionData<TypeId, FuncId, VarId, FuncRef, Location>& other) const { return !(*this == other); }
 };
 
+template<typename TVisitor, typename TypeId, typename FuncId, typename VarId, typename FuncRef, typename Location>
+void Reflect(TVisitor& visitor, FuncDefDefinitionData<TypeId, FuncId, VarId, FuncRef, Location>& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(usr);
+  REFLECT_MEMBER(short_name);
+  REFLECT_MEMBER(qualified_name);
+  REFLECT_MEMBER(definition);
+  REFLECT_MEMBER(declaring_type);
+  REFLECT_MEMBER(base);
+  REFLECT_MEMBER(locals);
+  REFLECT_MEMBER(callees);
+  REFLECT_MEMBER_END();
+}
+
 struct IndexedFuncDef {
   FuncDefDefinitionData<> def;
 
@@ -409,8 +441,7 @@ struct IndexedFuncDef {
 
   bool is_bad_def = true;
 
-  IndexedFuncDef() : def("") {} // For serialization
-
+  IndexedFuncDef() {} // For reflection.
   IndexedFuncDef(FuncId id, const std::string& usr) : id(id), def(usr) {
     //assert(usr.size() > 0);
   }
@@ -446,6 +477,7 @@ struct VarDefDefinitionData {
   // Type which declares this one (ie, it is a method)
   optional<TypeId> declaring_type;
 
+  VarDefDefinitionData() {} // For reflection.
   VarDefDefinitionData(const std::string& usr) : usr(usr) {}
 
   bool operator==(const VarDefDefinitionData<TypeId, FuncId, VarId, Location>& other) const {
@@ -461,6 +493,18 @@ struct VarDefDefinitionData {
   bool operator!=(const VarDefDefinitionData<TypeId, FuncId, VarId, Location>& other) const { return !(*this == other); }
 };
 
+template<typename TVisitor, typename TypeId, typename FuncId, typename VarId, typename Location>
+void Reflect(TVisitor& visitor, VarDefDefinitionData<TypeId, FuncId, VarId, Location>& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(usr);
+  REFLECT_MEMBER(short_name);
+  REFLECT_MEMBER(qualified_name);
+  REFLECT_MEMBER(definition);
+  REFLECT_MEMBER(variable_type);
+  REFLECT_MEMBER(declaring_type);
+  REFLECT_MEMBER_END();
+}
+
 struct IndexedVarDef {
   VarDefDefinitionData<> def;
 

ipc.cc

@@ -1,5 +1,6 @@
 #include "ipc.h"
 #include "serializer.h"
+#include "utils.h"
 
 namespace {
   // The absolute smallest partial payload we should send. This must be >0, ie, 1 is the
@@ -230,6 +231,7 @@ void IpcDispatch(PlatformMutex* mutex, std::function<DispatchResult()> action) {
         std::cerr << "[info]: shmem full, waiting (" << log_count++ << ")" << std::endl; // TODO: remove
       }
       ++log_iteration_count;
+      // TODO: See if we can figure out a way to use condition variables cross-process.
       std::this_thread::sleep_for(std::chrono::microseconds(0));
     }
     first = false;
@@ -263,6 +265,7 @@ void IpcDirectionalChannel::PushMessage(IpcMessage* message) {
   int partial_message_id = 0; // TODO
 
   std::cerr << "Starting dispatch of payload with size " << payload_size << std::endl;
+  int count = 0;
 
   IpcDispatch(mutex.get(), [&]() {
     assert(payload_size > 0);
@@ -280,7 +283,11 @@ void IpcDirectionalChannel::PushMessage(IpcMessage* message) {
       shared_buffer->free_message()->Setup(message->ipc_id, partial_message_id, true /*has_more_chunks*/, sent_payload_size, payload);
       shared_buffer->metadata()->bytes_used += sizeof(JsonMessage) + sent_payload_size;
       shared_buffer->free_message()->ipc_id = IpcId::Invalid;
-      std::cerr << "Sending partial message with payload_size=" << sent_payload_size << std::endl;
+
+      if (count++ > 50) {
+        std::cerr << "x50 Sending partial message with payload_size=" << sent_payload_size << std::endl;
+        count = 0;
+      }
 
       // Prepare for next time.
       payload_size -= sent_payload_size;
@@ -293,7 +300,7 @@ void IpcDirectionalChannel::PushMessage(IpcMessage* message) {
       shared_buffer->free_message()->Setup(message->ipc_id, partial_message_id, false /*has_more_chunks*/, payload_size, payload);
       shared_buffer->metadata()->bytes_used += sizeof(JsonMessage) + payload_size;
       shared_buffer->free_message()->ipc_id = IpcId::Invalid;
-      std::cerr << "Sending full message with payload_size=" << payload_size << std::endl;
+      //std::cerr << "Sending full message with payload_size=" << payload_size << std::endl;
 
       return DispatchResult::Break;
     }
@@ -312,55 +319,53 @@ void AddIpcMessageFromJsonMessage(std::vector<std::unique_ptr<IpcMessage>>& resu
 }
 
 std::vector<std::unique_ptr<IpcMessage>> IpcDirectionalChannel::TakeMessages() {
-  size_t remaining_bytes = 0;
-  // Move data from shared memory into a local buffer. Do this
-  // before parsing the blocks so that other processes can begin
-  // posting data as soon as possible.
-  {
-    std::unique_ptr<PlatformScopedMutexLock> lock = CreatePlatformScopedMutexLock(mutex.get());
-    assert(shared_buffer->metadata()->bytes_used <= shmem_size);
-    remaining_bytes = shared_buffer->metadata()->bytes_used;
-
-    memcpy(local.get(), shared->shared, sizeof(MessageBuffer::Metadata) + shared_buffer->metadata()->bytes_used);
-    shared_buffer->metadata()->bytes_used = 0;
-    shared_buffer->free_message()->ipc_id = IpcId::Invalid;
-  }
-
   std::vector<std::unique_ptr<IpcMessage>> result;
 
-  for (JsonMessage* message : *local_buffer) {
-    std::cerr << "Got message with payload_size=" << message->payload_size << std::endl;
+  do {
+    // Move data from shared memory into a local buffer. Do this
+    // before parsing the blocks so that other processes can begin
+    // posting data as soon as possible.
+    {
+      std::unique_ptr<PlatformScopedMutexLock> lock = CreatePlatformScopedMutexLock(mutex.get());
+      assert(shared_buffer->metadata()->bytes_used <= shmem_size);
+      memcpy(local.get(), shared->shared, sizeof(MessageBuffer::Metadata) + shared_buffer->metadata()->bytes_used);
+      shared_buffer->metadata()->bytes_used = 0;
+      shared_buffer->free_message()->ipc_id = IpcId::Invalid;
+    }
 
-    if (message->partial_message_id != 0) {
-      auto* buf = CreateOrFindResizableBuffer(message->partial_message_id);
-      buf->Append(message->payload(), message->payload_size);
-      if (!message->has_more_chunks) {
-        AddIpcMessageFromJsonMessage(result, message->ipc_id, buf->memory, buf->size);
-        RemoveResizableBuffer(message->partial_message_id);
+    // Parse blocks from shared memory.
+    for (JsonMessage* message : *local_buffer) {
+      //std::cerr << "Got message with payload_size=" << message->payload_size << std::endl;
+
+      if (message->partial_message_id != 0) {
+        auto* buf = CreateOrFindResizableBuffer(message->partial_message_id);
+        buf->Append(message->payload(), message->payload_size);
+        if (!message->has_more_chunks) {
+          AddIpcMessageFromJsonMessage(result, message->ipc_id, buf->memory, buf->size);
+          RemoveResizableBuffer(message->partial_message_id);
+        }
+      }
+      else {
+        assert(!message->has_more_chunks);
+        AddIpcMessageFromJsonMessage(result, message->ipc_id, message->payload(), message->payload_size);
       }
     }
-    else {
-      assert(!message->has_more_chunks);
-      AddIpcMessageFromJsonMessage(result, message->ipc_id, message->payload(), message->payload_size);
-    }
-  }
-  local_buffer->metadata()->bytes_used = 0;
+    local_buffer->metadata()->bytes_used = 0;
+
+    // Let other threads run. We still want to run as fast as possible, though.
+    std::this_thread::sleep_for(std::chrono::microseconds(0));
+  } while (resizable_buffers.size() > 0);
 
   return result;
 }
 
 
 
-IpcServer::IpcServer(const std::string& name)
-  : name_(name), server_(NameToServerName(name)) {}
+IpcServer::IpcServer(const std::string& name, int client_id)
+  : server_(NameToServerName(name)), client_(NameToClientName(name, client_id)) {}
 
-void IpcServer::SendToClient(int client_id, IpcMessage* message) {
-  // Find or create the client.
-  auto it = clients_.find(client_id);
-  if (it == clients_.end())
-    clients_[client_id] = MakeUnique<IpcDirectionalChannel>(NameToClientName(name_, client_id));
-
-  clients_[client_id]->PushMessage(message);
+void IpcServer::SendToClient(IpcMessage* message) {
+  client_.PushMessage(message);
 }
 
 std::vector<std::unique_ptr<IpcMessage>> IpcServer::TakeMessages() {

ipc.h

@@ -23,6 +23,9 @@ enum class IpcId : int {
   IsAlive,
   OpenProject,
 
+  IndexTranslationUnitRequest,
+  IndexTranslationUnitResponse,
+
   // This is a language server request. The actual request method
   // id is embedded within the request state.
   LanguageServerRequest,
@@ -111,15 +114,14 @@ struct IpcDirectionalChannel {
 };
 
 struct IpcServer {
-  IpcServer(const std::string& name);
+  IpcServer(const std::string& name, int client_id);
 
-  void SendToClient(int client_id, IpcMessage* message);
+  void SendToClient(IpcMessage* message);
   std::vector<std::unique_ptr<IpcMessage>> TakeMessages();
 
 private:
-  std::string name_;
   IpcDirectionalChannel server_;
-  std::unordered_map<int, std::unique_ptr<IpcDirectionalChannel>> clients_;
+  IpcDirectionalChannel client_;
 };
 
 struct IpcClient {

platform.h

@@ -14,7 +14,7 @@ struct PlatformSharedMemory {
   void* shared;
 };
 
-const int shmem_size = 200;// 1024 * 1024 * 32;  // number of chars/bytes (32mb)
+const int shmem_size = 1024 * 1024 * 32;  // number of chars/bytes (32mb)
 
 std::unique_ptr<PlatformMutex> CreatePlatformMutex(const std::string& name);
 std::unique_ptr<PlatformScopedMutexLock> CreatePlatformScopedMutexLock(PlatformMutex* mutex);

query.h

@@ -59,6 +59,15 @@ struct QueryableLocation {
   }
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, QueryableLocation& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(path);
+  REFLECT_MEMBER(line);
+  REFLECT_MEMBER(column);
+  REFLECT_MEMBER(interesting);
+  REFLECT_MEMBER_END();
+}
 
 struct UsrRef {
   Usr usr;
@@ -76,7 +85,13 @@ struct UsrRef {
   }
 };
 
-
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, UsrRef& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(usr);
+  REFLECT_MEMBER(loc);
+  REFLECT_MEMBER_END();
+}
 
 // There are two sources of reindex updates: the (single) definition of a
 // symbol has changed, or one of many users of the symbol has changed.
@@ -96,10 +111,20 @@ struct MergeableUpdate {
   std::vector<TValue> to_add;
   std::vector<TValue> to_remove;
 
+  MergeableUpdate() {} // For reflection
   MergeableUpdate(Usr usr, const std::vector<TValue>& to_add, const std::vector<TValue>& to_remove)
     : usr(usr), to_add(to_add), to_remove(to_remove) {}
 };
 
+template<typename TVisitor, typename TValue>
+void Reflect(TVisitor& visitor, MergeableUpdate<TValue>& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(usr);
+  REFLECT_MEMBER(to_add);
+  REFLECT_MEMBER(to_remove);
+  REFLECT_MEMBER_END();
+}
+
 struct QueryableFile {
   using OutlineUpdate = MergeableUpdate<UsrRef>;
 
@@ -107,9 +132,18 @@ struct QueryableFile {
   // Outline of the file (ie, all symbols).
   std::vector<UsrRef> outline;
 
+  QueryableFile() {} // For serialization.
   QueryableFile(const IndexedFile& indexed);
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, QueryableFile& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(file_id);
+  REFLECT_MEMBER(outline);
+  REFLECT_MEMBER_END();
+}
+
 struct QueryableTypeDef {
   using DefUpdate = TypeDefDefinitionData<Usr, Usr, Usr, QueryableLocation>;
   using DerivedUpdate = MergeableUpdate<Usr>;
@@ -119,9 +153,19 @@ struct QueryableTypeDef {
   std::vector<Usr> derived;
   std::vector<QueryableLocation> uses;
 
+  QueryableTypeDef() : def("") {} // For serialization.
   QueryableTypeDef(IdCache& id_cache, const IndexedTypeDef& indexed);
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, QueryableTypeDef& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(def);
+  REFLECT_MEMBER(derived);
+  REFLECT_MEMBER(uses);
+  REFLECT_MEMBER_END();
+}
+
 struct QueryableFuncDef {
   using DefUpdate = FuncDefDefinitionData<Usr, Usr, Usr, UsrRef, QueryableLocation>;
   using DeclarationsUpdate = MergeableUpdate<QueryableLocation>;
@@ -135,9 +179,21 @@ struct QueryableFuncDef {
   std::vector<UsrRef> callers;
   std::vector<QueryableLocation> uses;
 
+  QueryableFuncDef() : def("") {} // For serialization.
   QueryableFuncDef(IdCache& id_cache, const IndexedFuncDef& indexed);
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, QueryableFuncDef& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(def);
+  REFLECT_MEMBER(declarations);
+  REFLECT_MEMBER(derived);
+  REFLECT_MEMBER(callers);
+  REFLECT_MEMBER(uses);
+  REFLECT_MEMBER_END();
+}
+
 struct QueryableVarDef {
   using DefUpdate = VarDefDefinitionData<Usr, Usr, Usr, QueryableLocation>;
   using UsesUpdate = MergeableUpdate<QueryableLocation>;
@@ -145,9 +201,18 @@ struct QueryableVarDef {
   DefUpdate def;
   std::vector<QueryableLocation> uses;
 
+  QueryableVarDef() : def("") {} // For serialization.
   QueryableVarDef(IdCache& id_cache, const IndexedVarDef& indexed);
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, QueryableVarDef& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(def);
+  REFLECT_MEMBER(uses);
+  REFLECT_MEMBER_END();
+}
+
 enum class SymbolKind { Invalid, File, Type, Func, Var };
 struct SymbolIdx {
   SymbolKind kind;
@@ -219,6 +284,30 @@ struct IndexUpdate {
   void Merge(const IndexUpdate& update);
 };
 
+template<typename TVisitor>
+void Reflect(TVisitor& visitor, IndexUpdate& value) {
+  REFLECT_MEMBER_START();
+  REFLECT_MEMBER(files_removed);
+  REFLECT_MEMBER(files_added);
+  REFLECT_MEMBER(files_outline);
+  REFLECT_MEMBER(types_removed);
+  REFLECT_MEMBER(types_added);
+  REFLECT_MEMBER(types_def_changed);
+  REFLECT_MEMBER(types_derived);
+  REFLECT_MEMBER(types_uses);
+  REFLECT_MEMBER(funcs_removed);
+  REFLECT_MEMBER(funcs_added);
+  REFLECT_MEMBER(funcs_def_changed);
+  REFLECT_MEMBER(funcs_declarations);
+  REFLECT_MEMBER(funcs_derived);
+  REFLECT_MEMBER(funcs_callers);
+  REFLECT_MEMBER(funcs_uses);
+  REFLECT_MEMBER(vars_removed);
+  REFLECT_MEMBER(vars_added);
+  REFLECT_MEMBER(vars_def_changed);
+  REFLECT_MEMBER(vars_uses);
+  REFLECT_MEMBER_END();
+}
 
 
 // The query database is heavily optimized for fast queries. It is stored

serializer.cc

@@ -1,5 +1,7 @@
 #include "serializer.h"
 
+#include "indexer.h"
+
 
 // int
 void Reflect(Reader& visitor, int& value) {

serializer.h

@@ -2,8 +2,13 @@
 
 #include <rapidjson/document.h>
 #include <rapidjson/prettywriter.h>
+#include <vector>
+#include <string>
 
-#include "indexer.h"
+#include "optional.h"
+
+using std::experimental::optional;
+using std::experimental::nullopt;
 
 using Reader = rapidjson::GenericValue<rapidjson::UTF8<>>;
 using Writer = rapidjson::PrettyWriter<rapidjson::StringBuffer>;