params: safe and efficient async writing parameters (#25912)

* Safe and efficient asynchronous writing parameters

* call putNonBlocking in locationd

* remove space

* ->AsyncWriter

* remove semicolon

* use member function

* asyc write multiple times

* add test case for AsyncWriter

* merge master

* add missing include

* public

* cleanup

* create once

* cleanup

* update that

* explicit waiting

* improve test case

---------

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
old-commit-hash: 0d797f4e8b2552192669d0329c14849f0a970066

common/SConscript

@@ -29,7 +29,7 @@ Export('_common', '_gpucommon')
 
 if GetOption('extras'):
   env.Program('tests/test_common',
-              ['tests/test_runner.cc', 'tests/test_util.cc', 'tests/test_swaglog.cc', 'tests/test_ratekeeper.cc'],
+              ['tests/test_runner.cc', 'tests/test_params.cc', 'tests/test_util.cc', 'tests/test_swaglog.cc', 'tests/test_ratekeeper.cc'],
               LIBS=[_common, 'json11', 'zmq', 'pthread'])
 
 # Cython bindings

common/params.cc

@@ -7,6 +7,7 @@
 #include <csignal>
 #include <unordered_map>
 
+#include "common/queue.h"
 #include "common/swaglog.h"
 #include "common/util.h"
 #include "system/hardware/hw.h"
@@ -327,3 +328,33 @@ void Params::clearAll(ParamKeyType key_type) {
 
   fsync_dir(getParamPath());
 }
+
+void Params::putNonBlocking(const std::string &key, const std::string &val) {
+  static AsyncWriter async_writer;
+  async_writer.queue({params_path, key, val});
+}
+
+
+// AsyncWriter
+
+AsyncWriter::~AsyncWriter() {
+  if (future.valid()) {
+    future.wait();
+  }
+}
+
+void AsyncWriter::queue(const std::tuple<std::string, std::string, std::string> &dat) {
+  q.push(dat);
+  // start thread on demand
+  if (!future.valid() || future.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready) {
+    future = std::async(std::launch::async, &AsyncWriter::write, this);
+  }
+}
+
+void AsyncWriter::write() {
+  std::tuple<std::string, std::string, std::string> dat;
+  while (q.try_pop(dat, 0)) {
+    auto &[path, key, value] = dat;
+    Params(path).put(key, value);
+  }
+}

common/params.h

@@ -1,9 +1,13 @@
 #pragma once
 
+#include <future>
 #include <map>
 #include <string>
+#include <tuple>
 #include <vector>
 
+#include "common/queue.h"
+
 enum ParamKeyType {
   PERSISTENT = 0x02,
   CLEAR_ON_MANAGER_START = 0x04,
@@ -42,8 +46,25 @@ public:
   inline int putBool(const std::string &key, bool val) {
     return put(key.c_str(), val ? "1" : "0", 1);
   }
+  void putNonBlocking(const std::string &key, const std::string &val);
+  inline void putBoolNonBlocking(const std::string &key, bool val) {
+    putNonBlocking(key, val ? "1" : "0");
+  }
 
 private:
   std::string params_path;
   std::string prefix;
 };
+
+class AsyncWriter {
+public:
+  AsyncWriter() {}
+  ~AsyncWriter();
+  void queue(const std::tuple<std::string, std::string, std::string> &dat);
+
+private:
+  void write();
+
+  std::future<void> future;
+  SafeQueue<std::tuple<std::string, std::string, std::string>> q;
+};

common/params.py

@@ -1,10 +1,7 @@
-from openpilot.common.params_pyx import Params, ParamKeyType, UnknownKeyName, put_nonblocking, \
-                                        put_bool_nonblocking
+from openpilot.common.params_pyx import Params, ParamKeyType, UnknownKeyName
 assert Params
 assert ParamKeyType
 assert UnknownKeyName
-assert put_nonblocking
-assert put_bool_nonblocking
 
 if __name__ == "__main__":
   import sys

common/params_pyx.pyx

@@ -3,7 +3,6 @@
 from libcpp cimport bool
 from libcpp.string cimport string
 from libcpp.vector cimport vector
-import threading
 
 cdef extern from "common/params.h":
   cpdef enum ParamKeyType:
@@ -19,6 +18,8 @@ cdef extern from "common/params.h":
     bool getBool(string, bool) nogil
     int remove(string) nogil
     int put(string, string) nogil
+    void putNonBlocking(string, string) nogil
+    void putBoolNonBlocking(string, bool) nogil
     int putBool(string, bool) nogil
     bool checkKey(string) nogil
     string getParamPath(string) nogil
@@ -79,7 +80,7 @@ cdef class Params:
     """
     Warning: This function blocks until the param is written to disk!
     In very rare cases this can take over a second, and your code will hang.
-    Use the put_nonblocking helper function in time sensitive code, but
+    Use the put_nonblocking, put_bool_nonblocking in time sensitive code, but
     in general try to avoid writing params as much as possible.
     """
     cdef string k = self.check_key(key)
@@ -92,6 +93,17 @@ cdef class Params:
     with nogil:
       self.p.putBool(k, val)
 
+  def put_nonblocking(self, key, dat):
+    cdef string k = self.check_key(key)
+    cdef string dat_bytes = ensure_bytes(dat)
+    with nogil:
+      self.p.putNonBlocking(k, dat_bytes)
+
+  def put_bool_nonblocking(self, key, bool val):
+    cdef string k = self.check_key(key)
+    with nogil:
+      self.p.putBoolNonBlocking(k, val)
+
   def remove(self, key):
     cdef string k = self.check_key(key)
     with nogil:
@@ -103,9 +115,3 @@ cdef class Params:
 
   def all_keys(self):
     return self.p.allKeys()
-
-def put_nonblocking(key, val, d=""):
-  threading.Thread(target=lambda: Params(d).put(key, val)).start()
-
-def put_bool_nonblocking(key, bool val, d=""):
-  threading.Thread(target=lambda: Params(d).put_bool(key, val)).start()

common/tests/test_params.cc

@@ -0,0 +1,27 @@
+#include "catch2/catch.hpp"
+#define private public
+#include "common/params.h"
+#include "common/util.h"
+
+TEST_CASE("Params/asyncWriter") {
+  char tmp_path[] = "/tmp/asyncWriter_XXXXXX";
+  const std::string param_path = mkdtemp(tmp_path);
+  Params params(param_path);
+  auto param_names = {"CarParams", "IsMetric"};
+  {
+    AsyncWriter async_writer;
+    for (const auto &name : param_names) {
+      async_writer.queue({param_path, name, "1"});
+      // param is empty
+      REQUIRE(params.get(name).empty());
+    }
+
+    // check if thread is running
+    REQUIRE(async_writer.future.valid());
+    REQUIRE(async_writer.future.wait_for(std::chrono::milliseconds(0)) == std::future_status::timeout);
+  }
+  // check results
+  for (const auto &name : param_names) {
+    REQUIRE(params.get(name) == "1");
+  }
+}

common/tests/test_params.py

@@ -4,7 +4,7 @@ import time
 import uuid
 import unittest
 
-from openpilot.common.params import Params, ParamKeyType, UnknownKeyName, put_nonblocking, put_bool_nonblocking
+from openpilot.common.params import Params, ParamKeyType, UnknownKeyName
 
 class TestParams(unittest.TestCase):
   def setUp(self):
@@ -86,7 +86,7 @@ class TestParams(unittest.TestCase):
     q = Params()
     def _delayed_writer():
       time.sleep(0.1)
-      put_nonblocking("CarParams", "test")
+      Params().put_nonblocking("CarParams", "test")
     threading.Thread(target=_delayed_writer).start()
     assert q.get("CarParams") is None
     assert q.get("CarParams", True) == b"test"
@@ -95,7 +95,7 @@ class TestParams(unittest.TestCase):
     q = Params()
     def _delayed_writer():
       time.sleep(0.1)
-      put_bool_nonblocking("CarParams", True)
+      Params().put_bool_nonblocking("CarParams", True)
     threading.Thread(target=_delayed_writer).start()
     assert q.get("CarParams") is None
     assert q.get("CarParams", True) == b"1"