Repository: ndsev/mapget (language: python)

Issue:
Parallel execution of RemoteDataSource does not work due to httplib Client lock
Currently, all workers for a RemoteDataSource share the same httplib Client. However, it is not possible to run parallel requests through one socket, so the parallelization is effectively disabled.


Discussion / hints:

Fix patch:
diff --git a/libs/http-datasource/include/mapget/http-datasource/datasource-client.h b/libs/http-datasource/include/mapget/http-datasource/datasource-client.h
index e6e1fb98..4c008502 100644
--- a/libs/http-datasource/include/mapget/http-datasource/datasource-client.h
+++ b/libs/http-datasource/include/mapget/http-datasource/datasource-client.h
@@ -32,7 +32,12 @@ class RemoteDataSource : public DataSource
     TileFeatureLayer::Ptr get(MapTileKey const& k, Cache::Ptr& cache, DataSourceInfo const& info) override;
 
 private:
-    httplib::Client httpClient_;
+    // DataSourceInfo is fetched in the constructor
+    DataSourceInfo info_;
+
+    // Multiple http clients allow parallel GET requests
+    std::vector<httplib::Client> httpClients_;
+    std::atomic_uint64_t nextClient_{0};
 };
 
 /**
diff --git a/libs/http-datasource/src/datasource-client.cpp b/libs/http-datasource/src/datasource-client.cpp
index 8cb8c5fe..9f3dc047 100644
--- a/libs/http-datasource/src/datasource-client.cpp
+++ b/libs/http-datasource/src/datasource-client.cpp
@@ -8,17 +8,22 @@ namespace mapget
 {
 
 RemoteDataSource::RemoteDataSource(const std::string& host, uint16_t port)
-    : httpClient_(host, port)
 {
+    // Fetch data source info.
+    httplib::Client client(host, port);
+    auto fetchedInfoJson = client.Get("/info");
+    if (!fetchedInfoJson || fetchedInfoJson->status >= 300)
+        throw logRuntimeError("Failed to fetch datasource info.");
+    info_ = DataSourceInfo::fromJson(nlohmann::json::parse(fetchedInfoJson->body));
+
+    // Create as many clients as parallel requests are allowed.
+    for (auto i = 0; i < std::max(info_.maxParallelJobs_, 1); ++i)
+        httpClients_.emplace_back(host, port);
 }
 
 DataSourceInfo RemoteDataSource::info()
 {
-    auto fetchedInfoJson = httpClient_.Get("/info");
-    if (!fetchedInfoJson || fetchedInfoJson->status >= 300)
-        throw logRuntimeError("Failed to fetch datasource info.");
-    auto fetchedInfo = DataSourceInfo::fromJson(nlohmann::json::parse(fetchedInfoJson->body));
-    return fetchedInfo;
+    return info_;
 }
 
 void RemoteDataSource::fill(const TileFeatureLayer::Ptr& featureTile)
@@ -30,14 +35,17 @@ void RemoteDataSource::fill(const TileFeatureLayer::Ptr& featureTile)
 TileFeatureLayer::Ptr
 RemoteDataSource::get(const MapTileKey& k, Cache::Ptr& cache, const DataSourceInfo& info)
 {
-    // Send a GET tile request
-    auto tileResponse = httpClient_.Get(stx::format(
+    // Round-robin usage of http clients to facilitate parallel requests.
+    auto& client = httpClients_[(nextClient_++) % httpClients_.size()];
+
+    // Send a GET tile request.
+    auto tileResponse = client.Get(stx::format(
         "/tile?layer={}&tileId={}&fieldsOffset={}",
         k.layerId_,
         k.tileId_.value_,
         cachedFieldsOffset(info.nodeId_, cache)));
 
-    // Check that the response is OK
+    // Check that the response is OK.
     if (!tileResponse || tileResponse->status >= 300) {
         // Forward to base class get(). This will instantiate a
         // default TileFeatureLayer and call fill(). In our implementation
@@ -46,7 +54,7 @@ RemoteDataSource::get(const MapTileKey& k, Cache::Ptr& cache, const DataSourceIn
         return DataSource::get(k, cache, info);
     }
 
-    // Check the response body for expected content
+    // Check the response body for expected content.
     TileFeatureLayer::Ptr result;
     TileLayerStream::Reader reader(
         [&](auto&& mapId, auto&& layerId) { return info.getLayer(std::string(layerId)); },
diff --git a/libs/http-service/include/mapget/http-service/http-service.h b/libs/http-service/include/mapget/http-service/http-service.h
index 13d85ea1..f4558d84 100644
--- a/libs/http-service/include/mapget/http-service/http-service.h
+++ b/libs/http-service/include/mapget/http-service/http-service.h
@@ -21,6 +21,7 @@ class HttpService : public HttpServer, public Service
     void setup(httplib::Server& server) override;
 
     struct Impl;
+    friend struct Impl;
     std::unique_ptr<Impl> impl_;
 };
 
diff --git a/libs/http-service/src/http-client.cpp b/libs/http-service/src/http-client.cpp
index 5d2c354d..83fedd06 100644
--- a/libs/http-service/src/http-client.cpp
+++ b/libs/http-service/src/http-client.cpp
@@ -60,11 +60,16 @@ LayerTilesRequest::Ptr HttpClient::request(const LayerTilesRequest::Ptr& request
 
     using namespace nlohmann;
 
-    // TODO: Currently, cpp-httplib POST does not support async responses.
-    //  Those are only supported by GET.
+    // TODO: Currently, cpp-httplib client-POST does not support async responses.
+    //  Those are only supported by GET. So, currently, this HttpClient
+    //  does not profit from the streaming response. However, erdblick is
+    //  is fully able to process async responses as it uses the browser fetch()-API.
     auto tileResponse = impl_->client_.Post(
         "/tiles",
-        json::object({{"requests", json::array({request->toJson()})}}).dump(),
+        json::object({
+            {"requests", json::array({request->toJson()})},
+            {"maxKnownFieldIds", reader->fieldDictCache()->fieldDictOffsets()}
+        }).dump(),
         "application/json");
 
     if (tileResponse) {
diff --git a/libs/http-service/src/http-service.cpp b/libs/http-service/src/http-service.cpp
index 9e49de88..8254f036 100644
--- a/libs/http-service/src/http-service.cpp
+++ b/libs/http-service/src/http-service.cpp
@@ -158,6 +158,7 @@ struct HttpService::Impl
                 if (!strBuf.empty()) {
                     log().debug("Streaming {} bytes...", strBuf.size());
                     sink.write(strBuf.data(), strBuf.size());
+                    sink.os.flush();
                     state->buffer_.str("");  // Clear buffer after reading.
                 }
 
@@ -172,6 +173,7 @@ struct HttpService::Impl
             // cleanup callback to abort the requests.
             [state, this](bool success)
             {
+                log().debug("Request finished, success: {}", success);
                 if (!success) {
                     for (auto& request : state->requests_) {
                         self_.abort(request);
@@ -191,10 +193,21 @@ struct HttpService::Impl
 
     void handleStatusRequest(const httplib::Request&, httplib::Response& res)
     {
+        auto serviceStats = self_.getStatistics();
+        auto cacheStats = self_.cache()->getStatistics();
+
         std::ostringstream oss;
         oss << "<html><body>";
         oss << "<h1>Status Information</h1>";
-        oss << "<h2>Data Sources: " << self_.info().size() << "</h2>";
+
+        // Output serviceStats
+        oss << "<h2>Service Statistics</h2>";
+        oss << "<pre>" << serviceStats.dump(4) << "</pre>";  // Indentation of 4 for pretty printing
+
+        // Output cacheStats
+        oss << "<h2>Cache Statistics</h2>";
+        oss << "<pre>" << cacheStats.dump(4) << "</pre>";  // Indentation of 4 for pretty printing
+
         oss << "</body></html>";
         res.set_content(oss.str(), "text/html");
     }
diff --git a/libs/model/include/mapget/model/layer.h b/libs/model/include/mapget/model/layer.h
index e9673ba0..b647dcb6 100644
--- a/libs/model/include/mapget/model/layer.h
+++ b/libs/model/include/mapget/model/layer.h
@@ -47,7 +47,13 @@ struct MapTileKey
     /** Allow default ctor. */
     MapTileKey() = default;
 
