Delay requests if the document has not not indexed (#176)

This fixes a plethora of "not indexed" errors when the document has not been indexed.

* Message handler throws NotIndexed if not overdue
* The message is put into backlog and tagged with backlog_path
* path2backlog[path] tracks backlog associated with document `path`
* The backlog is cleared when the index is merged
* backlog[0] is forced to run if it becomes overdue

src/config.hh

@@ -242,6 +242,12 @@ struct Config {
     std::vector<std::string> whitelist;
   } index;
 
+  struct Request {
+    // If the document of a request has not been indexed, wait up to this many
+    // milleseconds before reporting error.
+    int64_t timeout = 5000;
+  } request;
+
   struct Session {
     int maxNum = 10;
   } session;
@@ -278,12 +284,14 @@ REFLECT_STRUCT(Config::Index, blacklist, comments, initialBlacklist,
                initialWhitelist, multiVersion, multiVersionBlacklist,
                multiVersionWhitelist, onChange, threads, trackDependency,
                whitelist);
+REFLECT_STRUCT(Config::Request, timeout);
 REFLECT_STRUCT(Config::Session, maxNum);
 REFLECT_STRUCT(Config::WorkspaceSymbol, caseSensitivity, maxNum, sort);
 REFLECT_STRUCT(Config::Xref, maxNum);
 REFLECT_STRUCT(Config, compilationDatabaseCommand, compilationDatabaseDirectory,
                cacheDirectory, cacheFormat, clang, client, codeLens, completion,
-               diagnostics, highlight, index, session, workspaceSymbol, xref);
+               diagnostics, highlight, index, request, session, workspaceSymbol,
+               xref);
 
 extern Config *g_config;
 

src/lsp.hh

@@ -21,8 +21,8 @@ limitations under the License.
 
 #include <rapidjson/fwd.h>
 
+#include <chrono>
 #include <iosfwd>
