Changes from a review.

src/parser/TurtleParser.cpp

@@ -822,13 +822,12 @@ bool TurtleStreamParser<T>::getLine(TurtleTriple* triple) {
 
 // We will use the  following trick: For a batch that is forwarded to the
 // parallel parser, we will first increment `numBatchesTotal_` and then call
-// the following lambda from within the thread that parses that batch after
-// it has been completely parsed and pushed to the `tripleCollector`. With
-// this invariant we get that `batchIdx_ == numBatchesTotal_` iff all
-// batches that have been inserted to the `parallelParser_` have been fully
-// processed. After the last batch we will push another call to this lambda
-// to the `parallelParser_` which will then finish the `tripleCollector_` as
-// soon as all batches have been computed.
+// the following lambda after the batch has completely parsed and the result
+// pushed to  the `tripleCollector_`. We thus get the invariant that `batchIdx_
+// == numBatchesTotal_` iff all batches that have been inserted to the
+// `parallelParser_` have been fully processed. After the last batch we will
+// push another call to this lambda to the `parallelParser_` which will then
+// finish the `tripleCollector_` as soon as all batches have been computed.
 template <typename Tokenizer_T>
 void TurtleParallelParser<Tokenizer_T>::finishTripleCollectorIfLastBatch() {
   if (batchIdx_.fetch_add(1) == numBatchesTotal_) {

src/parser/TurtleParser.h

@@ -567,8 +567,8 @@ class TurtleParallelParser : public TurtleParser<Tokenizer_T> {
   TurtleParallelParser() = default;
 
   // If the `sleepTimeForTesting` is set, then after the initialization the
-  // parser will sleep for the specified time for each batch s.t. certain corner
-  // cases can be tested.
+  // parser will sleep for the specified time before parsing each batch s.t.
+  // certain corner cases can be tested.
   explicit TurtleParallelParser(const string& filename,
                                 std::chrono::milliseconds sleepTimeForTesting =
                                     std::chrono::milliseconds{0})
@@ -636,7 +636,6 @@ class TurtleParallelParser : public TurtleParser<Tokenizer_T> {
   // the parser threads can be destroyed. The following two members are needed
   // for keeping track of this condition.
   std::atomic<size_t> batchIdx_ = 0;
-  // `max()` means `not yet set`.
   std::atomic<size_t> numBatchesTotal_ = 0;
 
   std::chrono::milliseconds sleepTimeForTesting_;

src/util/TaskQueue.h

@@ -127,7 +127,6 @@ class TaskQueue {
   // _________________________________________________________________________
   void function_for_thread() {
     while (auto task = queuedTasks_.pop()) {
-      // perform the task without actually holding the lock.
       task.value()();
     }
   }

test/TaskQueueTest.cpp

@@ -45,3 +45,7 @@ TEST(TaskQueue, SimpleSumWithWait) {
   q.finish();
   ASSERT_EQ(result, 500500);
 }
+
+TEST(TaskQueue, ThrowOnMaxQueueSizeZero) {
+  EXPECT_ANY_THROW((ad_utility::TaskQueue{0, 5}));
+}

test/TurtleParserTest.cpp

@@ -787,7 +787,6 @@ TEST(TurtleParserTest, exceptionPropagation) {
 // propagated.
 TEST(TurtleParserTest, exceptionPropagationFileBufferReading) {
   std::string filename{"turtleParserEmptyInput.dat"};
-  FILE_BUFFER_SIZE() = 40;
   auto testWithParser = [&]<typename Parser>(bool useBatchInterface,
                                              std::string_view input) {
     {
@@ -799,13 +798,17 @@ TEST(TurtleParserTest, exceptionPropagationFileBufferReading) {
         ::testing::ContainsRegex("Please increase the FILE_BUFFER_SIZE"));
     ad_utility::deleteFile(filename);
   };
+  // Deliberately chosen s.t. the first triple fits in a block, but the second
+  // one doesn't.
+  FILE_BUFFER_SIZE() = 40;
   forAllParallelParsers(testWithParser,
                         "<subject> <predicate> <object> . \n <veryLongSubject> "
                         "<veryLongPredicate> <veryLongObject> .");
 }
 
-// Test that the parallel parsers destructor can be run quickly, even when there
-// are still lots of blocks in the pipeline that are currently being parsed.
+// Test that the parallel parser's destructor can be run quickly and without
+// blocking, even when there are still lots of blocks in the pipeline that are
+// currently being parsed.
 TEST(TurtleParserTest, stopParsingOnOutsideFailure) {
   std::string filename{"turtleParserStopParsingOnOutsideFailure.dat"};
   auto testWithParser = [&]<typename Parser>(
@@ -822,7 +825,7 @@ TEST(TurtleParserTest, stopParsingOnOutsideFailure) {
     }
     EXPECT_LE(t.msecs(), 20ms);
   };
-  auto input = []() {
+  const std::string input = []() {
     std::string singleBlock = "<subject> <predicate> <object> . \n ";
     std::string longBlock;
     longBlock.reserve(200 * singleBlock.size());