-    /** Convert the key to a string. */
+    /** Convert the key to a string. The string will be in the form of
+     *  "(0):(1):(2):(3)", with
+     *   (0) being the layer type enum name,
+     *   (1) being the map id,
+     *   (2) being the layer id,
+     *   (3) being the hexadecimal tile id.
+     */
     [[nodiscard]] std::string toString() const;
 
     /** Operator <, allows this struct to be used as an std::map key. */
diff --git a/libs/model/include/mapget/model/stream.h b/libs/model/include/mapget/model/stream.h
index 93ef000f..a193806f 100644
--- a/libs/model/include/mapget/model/stream.h
+++ b/libs/model/include/mapget/model/stream.h
@@ -14,7 +14,7 @@ namespace mapget
  * Fields dictionary objects. The general stream encoding is a simple
  * Version-Type-Length-Value one:
  * - The version (6b) indicates the protocol version which was used to
- *   serialise the blob. This must be comatible with the current version
+ *   serialise the blob. This must be compatible with the current version
  *   which is used by the mapget library.
  * - The type (1B) must be either Fields (1) or TileFeatureLayer (2).
  * - The length (4b)  indicates the byte-length of the serialized object,
@@ -60,6 +60,9 @@ class TileLayerStream
         /** end-of-stream: Returns true if the internal buffer is exhausted. */
         [[nodiscard]] bool eos();
 