-#include <unordered_map>
 
 namespace ccls {
 struct RequestId {
@@ -43,6 +43,8 @@ struct InMessage {
   std::string method;
   std::unique_ptr<char[]> message;
   std::unique_ptr<rapidjson::Document> document;
+  std::chrono::steady_clock::time_point deadline;
+  std::string backlog_path;
 };
 
 enum class ErrorCode {

src/message_handler.cc

@@ -109,11 +109,8 @@ struct ScanLineEvent {
 };
 } // namespace
 
-void ReplyOnce::NotReady(bool file) {
-  if (file)
-    Error(ErrorCode::InvalidRequest, "not opened");
-  else
-    Error(ErrorCode::InternalError, "not indexed");
+void ReplyOnce::NotOpened(std::string_view path) {
+  Error(ErrorCode::InvalidRequest, std::string(path) + " is not opened");
 }
 
 void ReplyOnce::ReplyLocationLink(std::vector<LocationLink> &result) {
@@ -215,13 +212,11 @@ MessageHandler::MessageHandler() {
 
 void MessageHandler::Run(InMessage &msg) {
   rapidjson::Document &doc = *msg.document;
-  rapidjson::Value param;
+  rapidjson::Value null;
   auto it = doc.FindMember("params");
-  if (it != doc.MemberEnd())
-    param = it->value;
-  JsonReader reader(&param);
+  JsonReader reader(it != doc.MemberEnd() ? &it->value : &null);
   if (msg.id.Valid()) {
-    ReplyOnce reply{msg.id};
+    ReplyOnce reply{*this, msg.id};
     auto it = method2request.find(msg.method);
     if (it != method2request.end()) {
       try {
@@ -230,6 +225,8 @@ void MessageHandler::Run(InMessage &msg) {
         reply.Error(ErrorCode::InvalidParams,
                     "invalid params of " + msg.method + ": expected " +
                         ex.what() + " for " + reader.GetPath());
+      } catch (NotIndexed &) {
+        throw;
       } catch (...) {
         reply.Error(ErrorCode::InternalError, "failed to process " + msg.method);
       }
@@ -249,7 +246,8 @@ void MessageHandler::Run(InMessage &msg) {
   }
 }
 
-QueryFile *MessageHandler::FindFile(const std::string &path, int *out_file_id) {
+QueryFile *MessageHandler::FindFile(const std::string &path,
+                                    int *out_file_id) {
   QueryFile *ret = nullptr;
   auto it = db->name2file_id.find(LowerPathIfInsensitive(path));
   if (it != db->name2file_id.end()) {
@@ -266,6 +264,24 @@ QueryFile *MessageHandler::FindFile(const std::string &path, int *out_file_id) {
   return ret;
 }
 
+std::pair<QueryFile *, WorkingFile *>
+MessageHandler::FindOrFail(const std::string &path, ReplyOnce &reply,
+                           int *out_file_id) {
+  WorkingFile *wf = wfiles->GetFile(path);
+  if (!wf) {
+    reply.NotOpened(path);
+    return {nullptr, nullptr};
+  }
+  QueryFile *file = FindFile(path, out_file_id);
+  if (!file) {
+    if (!overdue)
+      throw NotIndexed{path};
+    reply.Error(ErrorCode::InvalidRequest, "not indexed");
+    return {nullptr, nullptr};
+  }
+  return {file, wf};
+}
+
 void EmitSkippedRanges(WorkingFile *wfile, QueryFile &file) {
   CclsSetSkippedRanges params;
   params.uri = DocumentUri::FromPath(wfile->filename);

src/message_handler.hh

@@ -199,7 +199,13 @@ REFLECT_STRUCT(Diagnostic, range, severity, code, source, message);
 REFLECT_STRUCT(ShowMessageParam, type, message);
 REFLECT_UNDERLYING_B(LanguageId);
 
+struct NotIndexed {
+  std::string path;
+};
+struct MessageHandler;
+
 struct ReplyOnce {
+  MessageHandler &handler;
   RequestId id;
   template <typename Res> void operator()(Res &&result) const {
     if (id.Valid())
@@ -210,7 +216,7 @@ struct ReplyOnce {
     if (id.Valid())
       pipeline::ReplyError(id, [&](JsonWriter &w) { Reflect(w, err); });
   }
-  void NotReady(bool file);
+  void NotOpened(std::string_view path);
   void ReplyLocationLink(std::vector<LocationLink> &result);
 };
 
@@ -225,10 +231,14 @@ struct MessageHandler {
   llvm::StringMap<std::function<void(JsonReader &)>> method2notification;
   llvm::StringMap<std::function<void(JsonReader &, ReplyOnce &)>>
       method2request;
+  bool overdue = false;
 
   MessageHandler();
   void Run(InMessage &msg);
   QueryFile *FindFile(const std::string &path, int *out_file_id = nullptr);
+  std::pair<QueryFile *, WorkingFile *> FindOrFail(const std::string &path,
+                                                   ReplyOnce &reply,
+                                                   int *out_file_id = nullptr);
 
 private:
   void Bind(const char *method, void (MessageHandler::*handler)(JsonReader &));

src/messages/ccls_call.cc

@@ -200,8 +200,7 @@ void MessageHandler::ccls_call(JsonReader &reader, ReplyOnce &reply) {
       Expand(this, &*result, param.callee, param.callType, param.qualified,
              param.levels);
   } else {
-    QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-    WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+    auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
     if (!wf)
       return;
     for (SymbolRef sym : FindSymbolsAtLocation(wf, file, param.position)) {

src/messages/ccls_inheritance.cc

@@ -146,10 +146,10 @@ void Inheritance(MessageHandler *m, Param &param, ReplyOnce &reply) {
           Expand(m, &*result, param.derived, param.qualified, param.levels)))
       result.reset();
   } else {
-    QueryFile *file = m->FindFile(param.textDocument.uri.GetPath());
-    if (!file)
+    auto [file, wf] = m->FindOrFail(param.textDocument.uri.GetPath(), reply);
+    if (!wf) {
       return;
-    WorkingFile *wf = m->wfiles->GetFile(file->def->path);
+    }
     for (SymbolRef sym : FindSymbolsAtLocation(wf, file, param.position))
       if (sym.kind == Kind::Func || sym.kind == Kind::Type) {
         result = BuildInitial(m, sym, param.derived, param.qualified,

src/messages/ccls_member.cc

@@ -281,12 +281,9 @@ void MessageHandler::ccls_member(JsonReader &reader, ReplyOnce &reply) {
                                             param.levels, param.kind)))
       result.reset();
   } else {
-    QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-    WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-    if (!wf) {
-      reply.NotReady(file);
+    auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+    if (!wf)
       return;
-    }
     for (SymbolRef sym : FindSymbolsAtLocation(wf, file, param.position)) {
       switch (sym.kind) {
       case Kind::Func:

src/messages/ccls_navigate.cc

@@ -41,10 +41,8 @@ Maybe<Range> FindParent(QueryFile *file, Pos pos) {
 void MessageHandler::ccls_navigate(JsonReader &reader, ReplyOnce &reply) {
   Param param;
   Reflect(reader, param);
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
   Position ls_pos = param.position;

src/messages/ccls_vars.cc

@@ -31,10 +31,8 @@ REFLECT_STRUCT(Param, textDocument, position, kind);
 void MessageHandler::ccls_vars(JsonReader &reader, ReplyOnce &reply) {
   Param param;
   Reflect(reader, param);
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
 

src/messages/initialize.cc

@@ -388,7 +388,7 @@ void MessageHandler::initialize(JsonReader &reader, ReplyOnce &reply) {
 void StandaloneInitialize(MessageHandler &handler, const std::string &root) {
   InitializeParam param;
   param.rootUri = DocumentUri::FromPath(root);
-  ReplyOnce reply;
+  ReplyOnce reply{handler};
   Initialize(&handler, param, reply);
 }
 

src/messages/textDocument_code.cc

@@ -35,11 +35,9 @@ REFLECT_STRUCT(CodeAction, title, kind, edit);
 }
 void MessageHandler::textDocument_codeAction(CodeActionParam &param,
                                              ReplyOnce &reply) {
-  WorkingFile *wf = wfiles->GetFile(param.textDocument.uri.GetPath());
-  if (!wf) {
-    reply.NotReady(true);
+  WorkingFile *wf = FindOrFail(param.textDocument.uri.GetPath(), reply).second;
+  if (!wf)
     return;
-  }
   std::vector<CodeAction> result;
   std::vector<Diagnostic> diagnostics;
   wfiles->WithLock([&]() { diagnostics = wf->diagnostics; });
@@ -96,16 +94,13 @@ struct CommonCodeLensParams {
 
 void MessageHandler::textDocument_codeLens(TextDocumentParam &param,
                                            ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!wf)
     return;
-  }
 
   std::vector<CodeLens> result;
-  auto Add = [&](const char *singular, Cmd_xref show, Range range, int num,
-                 bool force_display = false) {
+  auto Add = [&, wf = wf](const char *singular, Cmd_xref show, Range range,
+                          int num, bool force_display = false) {
     if (!num && !force_display)
       return;
     std::optional<lsRange> ls_range = GetLsRange(wf, range);

src/messages/textDocument_completion.cc

@@ -459,7 +459,7 @@ void MessageHandler::textDocument_completion(CompletionParam &param,
   std::string path = param.textDocument.uri.GetPath();
   WorkingFile *wf = wfiles->GetFile(path);
   if (!wf) {
-    reply.NotReady(true);
+    reply.NotOpened(path);
     return;
   }
 

src/messages/textDocument_definition.cc

@@ -48,12 +48,9 @@ std::vector<DeclRef> GetNonDefDeclarationTargets(DB *db, SymbolRef sym) {
 void MessageHandler::textDocument_declaration(TextDocumentPositionParam &param,
                                               ReplyOnce &reply) {
   int file_id;
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath(), &file_id);
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply, &file_id);
+  if (!wf)
     return;
-  }
 
   std::vector<LocationLink> result;
   Position &ls_pos = param.position;
@@ -69,12 +66,9 @@ void MessageHandler::textDocument_declaration(TextDocumentPositionParam &param,
 void MessageHandler::textDocument_definition(TextDocumentPositionParam &param,
                                              ReplyOnce &reply) {
   int file_id;
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath(), &file_id);
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply, &file_id);
+  if (!wf)
     return;
-  }
 
   std::vector<LocationLink> result;
   Maybe<DeclRef> on_def;
@@ -190,12 +184,9 @@ void MessageHandler::textDocument_definition(TextDocumentPositionParam &param,
 
 void MessageHandler::textDocument_typeDefinition(
     TextDocumentPositionParam &param, ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!file) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!file)
     return;
-  }
 
   std::vector<LocationLink> result;
   auto Add = [&](const QueryType &type) {

src/messages/textDocument_document.cc

@@ -44,12 +44,9 @@ REFLECT_STRUCT(DocumentHighlight, range, kind, role);
 void MessageHandler::textDocument_documentHighlight(
     TextDocumentPositionParam &param, ReplyOnce &reply) {
   int file_id;
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath(), &file_id);
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply, &file_id);
+  if (!wf)
     return;
-  }
 
   std::vector<DocumentHighlight> result;
   std::vector<SymbolRef> syms =
@@ -90,10 +87,8 @@ REFLECT_STRUCT(DocumentLink, range, target);
 
 void MessageHandler::textDocument_documentLink(TextDocumentParam &param,
                                                ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
 
@@ -165,10 +160,8 @@ void MessageHandler::textDocument_documentSymbol(JsonReader &reader,
   Reflect(reader, param);
 
   int file_id;
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath(), &file_id);
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply, &file_id);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
 

src/messages/textDocument_foldingRange.cc

@@ -31,12 +31,9 @@ REFLECT_STRUCT(FoldingRange, startLine, startCharacter, endLine, endCharacter,
 
 void MessageHandler::textDocument_foldingRange(TextDocumentParam &param,
                                                ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!wf)
     return;
-  }
   std::vector<FoldingRange> result;
   std::optional<lsRange> ls_range;
 

src/messages/textDocument_formatting.cc

@@ -80,21 +80,16 @@ void Format(ReplyOnce &reply, WorkingFile *wfile, tooling::Range range) {
 
 void MessageHandler::textDocument_formatting(DocumentFormattingParam &param,
                                              ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!wf)
     return;
-  }
   Format(reply, wf, {0, (unsigned)wf->buffer_content.size()});
 }
 
 void MessageHandler::textDocument_onTypeFormatting(
     DocumentOnTypeFormattingParam &param, ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
   std::string_view code = wf->buffer_content;
@@ -107,10 +102,8 @@ void MessageHandler::textDocument_onTypeFormatting(
 
 void MessageHandler::textDocument_rangeFormatting(
     DocumentRangeFormattingParam &param, ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
   std::string_view code = wf->buffer_content;

src/messages/textDocument_hover.cc

@@ -93,12 +93,9 @@ GetHover(DB *db, LanguageId lang, SymbolRef sym, int file_id) {
 
 void MessageHandler::textDocument_hover(TextDocumentPositionParam &param,
                                         ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!wf)
     return;
-  }
 
   Hover result;
   for (SymbolRef sym : FindSymbolsAtLocation(wf, file, param.position)) {

src/messages/textDocument_references.cc

@@ -45,12 +45,9 @@ void MessageHandler::textDocument_references(JsonReader &reader,
                                              ReplyOnce &reply) {
   ReferenceParam param;
   Reflect(reader, param);
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
-  if (!wf) {
-    reply.NotReady(file);
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
+  if (!wf)
     return;
-  }
 
   for (auto &folder : param.folders)
     EnsureEndsInSlash(folder);

src/messages/textDocument_rename.cc

@@ -56,10 +56,8 @@ WorkspaceEdit BuildWorkspaceEdit(DB *db, WorkingFiles *wfiles, SymbolRef sym,
 } // namespace
 
 void MessageHandler::textDocument_rename(RenameParam &param, ReplyOnce &reply) {
-  QueryFile *file = FindFile(param.textDocument.uri.GetPath());
-  WorkingFile *wf = file ? wfiles->GetFile(file->def->path) : nullptr;
+  auto [file, wf] = FindOrFail(param.textDocument.uri.GetPath(), reply);
   if (!wf) {
-    reply.NotReady(file);
     return;
   }
 

src/messages/textDocument_signatureHelp.cc

@@ -152,12 +152,11 @@ public:
 void MessageHandler::textDocument_signatureHelp(
     TextDocumentPositionParam &param, ReplyOnce &reply) {
   static CompleteConsumerCache<SignatureHelp> cache;
-
-  std::string path = param.textDocument.uri.GetPath();
   Position begin_pos = param.position;
+  std::string path = param.textDocument.uri.GetPath();
   WorkingFile *wf = wfiles->GetFile(path);
   if (!wf) {
-    reply.NotReady(true);
+    reply.NotOpened(path);
     return;
   }
   {

src/pipeline.cc

@@ -32,7 +32,6 @@ limitations under the License.
 #include <llvm/Support/Path.h>
 #include <llvm/Support/Process.h>
 #include <llvm/Support/Threading.h>
-using namespace llvm;
 
 #include <chrono>
 #include <mutex>
@@ -41,6 +40,8 @@ using namespace llvm;
 #ifndef _WIN32
 #include <unistd.h>
 #endif
+using namespace llvm;
+namespace chrono = std::chrono;
 
 namespace ccls {
 namespace {
@@ -527,8 +528,11 @@ void LaunchStdin() {
       if (method.empty())
         continue;
       bool should_exit = method == "exit";
+      // g_config is not available before "initialize". Use 0 in that case.
       on_request->PushBack(
-          {id, std::move(method), std::move(message), std::move(document)});
+          {id, std::move(method), std::move(message), std::move(document),
+           chrono::steady_clock::now() +
+               chrono::milliseconds(g_config ? g_config->request.timeout : 0)});
 
       if (should_exit)
         break;
@@ -590,11 +594,34 @@ void MainLoop() {
   handler.include_complete = &include_complete;
 
   bool has_indexed = false;
+  std::deque<InMessage> backlog;
+  StringMap<std::deque<InMessage *>> path2backlog;
   while (true) {
+    if (backlog.size()) {
+      auto now = chrono::steady_clock::now();
+      handler.overdue = true;
+      while (backlog.size()) {
+        if (backlog[0].backlog_path.size()) {
+          if (now < backlog[0].deadline)
+            break;
+          handler.Run(backlog[0]);
+          path2backlog[backlog[0].backlog_path].pop_front();
+        }
+        backlog.pop_front();
+      }
+      handler.overdue = false;
+    }
+
     std::vector<InMessage> messages = on_request->DequeueAll();
     bool did_work = messages.size();
     for (InMessage &message : messages)
+      try {
         handler.Run(message);
+      } catch (NotIndexed &ex) {
+        backlog.push_back(std::move(message));
+        backlog.back().backlog_path = ex.path;
+        path2backlog[ex.path].push_back(&backlog.back());
+      }
 
     bool indexed = false;
     for (int i = 20; i--;) {
@@ -604,6 +631,16 @@ void MainLoop() {
       did_work = true;
       indexed = true;
       Main_OnIndexed(&db, &wfiles, &*update);
+      if (update->files_def_update) {
+        auto it = path2backlog.find(update->files_def_update->first.path);
+        if (it != path2backlog.end()) {
+          for (auto &message : it->second) {
+            handler.Run(*message);
+            message->backlog_path.clear();
+          }
+          path2backlog.erase(it);
+        }
+      }
     }
 
     if (did_work) {
@@ -615,7 +652,10 @@ void MainLoop() {
         FreeUnusedMemory();
         has_indexed = false;
       }
+      if (backlog.empty())
         main_waiter->Wait(quit, on_indexed, on_request);
+      else
+        main_waiter->WaitUntil(backlog[0].deadline, on_indexed, on_request);
     }
   }
 

src/threaded_queue.hh

@@ -18,6 +18,7 @@ limitations under the License.
 #include "utils.hh"
 
 #include <atomic>
+#include <chrono>
 #include <condition_variable>
 #include <deque>
 #include <mutex>
@@ -76,6 +77,14 @@ struct MultiQueueWaiter {
     }
     return true;
   }
+
+  template <typename... BaseThreadQueue>
+  void WaitUntil(std::chrono::steady_clock::time_point t,
+                 BaseThreadQueue... queues) {
+    MultiQueueLock<BaseThreadQueue...> l(queues...);
+    if (!HasState({queues...}))
+      cv.wait_until(l, t);
+  }
 };
 
 // A threadsafe-queue. http://stackoverflow.com/a/16075550