This commit is contained in:
Jacob Dufault 2017-03-14 21:59:05 -07:00
parent 24eb626cce
commit 72776d4c11
8 changed files with 296 additions and 153 deletions

View File

@ -21,6 +21,11 @@
#include <fcntl.h>
#endif
const char* kIpcIndexerName = "indexer";
const char* kIpcLanguageClientName = "language_client";
const int kNumIndexers = 8 - 1;
std::unordered_map<std::string, std::string> ParseOptions(int argc, char** argv) {
std::unordered_map<std::string, std::string> output;
@ -204,6 +209,7 @@ struct IpcMessage_IndexTranslationUnitResponse : public BaseIpcMessage<IpcMessag
static constexpr IpcId kIpcId = IpcId::IndexTranslationUnitResponse;
IndexUpdate update;
IpcMessage_IndexTranslationUnitResponse() {}
explicit IpcMessage_IndexTranslationUnitResponse(IndexUpdate& update) : update(update) {}
};
template<typename TVisitor>
@ -310,10 +316,10 @@ struct Timer {
};
void IndexMainLoop(IpcClient* ipc) {
std::vector<std::unique_ptr<IpcMessage>> messages = ipc->TakeMessages();
void IndexMainLoop(IpcClient* client) {
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;
//std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
switch (message->ipc_id) {
case IpcId::Quit: {
@ -335,7 +341,7 @@ void IndexMainLoop(IpcClient* ipc) {
std::cerr << "Creating index update took " << time.ElapsedMilliseconds() << "ms" << std::endl;
time.Reset();
ipc->SendToServer(&response);
client->SendToServer(&response);
std::cerr << "Sending to server took " << time.ElapsedMilliseconds() << "ms" << std::endl;
break;
@ -345,10 +351,7 @@ void IndexMainLoop(IpcClient* ipc) {
}
void IndexMain(int id) {
return;
IpcClient client_ipc("indexer", id);
IpcClient client_ipc(kIpcIndexerName, id);
while (true) {
IndexMainLoop(&client_ipc);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
@ -367,11 +370,41 @@ void IndexMain(int id) {
void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>& indexers, QueryableDatabase* db) {
std::vector<std::unique_ptr<IpcMessage>> messages = client->TakeMessages();
void QueryDbMainLoop(IpcServer* language_client, IpcServer* indexers, QueryableDatabase* db) {
std::vector<std::unique_ptr<IpcMessage>> messages = language_client->TakeMessages();
for (auto& message : messages) {
std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
//std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
switch (message->ipc_id) {
case IpcId::Quit: {
@ -380,7 +413,7 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
case IpcId::IsAlive: {
IpcMessage_IsAlive response;
client->SendToClient(&response); // todo: make non-blocking
language_client->SendToClient(0, &response); // todo: make non-blocking
break;
}
@ -395,11 +428,11 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
std::string filepath = path + "/" + entry.filename;
std::cerr << "[" << i << "/" << (entries.size() - 1) << "] Dispatching index request for file " << filepath << std::endl;
// TODO: indexers should steal work. load balance.
// TODO: indexers should steal work and load balance.
IpcMessage_IndexTranslationUnitRequest request;
request.path = filepath;
request.args = entry.args;
indexers[i % indexers.size()]->SendToClient(&request);
indexers->SendToClient(i % indexers->num_clients(), &request);
//IndexedFile file = Parse(filepath, entry.args);
//IndexUpdate update(file);
//db->ApplyIndexUpdate(&update);
@ -408,14 +441,6 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
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());
@ -487,7 +512,7 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
client->SendToClient(&response);
language_client->SendToClient(0, &response);
break;
}
@ -577,7 +602,27 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
}
client->SendToClient(&response);
language_client->SendToClient(0, &response);
break;
}
default: {
std::cerr << "Unhandled IPC message with kind " << static_cast<int>(message->ipc_id) << std::endl;
exit(1);
}
}
}
messages = indexers->TakeMessages();
for (auto& message : messages) {
//std::cerr << "Processing message " << static_cast<int>(message->ipc_id) << std::endl;
switch (message->ipc_id) {
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;
}
@ -588,6 +633,54 @@ void QueryDbMainLoop(IpcServer* client, std::vector<std::unique_ptr<IpcServer>>&
}
}
}
void QueryDbMain() {
std::cerr << "Running QueryDb" << std::endl;
IpcServer language_client(kIpcLanguageClientName, 1 /*num_clients*/);
IpcServer indexers(kIpcIndexerName, kNumIndexers);
QueryableDatabase db;
std::cerr << "!! starting processes" << std::endl;
// Start indexer processes.
for (int i = 0; i < kNumIndexers; ++i) {
//new Process(process_name + " --indexer " + std::to_string(i + 1));
new std::thread([i]() {
IndexMain(i);
});
}
std::cerr << "!! done processes" << std::endl;
// Pump query db main loop.
while (true) {
QueryDbMainLoop(&language_client, &indexers, &db);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
@ -722,10 +815,8 @@ void LanguageServerMainLoop(IpcClient* ipc) {
}
}
const int kNumIndexers = 8 - 1;
void LanguageServerMain(std::string process_name) {
IpcClient client_ipc("languageserver", 0);
IpcClient client_ipc(kIpcLanguageClientName, 0);
// Discard any left-over messages from previous runs.
client_ipc.TakeMessages();
@ -767,109 +858,77 @@ void LanguageServerMain(std::string process_name) {
// Pass empty process name so we only try to start the querydb once.
LanguageServerMain("");
return;
}
}
#endif
// for debugging attach
//std::this_thread::sleep_for(std::chrono::seconds(4));
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([&]() {
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, server_indexers, &db);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
QueryDbMain();
});
}
// Run language client.
std::thread stdio_reader(&LanguageServerStdinLoop, &client_ipc);
while (true) {
LanguageServerMainLoop(&client_ipc);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
int main(int argc, char** argv) {
bool loop = false;
while (loop)
std::this_thread::sleep_for(std::chrono::milliseconds(10));
if (argc == 1) {
RunTests();
return 0;
}
void PreMain() {
// We need to write to stdout in binary mode because in Windows, writing
// \n will implicitly write \r\n. Language server API will ignore a
// \r\r\n split request.
@ -878,21 +937,40 @@ int main(int argc, char** argv) {
_setmode(_fileno(stdin), O_BINARY);
#endif
IpcRegistry::instance()->Register<IpcMessage_Quit>(IpcId::Quit);
IpcRegistry::instance()->Register<IpcMessage_IsAlive>(IpcId::IsAlive);
IpcRegistry::instance()->Register<IpcMessage_OpenProject>(IpcId::OpenProject);
IpcRegistry::instance()->Register<IpcMessage_Quit>(IpcMessage_Quit::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_IsAlive>(IpcMessage_IsAlive::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_OpenProject>(IpcMessage_OpenProject::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_DocumentSymbolsRequest>(IpcId::DocumentSymbolsRequest);
IpcRegistry::instance()->Register<IpcMessage_DocumentSymbolsResponse>(IpcId::DocumentSymbolsResponse);
IpcRegistry::instance()->Register<IpcMessage_WorkspaceSymbolsRequest>(IpcId::WorkspaceSymbolsRequest);
IpcRegistry::instance()->Register<IpcMessage_WorkspaceSymbolsResponse>(IpcId::WorkspaceSymbolsResponse);
IpcRegistry::instance()->Register<IpcMessage_IndexTranslationUnitRequest>(IpcMessage_IndexTranslationUnitRequest::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_IndexTranslationUnitResponse>(IpcMessage_IndexTranslationUnitResponse::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_DocumentSymbolsRequest>(IpcMessage_DocumentSymbolsRequest::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_DocumentSymbolsResponse>(IpcMessage_DocumentSymbolsResponse::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_WorkspaceSymbolsRequest>(IpcMessage_WorkspaceSymbolsRequest::kIpcId);
IpcRegistry::instance()->Register<IpcMessage_WorkspaceSymbolsResponse>(IpcMessage_WorkspaceSymbolsResponse::kIpcId);
MessageRegistry::instance()->Register<In_CancelRequest>();
MessageRegistry::instance()->Register<In_InitializeRequest>();
MessageRegistry::instance()->Register<In_InitializedNotification>();
MessageRegistry::instance()->Register<In_DocumentSymbolRequest>();
MessageRegistry::instance()->Register<In_WorkspaceSymbolRequest>();
}
int main(int argc, char** argv) {
bool loop = false;
while (loop)
std::this_thread::sleep_for(std::chrono::milliseconds(10));
PreMain();
if (argc == 1) {
QueryDbMain();
return 0;
}
if (argc == 1) {
RunTests();
return 0;
}
@ -909,7 +987,7 @@ int main(int argc, char** argv) {
else if (HasOption(options, "--querydb")) {
std::cerr << "Running querydb" << std::endl;
QueryableDatabase db;
IpcServer ipc("languageserver");
IpcServer ipc(kIpcServername);
while (true) {
QueryDbMainLoop(&ipc, &db);
// TODO: use a condition variable.

View File

@ -0,0 +1,9 @@
#include "a.h"
#if RANDOM_DEFINE
static void LocalC() {
Common();
}
#endif // RANDOM_DEFINE
static void LocalD() {}

32
ipc.cc
View File

@ -45,14 +45,18 @@ namespace {
IpcRegistry* IpcRegistry::instance_ = nullptr;
std::unique_ptr<IpcMessage> IpcRegistry::Allocate(IpcId id) {
return std::unique_ptr<IpcMessage>((*allocators)[id]());
assert(allocators_);
auto it = allocators_->find(id);
assert(it != allocators_->end() && "No registered allocator for id");
return std::unique_ptr<IpcMessage>(it->second());
}
struct IpcDirectionalChannel::MessageBuffer {
MessageBuffer(void* buffer, size_t buffer_size) {
MessageBuffer(void* buffer, size_t buffer_size, bool initialize) {
real_buffer = buffer;
real_buffer_size = buffer_size;
new(real_buffer) Metadata();
if (initialize)
new(real_buffer) Metadata();
}
// Pointer to the start of the actual buffer and the
@ -201,15 +205,15 @@ void IpcDirectionalChannel::RemoveResizableBuffer(int id) {
resizable_buffers.erase(id);
}
IpcDirectionalChannel::IpcDirectionalChannel(const std::string& name) {
IpcDirectionalChannel::IpcDirectionalChannel(const std::string& name, bool initialize_shared_memory) {
shared = CreatePlatformSharedMemory(name + "memory");
mutex = CreatePlatformMutex(name + "mutex");
local = std::unique_ptr<char>(new char[shmem_size]);
// TODO: connecting a client will allocate reset shared state on the
// buffer. We need to store if we "initialized".
shared_buffer = MakeUnique<MessageBuffer>(shared->shared, shmem_size);
local_buffer = MakeUnique<MessageBuffer>(local.get(), shmem_size);
shared_buffer = MakeUnique<MessageBuffer>(shared->shared, shmem_size, initialize_shared_memory);
local_buffer = MakeUnique<MessageBuffer>(local.get(), shmem_size, true /*initialize*/);
}
IpcDirectionalChannel::~IpcDirectionalChannel() {}
@ -361,11 +365,16 @@ std::vector<std::unique_ptr<IpcMessage>> IpcDirectionalChannel::TakeMessages() {
IpcServer::IpcServer(const std::string& name, int client_id)
: server_(NameToServerName(name)), client_(NameToClientName(name, client_id)) {}
IpcServer::IpcServer(const std::string& name, int num_clients)
: server_(NameToServerName(name), true /*initialize_shared_memory*/) {
void IpcServer::SendToClient(IpcMessage* message) {
client_.PushMessage(message);
for (int i = 0; i < num_clients; ++i) {
clients_.push_back(MakeUnique<IpcDirectionalChannel>(NameToClientName(name, i), true /*initialize_shared_memory*/));
}
}
void IpcServer::SendToClient(int client_id, IpcMessage* message) {
clients_[client_id]->PushMessage(message);
}
std::vector<std::unique_ptr<IpcMessage>> IpcServer::TakeMessages() {
@ -373,7 +382,8 @@ std::vector<std::unique_ptr<IpcMessage>> IpcServer::TakeMessages() {
}
IpcClient::IpcClient(const std::string& name, int client_id)
: server_(NameToServerName(name)), client_(NameToClientName(name, client_id)) {}
: server_(NameToServerName(name), false /*initialize_shared_memory*/),
client_(NameToClientName(name, client_id), false /*initialize_shared_memory*/) {}
void IpcClient::SendToServer(IpcMessage* message) {
server_.PushMessage(message);

22
ipc.h
View File

@ -61,7 +61,7 @@ struct IpcRegistry {
// Use unique_ptrs so we can initialize on first use
// (static init order might not be right).
std::unique_ptr<std::unordered_map<IpcId, Allocator>> allocators;
std::unique_ptr<std::unordered_map<IpcId, Allocator>> allocators_;
template<typename T>
void Register(IpcId id);
@ -78,13 +78,13 @@ struct IpcRegistry {
template<typename T>
void IpcRegistry::Register(IpcId id) {
if (!allocators)
allocators = MakeUnique<std::unordered_map<IpcId, Allocator>>();
if (!allocators_)
allocators_ = MakeUnique<std::unordered_map<IpcId, Allocator>>();
assert(allocators->find(id) == allocators->end() &&
assert(allocators_->find(id) == allocators_->end() &&
"There is already an IPC message with the given id");
(*allocators)[id] = [id]() {
(*allocators_)[id] = [id]() {
return new T();
};
}
@ -98,7 +98,7 @@ struct IpcDirectionalChannel {
// NOTE: We keep all pointers in terms of char* so pointer arithmetic is
// always relative to bytes.
explicit IpcDirectionalChannel(const std::string& name);
explicit IpcDirectionalChannel(const std::string& name, bool initialize_shared_memory);
~IpcDirectionalChannel();
void PushMessage(IpcMessage* message);
@ -123,14 +123,16 @@ struct IpcDirectionalChannel {
};
struct IpcServer {
IpcServer(const std::string& name, int client_id);
IpcServer(const std::string& name, int num_clients);
void SendToClient(IpcMessage* message);
void SendToClient(int client_id, IpcMessage* message);
std::vector<std::unique_ptr<IpcMessage>> TakeMessages();
int num_clients() const { return clients_.size(); }
private:
IpcDirectionalChannel server_;
IpcDirectionalChannel client_;
IpcDirectionalChannel server_; // Local / us.
std::vector<std::unique_ptr<IpcDirectionalChannel>> clients_;
};
struct IpcClient {

View File

@ -12,6 +12,7 @@ struct PlatformScopedMutexLock {
struct PlatformSharedMemory {
virtual ~PlatformSharedMemory() {}
void* shared;
std::string name;
};
const int shmem_size = 1024 * 1024 * 32; // number of chars/bytes (32mb)

View File

@ -37,6 +37,8 @@ struct PlatformSharedMemoryWin : public PlatformSharedMemory {
HANDLE shmem_;
PlatformSharedMemoryWin(const std::string& name) {
this->name = name;
shmem_ = CreateFileMapping(
INVALID_HANDLE_VALUE,
NULL,

View File

@ -512,37 +512,77 @@ void QueryableDatabase::RemoveUsrs(const std::vector<Usr>& to_remove) {
void QueryableDatabase::Import(const std::vector<QueryableFile>& defs) {
for (auto& def : defs) {
qualified_names.push_back(def.file_id);
symbols.push_back(SymbolIdx(SymbolKind::File, files.size()));
usr_to_symbol[def.file_id] = SymbolIdx(SymbolKind::File, files.size());
files.push_back(def);
auto it = usr_to_symbol.find(def.file_id);
if (it == usr_to_symbol.end()) {
qualified_names.push_back(def.file_id);
symbols.push_back(SymbolIdx(SymbolKind::File, files.size()));
usr_to_symbol[def.file_id] = SymbolIdx(SymbolKind::File, files.size());
files.push_back(def);
}
else {
QueryableFile& existing = files[it->second.idx];
// Replace the entire file. We don't ever want to merge files.
existing = def;
}
}
}
void QueryableDatabase::Import(const std::vector<QueryableTypeDef>& defs) {
for (auto& def : defs) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Type, types.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Type, types.size());
types.push_back(def);
auto it = usr_to_symbol.find(def.def.usr);
if (it == usr_to_symbol.end()) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Type, types.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Type, types.size());
types.push_back(def);
}
else {
QueryableTypeDef& existing = types[it->second.idx];
if (def.def.definition)
existing.def = def.def;
AddRange(&existing.derived, def.derived);
AddRange(&existing.uses, def.uses);
}
}
}
void QueryableDatabase::Import(const std::vector<QueryableFuncDef>& defs) {
for (auto& def : defs) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Func, funcs.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Func, funcs.size());
funcs.push_back(def);
auto it = usr_to_symbol.find(def.def.usr);
if (it == usr_to_symbol.end()) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Func, funcs.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Func, funcs.size());
funcs.push_back(def);
}
else {
QueryableFuncDef& existing = funcs[it->second.idx];
if (def.def.definition)
existing.def = def.def;
AddRange(&existing.callers, def.callers);
AddRange(&existing.declarations, def.declarations);
AddRange(&existing.derived, def.derived);
AddRange(&existing.uses, def.uses);
}
}
}
void QueryableDatabase::Import(const std::vector<QueryableVarDef>& defs) {
for (auto& def : defs) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Var, vars.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Var, vars.size());
vars.push_back(def);
auto it = usr_to_symbol.find(def.def.usr);
if (it == usr_to_symbol.end()) {
qualified_names.push_back(def.def.qualified_name);
symbols.push_back(SymbolIdx(SymbolKind::Var, vars.size()));
usr_to_symbol[def.def.usr] = SymbolIdx(SymbolKind::Var, vars.size());
vars.push_back(def);
}
else {
QueryableVarDef& existing = vars[it->second.idx];
if (def.def.definition)
existing.def = def.def;
AddRange(&existing.uses, def.uses);
}
}
}

View File

@ -271,6 +271,7 @@ struct IndexUpdate {
std::vector<QueryableVarDef::DefUpdate> vars_def_changed;
std::vector<QueryableVarDef::UsesUpdate> vars_uses;
IndexUpdate() {}
// Creates a new IndexUpdate that will import |file|.
explicit IndexUpdate(IndexedFile& file);