+        /** Obtain the fields dict provider used by this Reader. */
+        std::shared_ptr<CachedFieldsProvider> fieldDictCache();
+
     private:
         enum class Phase { ReadHeader, ReadValue };
 
@@ -97,7 +100,7 @@ class TileLayerStream
         void write(TileFeatureLayer::Ptr const& tileFeatureLayer);
 
     private:
-        void sendMessage(std::string const& bytes, MessageType msgType);
+        void sendMessage(std::string&& bytes, MessageType msgType);
 
         std::function<void(std::string, MessageType)> onMessage_;
         FieldOffsetMap& fieldsOffsets_;
@@ -113,7 +116,21 @@ class TileLayerStream
      */
     struct CachedFieldsProvider
     {
-        virtual std::shared_ptr<Fields> operator() (std::string_view const&);
+        /** Virtual destructor for memory-safe inheritance */
+        virtual ~CachedFieldsProvider() = default;
+
+        /**
+         * This operator is called by the Reader to obtain the fields
+         * dictionary for a particular node id.
+         */
+        virtual std::shared_ptr<Fields> operator() (std::string_view const& nodeIf);
+
+        /**
+         * Obtain the highest known field id for each data source node id,
+         * as currently present in the cache. The resulting dict may be
+         * used by a mapget http client to set the `maxKnownFieldIds` info.
+         */
+        virtual FieldOffsetMap fieldDictOffsets() const;
 
     protected:
         std::shared_mutex fieldCacheMutex_;
diff --git a/libs/model/src/stream.cpp b/libs/model/src/stream.cpp
index 132a21a0..fd2ff6f8 100644
--- a/libs/model/src/stream.cpp
+++ b/libs/model/src/stream.cpp
@@ -62,10 +62,15 @@ bool TileLayerStream::Reader::continueReading()
 
     if (nextValueType_ == MessageType::TileFeatureLayer)
     {
-        onParsedLayer_(std::make_shared<TileFeatureLayer>(
+        auto start = std::chrono::system_clock::now();
+        auto tileFeatureLayer = std::make_shared<TileFeatureLayer>(
             buffer_,
             layerInfoProvider_,
-            [this](auto&& nodeId){return (*fieldCacheProvider_)(nodeId);}));
+            [this](auto&& nodeId){return (*fieldCacheProvider_)(nodeId);});
+        // Calculate duration
+        auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start);
+        log().debug("Reading {} kB took {} ms.", nextValueSize_/1000, elapsed.count());
+        onParsedLayer_(tileFeatureLayer);
     }
     else if (nextValueType_ == MessageType::Fields)
     {
@@ -78,6 +83,11 @@ bool TileLayerStream::Reader::continueReading()
     return true;
 }
 
+std::shared_ptr<TileLayerStream::CachedFieldsProvider> TileLayerStream::Reader::fieldDictCache()
+{
+    return fieldCacheProvider_;
+}
+
 TileLayerStream::Writer::Writer(
     std::function<void(std::string, MessageType)> onMessage,
     FieldOffsetMap& fieldsOffsets)
