mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-21 16:23:46 +08:00
Revert "params: safe and efficient async writing parameters (#25912)"
This reverts commit 780669c33fea1b2a14a0bd6e2eac82c9b8893aa5. old-commit-hash: ec479322d3d8d789290bee68919d14b040481d53
This commit is contained in:
+1
-1
@@ -29,7 +29,7 @@ Export('_common', '_gpucommon')
|
||||
|
||||
if GetOption('extras'):
|
||||
env.Program('tests/test_common',
|
||||
['tests/test_runner.cc', 'tests/test_params.cc', 'tests/test_util.cc', 'tests/test_swaglog.cc', 'tests/test_ratekeeper.cc'],
|
||||
['tests/test_runner.cc', 'tests/test_util.cc', 'tests/test_swaglog.cc', 'tests/test_ratekeeper.cc'],
|
||||
LIBS=[_common, 'json11', 'zmq', 'pthread'])
|
||||
|
||||
# Cython bindings
|
||||
|
||||
@@ -7,7 +7,6 @@
|
||||
#include <csignal>
|
||||
#include <unordered_map>
|
||||
|
||||
#include "common/queue.h"
|
||||
#include "common/swaglog.h"
|
||||
#include "common/util.h"
|
||||
#include "system/hardware/hw.h"
|
||||
@@ -328,33 +327,3 @@ 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,13 +1,9 @@
|
||||
#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,
|
||||
@@ -46,25 +42,8 @@ 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;
|
||||
};
|
||||
|
||||
+4
-1
@@ -1,7 +1,10 @@
|
||||
from openpilot.common.params_pyx import Params, ParamKeyType, UnknownKeyName
|
||||
from openpilot.common.params_pyx import Params, ParamKeyType, UnknownKeyName, put_nonblocking, \
|
||||
put_bool_nonblocking
|
||||
assert Params
|
||||
assert ParamKeyType
|
||||
assert UnknownKeyName
|
||||
assert put_nonblocking
|
||||
assert put_bool_nonblocking
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
+8
-14
@@ -3,6 +3,7 @@
|
||||
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:
|
||||
@@ -18,8 +19,6 @@ 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
|
||||
@@ -80,7 +79,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, put_bool_nonblocking in time sensitive code, but
|
||||
Use the put_nonblocking helper function in time sensitive code, but
|
||||
in general try to avoid writing params as much as possible.
|
||||
"""
|
||||
cdef string k = self.check_key(key)
|
||||
@@ -93,17 +92,6 @@ 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:
|
||||
@@ -115,3 +103,9 @@ 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()
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
#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");
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import time
|
||||
import uuid
|
||||
import unittest
|
||||
|
||||
from openpilot.common.params import Params, ParamKeyType, UnknownKeyName
|
||||
from openpilot.common.params import Params, ParamKeyType, UnknownKeyName, put_nonblocking, put_bool_nonblocking
|
||||
|
||||
class TestParams(unittest.TestCase):
|
||||
def setUp(self):
|
||||
@@ -86,7 +86,7 @@ class TestParams(unittest.TestCase):
|
||||
q = Params()
|
||||
def _delayed_writer():
|
||||
time.sleep(0.1)
|
||||
Params().put_nonblocking("CarParams", "test")
|
||||
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)
|
||||
Params().put_bool_nonblocking("CarParams", True)
|
||||
put_bool_nonblocking("CarParams", True)
|
||||
threading.Thread(target=_delayed_writer).start()
|
||||
assert q.get("CarParams") is None
|
||||
assert q.get("CarParams", True) == b"1"
|
||||
|
||||
Reference in New Issue
Block a user