Abort dispatch if channel size grows beyond a configured threshold.

ebarlas · May 9, 2024 · 7619d0d · 7619d0d
1 parent 492ba78
commit 7619d0d
Show file tree

Hide file tree

Showing 6 changed files with 26 additions and 6 deletions.
diff --git a/config/config.json b/config/config.json
@@ -1,12 +1,13 @@
 {
   "version": {
     "major": 5,
-    "minor": 0
+    "minor": 1
   },
   "windowSize": {
     "x": 1200,
     "y": 600
   },
+  "maxDispatchChannelSize": 100,
   "extraLives": 2,
   "highScores": 15,
   "levelBonus": 1000,

diff --git a/src/game/Configuration.cpp b/src/game/Configuration.cpp
@@ -102,6 +102,7 @@ void trippin::from_json(const nlohmann::json &j, Configuration &config) {
         j.at("windowSize").at("x").get_to(config.windowSize.x);
         j.at("windowSize").at("y").get_to(config.windowSize.y);
     }
+    j.at("maxDispatchChannelSize").get_to(config.maxDispatchChannelSize);
     j.at("extraLives").get_to(config.extraLives);
     j.at("highScores").get_to(config.highScores);
     j.at("levelBonus").get_to(config.levelBonus);

diff --git a/src/game/Configuration.h b/src/game/Configuration.h
@@ -59,6 +59,7 @@ namespace trippin {
 
         Version version;
         Point<int> windowSize;
+        int maxDispatchChannelSize;
         int extraLives;
         int highScores;
         int levelBonus;

diff --git a/src/game/Db.h b/src/game/Db.h
@@ -24,10 +24,12 @@ namespace trippin {
         explicit Db(
                 const char *name,
                 int version,
+                int maxChannelSize,
                 std::function<void(const T &)> addCallback,
                 std::function<bool(const T &)> dispatchFn) :
                 name(name),
                 fileName(makeFileName(name, version)),
+                maxChannelSize(maxChannelSize),
                 addCallback(addCallback),
                 dispatchFn(dispatchFn) {
         }
@@ -48,6 +50,7 @@ namespace trippin {
         std::string fileName;
         Channel<Event> inChannel;
         Channel<T> outChannel;
+        const int maxChannelSize;
         std::function<void(const T &)> addCallback;
         std::function<bool(const T &)> dispatchFn;
 
@@ -137,6 +140,11 @@ namespace trippin {
                     break;
                 }
                 while (!dispatchFn(*val)) {
+                    auto sz = outChannel.size();
+                    if (sz >= maxChannelSize) {
+                        SDL_Log("aborting dispatch due to full channel, iter=%lu, type=%s, size=%d", iter, name, sz);
+                        break;
+                    }
                     std::this_thread::sleep_for(std::chrono::seconds(5));
                 }
                 inChannel.put({*val, false});

diff --git a/src/game/Game.cpp b/src/game/Game.cpp
@@ -87,14 +87,15 @@ namespace trippin {
 };
 
 void trippin::Game::initDbSynchronizer() {
+    int version = configuration.version.major;
     transport = std::make_unique<Transport>(
             configuration.db.host,
             configuration.db.port,
-            configuration.version.major,
+            version,
             configuration.highScores);
     stagingArea = std::make_unique<StagingArea>(*transport);
     stagingArea->start();
-    myScores = std::make_unique<MyScores>(configuration.version.major, 10);
+    myScores = std::make_unique<MyScores>(version, 10);
     auto scoreAddCallback = [m = myScores.get()](const Score &s) {
         m->addScore(s);
     };
@@ -104,15 +105,16 @@ void trippin::Game::initDbSynchronizer() {
                 s.id.c_str(), s.game, s.name.c_str(), s.score, toString(result));
         return result == AddResult::success || result == AddResult::clientError;
     };
-    scoreDb = std::make_unique<Db<Score>>("scores", configuration.version.major, scoreAddCallback, scoreDispatchFn);
+    int mdcs = configuration.maxDispatchChannelSize;
+    scoreDb = std::make_unique<Db<Score>>("scores", version, mdcs, scoreAddCallback, scoreDispatchFn);
     scoreDb->start();
     auto logAddCallback = [](const LogEvent &e) {};
     auto logDispatchFn = [t = transport.get()](const LogEvent &e) {
         auto result = t->addLogEvent(e);
         SDL_Log("add log event attempted, index=%d, result=%s", e.index, toString(result));
         return result == AddResult::success || result == AddResult::clientError;
     };
-    logDb = std::make_unique<Db<LogEvent>>("logs", configuration.version.major, logAddCallback, logDispatchFn);
+    logDb = std::make_unique<Db<LogEvent>>("logs", version, mdcs, logAddCallback, logDispatchFn);
     logDb->start();
 }
 

diff --git a/src/net/Channel.h b/src/net/Channel.h
@@ -21,7 +21,7 @@ namespace trippin {
             return e;
         }
 
-        bool put(const T& elem) {
+        bool put(const T &elem) {
             {
                 std::lock_guard lock(mutex);
                 if (closed) {
@@ -41,6 +41,13 @@ namespace trippin {
             cv.notify_all();
         }
 
+        int size() {
+            {
+                std::lock_guard lock(mutex);
+                return queue.size();
+            }
+        }
+
     private:
         std::queue<T> queue;
         std::mutex mutex;