Double-buffering for JSON async stream

Decreases time needed to process single JSON, and
reduces memory footprint size.

Relates-To: DATASDK-57

Signed-off-by: Andrey Kashcheev <[email protected]>

Author: andrey-kashcheev

olp-cpp-sdk-dataservice-read/src/repositories/AsyncJsonStream.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 HERE Europe B.V.
+ * Copyright (C) 2023-2024 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,47 +19,51 @@
 
 #include "AsyncJsonStream.h"
 
-#include <cstring>
-
 namespace olp {
 namespace dataservice {
 namespace read {
 namespace repository {
 
-RapidJsonByteStream::Ch RapidJsonByteStream::Peek() const {
-  std::unique_lock<std::mutex> lock(mutex_);
-  cv_.wait(lock, [=]() { return !Empty(); });
-  return buffer_[count_];
+RapidJsonByteStream::Ch RapidJsonByteStream::Peek() {
+  if (ReadEmpty()) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    cv_.wait(lock, [&]() { return !WriteEmpty(); });
+    std::swap(read_buffer_, write_buffer_);
+    write_buffer_.clear();
+    count_ = 0;
+  }
+  return read_buffer_[count_];
 }
 
 RapidJsonByteStream::Ch RapidJsonByteStream::Take() {
-  std::unique_lock<std::mutex> lock(mutex_);
-  cv_.wait(lock, [=]() { return !Empty(); });
-  return buffer_[count_++];
+  if (ReadEmpty()) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    cv_.wait(lock, [&]() { return !WriteEmpty(); });
+    std::swap(read_buffer_, write_buffer_);
+  }
+  full_count_++;
+  return read_buffer_[count_++];
 }
 
-size_t RapidJsonByteStream::Tell() const { return count_; }
+size_t RapidJsonByteStream::Tell() const { return full_count_; }
 
 // Not implemented
 char* RapidJsonByteStream::PutBegin() { return 0; }
 void RapidJsonByteStream::Put(char) {}
 void RapidJsonByteStream::Flush() {}
 size_t RapidJsonByteStream::PutEnd(char*) { return 0; }
 
-bool RapidJsonByteStream::Empty() const { return count_ == buffer_.size(); }
+bool RapidJsonByteStream::ReadEmpty() const {
+  return count_ == read_buffer_.size();
+}
+bool RapidJsonByteStream::WriteEmpty() const { return write_buffer_.empty(); }
 
 void RapidJsonByteStream::AppendContent(const char* content, size_t length) {
   std::unique_lock<std::mutex> lock(mutex_);
 
-  if (Empty()) {
-    buffer_.resize(length);
-    std::memcpy(buffer_.data(), content, length);
-    count_ = 0;
-  } else {
-    const auto buffer_size = buffer_.size();
-    buffer_.resize(buffer_size + length);
-    std::memcpy(buffer_.data() + buffer_size, content, length);
-  }
+  const auto buffer_size = write_buffer_.size();
+  write_buffer_.reserve(buffer_size + length);
+  write_buffer_.insert(write_buffer_.end(), content, content + length);
 
   cv_.notify_one();
 }
@@ -97,7 +101,7 @@ void AsyncJsonStream::CloseStream(boost::optional<client::ApiError> error) {
     return;
   }
   current_stream_->AppendContent("\0", 1);
-  error_ = error;
+  error_ = std::move(error);
   closed_ = true;
 }
 

olp-cpp-sdk-dataservice-read/src/repositories/AsyncJsonStream.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2023 HERE Europe B.V.
+ * Copyright (C) 2023-2024 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,13 +32,13 @@ namespace dataservice {
 namespace read {
 namespace repository {
 
-// Json byte stream class. Implements rapidjson input stream concept.
+/// Json byte stream class. Implements rapidjson input stream concept.
 class RapidJsonByteStream {
  public:
   typedef char Ch;
 
   /// Read the current character from stream without moving the read cursor.
-  Ch Peek() const;
+  Ch Peek();
 
   /// Read the current character from stream and moving the read cursor to next
   /// character.
@@ -47,19 +47,22 @@ class RapidJsonByteStream {
   /// Get the current read cursor.
   size_t Tell() const;
 
-  // Not needed for reading.
+  /// Not needed for reading.
   char* PutBegin();
   void Put(char);
   void Flush();
   size_t PutEnd(char*);
 
-  bool Empty() const;
+  bool ReadEmpty() const;
+  bool WriteEmpty() const;
 
   void AppendContent(const char* content, size_t length);
 
  private:
   mutable std::mutex mutex_;
-  std::vector<char> buffer_;            // Current buffer
+  std::vector<char> read_buffer_;       // Current buffer
+  std::vector<char> write_buffer_;      // Current buffer
+  size_t full_count_{0};                // Bytes read from the buffer
   size_t count_{0};                     // Bytes read from the buffer
   mutable std::condition_variable cv_;  // Condition for next portion of content
 };