Avoid spinning up a separate thread just for waiting when you can just as easily do it yourself.

2 files changed, 17 insertions, 11 deletions

diff --git a/searchlib/src/vespa/searchlib/docstore/filechunk.cpp b/searchlib/src/vespa/searchlib/docstore/filechunk.cpp
index 180d6aa8ced..b7b72552914 100644
--- a/searchlib/src/vespa/searchlib/docstore/filechunk.cpp
+++ b/searchlib/src/vespa/searchlib/docstore/filechunk.cpp
@@ -10,8 +10,9 @@
 #include <vespa/vespalib/data/fileheader.h>
 #include <vespa/vespalib/data/databuffer.h>
 #include <vespa/vespalib/stllike/asciistream.h>
-#include <vespa/vespalib/util/blockingthreadstackexecutor.h>
 #include <vespa/vespalib/objects/nbostream.h>
+#include <vespa/vespalib/util/executor.h>
+#include <vespa/vespalib/util/arrayqueue.hpp>
 #include <vespa/vespalib/util/array.hpp>
 #include <vespa/vespalib/stllike/hash_map.hpp>
 #include <vespa/fastos/file.h>
@@ -304,8 +305,6 @@ FileChunk::getModificationTime() const
 
 namespace {
 
-using FutureChunk = std::future<Chunk::UP>;
-
 struct FixedParams {
     const IGetLid & db;
     IWriteData & dest;
@@ -337,14 +336,15 @@ appendChunks(FixedParams * args, Chunk::UP chunk)
 }
 
 void
-FileChunk::appendTo(vespalib::ThreadExecutor & executor, const IGetLid & db, IWriteData & dest,
+FileChunk::appendTo(vespalib::Executor & executor, const IGetLid & db, IWriteData & dest,
                     uint32_t numChunks, IFileChunkVisitorProgress *visitorProgress)
 {
     assert(frozen() || visitorProgress);
     vespalib::GenerationHandler::Guard lidReadGuard(db.getLidReadGuard());
     assert(numChunks <= getNumChunks());
     FixedParams fixedParams = {db, dest, lidReadGuard, getFileId().getId(), visitorProgress};
-    vespalib::BlockingThreadStackExecutor singleExecutor(1, 64_Ki, executor.getNumThreads()*2);
+    size_t limit = std::thread::hardware_concurrency();
+    vespalib::ArrayQueue<std::future<Chunk::UP>> queue;
     for (size_t chunkId(0); chunkId < numChunks; chunkId++) {
         std::promise<Chunk::UP> promisedChunk;
         std::future<Chunk::UP> futureChunk = promisedChunk.get_future();
@@ -355,11 +355,17 @@ FileChunk::appendTo(vespalib::ThreadExecutor & executor, const IGetLid & db, IWr
             promise.set_value(std::make_unique<Chunk>(chunkId, whole.getData(), whole.getDataLen()));
         }));
 
-        singleExecutor.execute(vespalib::makeLambdaTask([args = &fixedParams, chunk = std::move(futureChunk)]() mutable {
-            appendChunks(args, chunk.get());
-        }));
+        while (queue.size() >= limit) {
+            appendChunks(&fixedParams, queue.front().get());
+            queue.pop();
+        }
+
+        queue.push(std::move(futureChunk));
+    }
+    while ( ! queue.empty() ) {
+        appendChunks(&fixedParams, queue.front().get());
+        queue.pop();
     }
-    singleExecutor.sync();
     dest.close();
 }
 
diff --git a/searchlib/src/vespa/searchlib/docstore/filechunk.h b/searchlib/src/vespa/searchlib/docstore/filechunk.h
index eee69602f8b..0d669d9cfde 100644
--- a/searchlib/src/vespa/searchlib/docstore/filechunk.h
+++ b/searchlib/src/vespa/searchlib/docstore/filechunk.h
@@ -18,7 +18,7 @@ class FastOS_FileInterface;
 namespace vespalib {
     class DataBuffer;
     class GenericHeader;
-    class ThreadExecutor;
+    class Executor;
 }
 
 namespace search {
@@ -163,7 +163,7 @@ public:
     virtual bool frozen() const { return true; }
     const vespalib::string & getName() const { return _name; }
     void compact(const IGetLid & iGetLid);
-    void appendTo(vespalib::ThreadExecutor & executor, const IGetLid & db, IWriteData & dest,
+    void appendTo(vespalib::Executor & executor, const IGetLid & db, IWriteData & dest,
                   uint32_t numChunks, IFileChunkVisitorProgress *visitorProgress);
     /**
      * Must be called after chunk has been created to allow correct