@@ -103,11 +113,15 @@ void TileLayerStream::Writer::write(TileFeatureLayer::Ptr const& tileFeatureLaye
 
     // Send actual tileFeatureLayer
     std::stringstream serializedFeatureLayer;
+    auto start = std::chrono::system_clock::now();
     tileFeatureLayer->write(serializedFeatureLayer);
-    sendMessage(serializedFeatureLayer.str(), MessageType::TileFeatureLayer);
+    auto bytes = serializedFeatureLayer.str();
+    auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start);
+    log().debug("Writing {} kB took {} ms.", bytes.size()/1000, elapsed.count());
+    sendMessage(std::move(bytes), MessageType::TileFeatureLayer);
 }
 
-void TileLayerStream::Writer::sendMessage(std::string const& bytes, TileLayerStream::MessageType msgType)
+void TileLayerStream::Writer::sendMessage(std::string&& bytes, TileLayerStream::MessageType msgType)
 {
     std::stringstream message;
     bitsery::Serializer<bitsery::OutputStreamAdapter> s(message);
@@ -149,4 +163,12 @@ std::shared_ptr<Fields> TileLayerStream::CachedFieldsProvider::operator()(const
     }
 }
 
+TileLayerStream::FieldOffsetMap TileLayerStream::CachedFieldsProvider::fieldDictOffsets() const
+{
+    auto result = FieldOffsetMap();
+    for (auto const& [nodeId, fieldsDict] : fieldsPerNodeId_)
+        result.emplace(nodeId, fieldsDict->highest());
+    return result;
+}
+
 }
\ No newline at end of file
diff --git a/libs/service/include/mapget/service/cache.h b/libs/service/include/mapget/service/cache.h
index 9b87a9f7..453c5993 100644
--- a/libs/service/include/mapget/service/cache.h
+++ b/libs/service/include/mapget/service/cache.h
@@ -44,6 +44,16 @@ class Cache : public TileLayerStream::CachedFieldsProvider, public std::enable_s
     /** Abstract: Upsert (Update or Insert) a Fields-dict blob. */
     virtual void putFields(std::string const& sourceNodeId, std::string const& v) = 0;
 
+    // Override this method if your cache implementation has special stats
+
+    /**
+     * Get diagnostic statistics. The default implementation returns the following:
+     * `cache-hits`: Number of fulfilled cache requests.
+     * `cache-misses`: Number of cache misses (unfulfilled cache requests).
+     * `loaded-field-dicts`: Number of fields-dictionaries currently held in memory.
+     */
+    virtual nlohmann::json getStatistics() const;
+
     // The following methods are already implemented,
     // they forward to the above methods on-demand.
 
@@ -63,6 +73,10 @@ class Cache : public TileLayerStream::CachedFieldsProvider, public std::enable_s
     // Mutex for fieldCacheOffsets_
     std::mutex fieldCacheOffsetMutex_;
     TileLayerStream::FieldOffsetMap fieldCacheOffsets_;
+
+    // Statistics
+    int64_t cacheHits_ = 0;
+    int64_t cacheMisses_ = 0;
 };
 
 }
diff --git a/libs/service/include/mapget/service/memcache.h b/libs/service/include/mapget/service/memcache.h
index de560f08..38baaa7c 100644
--- a/libs/service/include/mapget/service/memcache.h
+++ b/libs/service/include/mapget/service/memcache.h
@@ -35,6 +35,9 @@ class MemCache : public Cache
     /** Upsert a Fields-dict blob. -> No-Op */
     void putFields(std::string const& sourceNodeId, std::string const& v) override {}
 
+    /** Enriches the statistics with info about the number of cached tiles. */
+    nlohmann::json getStatistics() const override;
+
 private:
     // Cached tile blobs.
     std::shared_mutex cacheMutex_;
diff --git a/libs/service/include/mapget/service/service.h b/libs/service/include/mapget/service/service.h
index 1cf6341f..688f754c 100644
--- a/libs/service/include/mapget/service/service.h
+++ b/libs/service/include/mapget/service/service.h
@@ -144,6 +144,15 @@ class Service
     /** Checks if a DataSource can serve the requested map+layer combination. */
     bool hasLayer(std::string const& mapId, std::string const& layerId);
 
+    /**
+     * Get Statistics about the operation of this service.
+     * Returns the following values:
+     * - `workers`: Number of active workers.
+     * - `datasources`: Number of active data sources.
+     * - `active-requests`: Number of in-flight requests.
+     */
+    [[nodiscard]] nlohmann::json getStatistics() const;
+
     /** Get the Cache which this service was constructed with. */
     [[nodiscard]] Cache::Ptr cache();
 
diff --git a/libs/service/src/cache.cpp b/libs/service/src/cache.cpp
index af438a8f..d365057c 100644
--- a/libs/service/src/cache.cpp
+++ b/libs/service/src/cache.cpp
@@ -42,11 +42,21 @@ std::shared_ptr<Fields> Cache::operator()(const std::string_view& nodeId)
     }
 }
 
+nlohmann::json Cache::getStatistics() const {
+    return {
+        {"cache-hits", cacheHits_},
+        {"cache-misses", cacheMisses_},
+        {"loaded-field-dicts", (int64_t)fieldDictOffsets().size()}
+    };
+}
+
 TileFeatureLayer::Ptr Cache::getTileFeatureLayer(const MapTileKey& k, DataSourceInfo const& i)
 {
     auto tileBlob = getTileLayer(k);
-    if (!tileBlob)
+    if (!tileBlob) {
+        ++cacheMisses_;
         return nullptr;
+    }
     TileFeatureLayer::Ptr result;
     TileLayerStream::Reader tileReader(
         [&i, &k](auto&& mapId, auto&& layerId){
@@ -59,6 +69,7 @@ TileFeatureLayer::Ptr Cache::getTileFeatureLayer(const MapTileKey& k, DataSource
         [&](auto&& parsedLayer){result = parsedLayer;},
         shared_from_this());
     tileReader.read(*tileBlob);
+    ++cacheHits_;
     return result;
 }
 
diff --git a/libs/service/src/memcache.cpp b/libs/service/src/memcache.cpp
index 78e806ae..b9e2d2b5 100644
--- a/libs/service/src/memcache.cpp
+++ b/libs/service/src/memcache.cpp
@@ -29,4 +29,11 @@ void MemCache::putTileLayer(const MapTileKey& k, const std::string& v)
     }
 }
 
+nlohmann::json MemCache::getStatistics() const {
+    auto result = Cache::getStatistics();
+    result["memcache-map-size"] = (int64_t)cachedTiles_.size();
+    result["memcache-fifo-size"] = (int64_t)fifo_.size();
+    return result;
+}
+
 }
\ No newline at end of file
diff --git a/libs/service/src/service.cpp b/libs/service/src/service.cpp
index a69059ca..1affb7da 100644
--- a/libs/service/src/service.cpp
+++ b/libs/service/src/service.cpp
@@ -415,4 +415,20 @@ bool Service::hasLayer(std::string const& mapId, std::string const& layerId)
     return false;
 }
 
+nlohmann::json Service::getStatistics() const
+{
+    auto datasources = nlohmann::json::array();
+    for (auto const& [dataSource, info] : impl_->dataSourceInfo_) {
+        datasources.push_back({
+            {"name", info.mapId_},
+            {"workers", impl_->dataSourceWorkers_[dataSource].size()}
+        });
+    }
+
+    return {
+        {"datasources", datasources},
+        {"active-requests", impl_->requests_.size()}
+    };
+}
+
 }  // namespace mapget


Test patch:
diff --git a/test/unit/test-http-datasource.cpp b/test/unit/test-http-datasource.cpp
index edc96ea2..0e17301f 100644
--- a/test/unit/test-http-datasource.cpp
+++ b/test/unit/test-http-datasource.cpp
@@ -46,7 +46,7 @@ TEST_CASE("HttpDataSource", "[HttpDataSource]")
 
     // Initialize a DataSource.
     mapget::DataSourceServer ds(info);
-    auto dataSourceRequestCount = 0;
+    std::atomic_uint32_t dataSourceRequestCount = 0;
     ds.onTileRequest(
         [&](auto&& tile)
         {