0ba93ab031
The packaged service becomes a systemd user unit that runs each user's own daemon on the CLI's default per-user socket, with that user's config and data. There is no system daemon left to discover, and the daemon always runs as the same user as the CLI, so both workarounds for the system-service design go: - Revert "feat(cli): find the packaged system daemon socket without configuration" (client-side /run/yams/yams-daemon.sock discovery, the systemd refusal in `daemon stop|restart`, the spawn refusal and the status mismatch suppression, plus its tests). - Revert "fix(cli): send file content to a daemon that runs as another user" (peer-UID probe and content upload in `yams add`). Per-user socket resolution is back to: explicit flag/env/config, then $XDG_RUNTIME_DIR/yams-daemon.sock (or the /tmp fallback). Signed-off-by: trvon <me@trevon.dev>
661 lines
24 KiB
C++
661 lines
24 KiB
C++
// Daemon components test suite (Catch2)
|
|
// Consolidates: SocketServer lifecycle, socket utils, resource pool
|
|
// Covers: IPC infrastructure, socket path resolution, resource lifecycle management
|
|
|
|
// pi-lens-ignore: fatal error
|
|
#include <catch2/catch_test_macros.hpp>
|
|
#include <catch2/generators/catch_generators.hpp>
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdlib>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <future>
|
|
#include <iostream>
|
|
#include <optional>
|
|
#include <thread>
|
|
#include <vector>
|
|
#include "../../common/test_helpers_catch2.h"
|
|
#include <yams/compat/unistd.h>
|
|
|
|
#include <yams/daemon/components/IOCoordinator.h>
|
|
#include <yams/daemon/components/LifecycleComponent.h>
|
|
#include <yams/daemon/components/ResourceGovernor.h>
|
|
#include <yams/daemon/components/SearchEngineManager.h>
|
|
#include <yams/daemon/components/SocketServer.h>
|
|
#include <yams/daemon/components/StateComponent.h>
|
|
#include <yams/daemon/components/WorkCoordinator.h>
|
|
#include <yams/daemon/ipc/socket_utils.h>
|
|
#include <yams/daemon/resource/resource_pool.h>
|
|
|
|
#ifdef _WIN32
|
|
inline int geteuid() {
|
|
return 1000;
|
|
}
|
|
inline int getuid() {
|
|
return 1000;
|
|
}
|
|
#endif
|
|
|
|
using namespace yams::daemon;
|
|
using namespace std::chrono_literals;
|
|
namespace fs = std::filesystem;
|
|
|
|
#ifndef _WIN32
|
|
#include <sys/un.h>
|
|
#endif
|
|
|
|
// =============================================================================
|
|
// Test Helpers
|
|
// =============================================================================
|
|
|
|
namespace {
|
|
fs::path makeTempRuntimeDir(const std::string& name) {
|
|
auto base = fs::temp_directory_path();
|
|
#ifndef _WIN32
|
|
constexpr std::size_t maxUnixPath = sizeof(sockaddr_un::sun_path) - 1;
|
|
auto candidate = base / "yams-component-tests" / name;
|
|
auto candidateWithSocket = candidate / "ipc.sock";
|
|
std::cerr << "candidate path: " << candidateWithSocket
|
|
<< " len=" << candidateWithSocket.native().size() << " limit=" << maxUnixPath << "\n";
|
|
if (candidateWithSocket.native().size() >= maxUnixPath) {
|
|
base = fs::path("/tmp");
|
|
candidate = base / "yams-component-tests" / name;
|
|
candidateWithSocket = candidate / "ipc.sock";
|
|
std::cerr << "fallback path: " << candidateWithSocket
|
|
<< " len=" << candidateWithSocket.native().size() << "\n";
|
|
}
|
|
#endif
|
|
auto dir = base / "yams-component-tests" / name;
|
|
std::error_code ec;
|
|
fs::create_directories(dir, ec);
|
|
return dir;
|
|
}
|
|
|
|
TEST_CASE("LifecycleComponent test safe-instance guard overrides production aggressive mode",
|
|
"[daemon][lifecycle][single-instance]") {
|
|
yams::test::ScopedEnvVar safeGuard{"YAMS_TEST_SAFE_SINGLE_INSTANCE", std::string{"1"}};
|
|
yams::test::ScopedEnvVar killOthers{"YAMS_DAEMON_KILL_OTHERS", std::string{"1"}};
|
|
CHECK_FALSE(LifecycleComponent::testingAggressiveModeEnabled());
|
|
|
|
safeGuard.unset();
|
|
CHECK((LifecycleComponent::testingAggressiveModeEnabled()));
|
|
|
|
killOthers.set("0");
|
|
CHECK_FALSE(LifecycleComponent::testingAggressiveModeEnabled());
|
|
}
|
|
|
|
TEST_CASE("SearchEngineManager snapshots topology compatibility policy once",
|
|
"[daemon][search][config][topology][snapshot]") {
|
|
yams::test::ScopedEnvVar canonicalConfig{"YAMS_CONFIG", std::nullopt};
|
|
yams::test::ScopedEnvVar compatibilityConfig{"YAMS_CONFIG_PATH", std::nullopt};
|
|
yams::test::ScopedEnvVar maxClusters{"YAMS_SEARCH_TOPOLOGY_MAX_CLUSTERS", std::string{"3"}};
|
|
|
|
SearchEngineManager manager;
|
|
maxClusters.set("7");
|
|
|
|
REQUIRE(manager.topologyMaxClustersSnapshot().has_value());
|
|
CHECK((*manager.topologyMaxClustersSnapshot() == 3));
|
|
}
|
|
|
|
TEST_CASE("SearchEngineManager runtime executor is isolated from unavailable shared workers",
|
|
"[daemon][search][executor][liveness]") {
|
|
constexpr std::size_t kRequestConcurrency = 8;
|
|
boost::asio::io_context unavailableSharedContext;
|
|
std::atomic<std::size_t> sharedWorkRan{0};
|
|
for (std::size_t i = 0; i < kRequestConcurrency; ++i) {
|
|
boost::asio::post(unavailableSharedContext, [&sharedWorkRan]() {
|
|
sharedWorkRan.fetch_add(1, std::memory_order_release);
|
|
});
|
|
}
|
|
|
|
SearchEngineManager manager;
|
|
std::vector<std::promise<void>> searchWorkCompleted(kRequestConcurrency);
|
|
std::vector<std::future<void>> searchWorkFutures;
|
|
searchWorkFutures.reserve(kRequestConcurrency);
|
|
for (std::size_t i = 0; i < searchWorkCompleted.size(); ++i) {
|
|
searchWorkFutures.push_back(searchWorkCompleted[i].get_future());
|
|
boost::asio::post(manager.testingRuntimeExecutor(),
|
|
[&searchWorkCompleted, i]() { searchWorkCompleted[i].set_value(); });
|
|
}
|
|
|
|
const auto deadline = std::chrono::steady_clock::now() + 500ms;
|
|
for (auto& future : searchWorkFutures) {
|
|
CHECK((future.wait_until(deadline) == std::future_status::ready));
|
|
}
|
|
CHECK((sharedWorkRan.load(std::memory_order_acquire) == 0));
|
|
|
|
unavailableSharedContext.run();
|
|
CHECK((sharedWorkRan.load(std::memory_order_acquire) == kRequestConcurrency));
|
|
}
|
|
|
|
std::string randomSuffix() {
|
|
return std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
|
|
}
|
|
|
|
template <typename ResultLike> bool isPermissionDenied(const ResultLike& result) {
|
|
if (result) {
|
|
return false;
|
|
}
|
|
std::string_view message{result.error().message};
|
|
return message.find("Operation not permitted") != std::string_view::npos ||
|
|
message.find("Permission denied") != std::string_view::npos;
|
|
}
|
|
|
|
// Test resource for pool testing
|
|
class TestResource {
|
|
public:
|
|
TestResource() { creationCount_++; }
|
|
|
|
bool isValid() const { return valid_; }
|
|
void invalidate() { valid_ = false; }
|
|
|
|
void doWork() const {
|
|
workCount_++;
|
|
std::this_thread::sleep_for(1ms);
|
|
}
|
|
|
|
static void resetCounters() {
|
|
creationCount_ = 0;
|
|
workCount_ = 0;
|
|
}
|
|
|
|
static int getCreationCount() { return creationCount_; }
|
|
static int getWorkCount() { return workCount_; }
|
|
|
|
private:
|
|
bool valid_{true};
|
|
static std::atomic<int> creationCount_;
|
|
static std::atomic<int> workCount_;
|
|
};
|
|
|
|
std::atomic<int> TestResource::creationCount_{0};
|
|
std::atomic<int> TestResource::workCount_{0};
|
|
|
|
using yams::test::ScopedEnvVar;
|
|
} // namespace
|
|
|
|
// =============================================================================
|
|
// SocketServer Lifecycle Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("SocketServer: Lifecycle management", "[daemon][components][socket]") {
|
|
auto runtimeDir = makeTempRuntimeDir("socket-lifecycle-" + randomSuffix());
|
|
auto socketPath = runtimeDir / "ipc.sock";
|
|
|
|
SocketServer::Config config;
|
|
config.socketPath = socketPath;
|
|
config.connectionTimeout = 1500ms;
|
|
|
|
StateComponent state;
|
|
state.stats.startTime = std::chrono::steady_clock::now();
|
|
|
|
SECTION("Start and stop clears stopping flag") {
|
|
WorkCoordinator coordinator;
|
|
coordinator.start(2);
|
|
IOCoordinator io;
|
|
io.start();
|
|
SocketServer server(config, &io, &coordinator, nullptr, &state);
|
|
|
|
auto first = server.start();
|
|
if (!first) {
|
|
UNSCOPED_INFO(first.error().message);
|
|
}
|
|
if (!first && isPermissionDenied(first)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE((first));
|
|
|
|
auto stopped = server.stop();
|
|
REQUIRE((stopped));
|
|
|
|
// Second start should work (stopping flag cleared)
|
|
auto second = server.start();
|
|
if (!second) {
|
|
UNSCOPED_INFO(second.error().message);
|
|
}
|
|
if (!second && isPermissionDenied(second)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE((second));
|
|
|
|
REQUIRE((server.stop()));
|
|
|
|
io.stop();
|
|
io.join();
|
|
coordinator.stop();
|
|
coordinator.join();
|
|
}
|
|
|
|
SECTION("Multiple start attempts are safe") {
|
|
WorkCoordinator coordinator;
|
|
coordinator.start(2);
|
|
IOCoordinator io;
|
|
io.start();
|
|
SocketServer server(config, &io, &coordinator, nullptr, &state);
|
|
|
|
auto first = server.start();
|
|
if (!first) {
|
|
UNSCOPED_INFO(first.error().message);
|
|
}
|
|
if (!first && isPermissionDenied(first)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE((first));
|
|
|
|
// Second start while running should fail gracefully
|
|
auto second = server.start();
|
|
REQUIRE((!second)); // Already running
|
|
|
|
REQUIRE((server.stop()));
|
|
|
|
io.stop();
|
|
io.join();
|
|
coordinator.stop();
|
|
coordinator.join();
|
|
}
|
|
|
|
SECTION("Multi-threaded main and proxy accept shutdown is serialized") {
|
|
WorkCoordinator coordinator;
|
|
coordinator.start(2);
|
|
IOCoordinator io(IOCoordinator::Config{.num_threads = 4});
|
|
io.start();
|
|
|
|
auto proxyConfig = config;
|
|
proxyConfig.proxySocketPath = runtimeDir / "proxy.sock";
|
|
for (int iteration = 0; iteration < 16; ++iteration) {
|
|
SocketServer server(proxyConfig, &io, &coordinator, nullptr, &state);
|
|
auto started = server.start();
|
|
if (!started) {
|
|
UNSCOPED_INFO(started.error().message);
|
|
}
|
|
if (!started && isPermissionDenied(started)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE((started));
|
|
|
|
auto firstStop = std::async(std::launch::async, [&server] { return server.stop(); });
|
|
auto secondStop = std::async(std::launch::async, [&server] { return server.stop(); });
|
|
REQUIRE((firstStop.get()));
|
|
REQUIRE((secondStop.get()));
|
|
}
|
|
|
|
io.stop();
|
|
io.join();
|
|
coordinator.stop();
|
|
coordinator.join();
|
|
}
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(runtimeDir, ec);
|
|
}
|
|
|
|
TEST_CASE("SocketServer: main-socket admission ignores proxy connection load",
|
|
"[daemon][components][socket][admission]") {
|
|
SocketServer::Config config;
|
|
config.socketPath = "/tmp/yams-socket-admission-test.sock";
|
|
config.maxConnections = 4;
|
|
config.proxyMaxConnections = 64;
|
|
|
|
StateComponent state;
|
|
SocketServer server(config, nullptr, nullptr, nullptr, &state);
|
|
REQUIRE((server.resizeConnectionSlots(config.maxConnections)));
|
|
|
|
AddDocumentRequest addRequest;
|
|
addRequest.path = "src";
|
|
addRequest.recursive = true;
|
|
|
|
SECTION("Main-socket write admission keys off main connections, not total connections") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/0, /*proxyActive=*/10);
|
|
|
|
CHECK((server.activeConnections() == 10));
|
|
CHECK((server.proxyActiveConnections() == 10));
|
|
CHECK((server.testing_mainSocketAdmissionVerdict(addRequest) ==
|
|
SocketAdmissionVerdict::admit));
|
|
CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());
|
|
CHECK((server.getSlotUtilization() == 0.0));
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Main-socket pressure still rejects once main connections cross the limit bands") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/0);
|
|
|
|
CHECK((server.testing_mainSocketAdmissionVerdict(addRequest) ==
|
|
SocketAdmissionVerdict::reject));
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Control requests remain bypassed on the main socket") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/20);
|
|
|
|
CHECK((server.testing_mainSocketAdmissionVerdict(StatusRequest{}) ==
|
|
SocketAdmissionVerdict::bypass));
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Emergency guard is driven only by main active connections") {
|
|
const auto hardLimit = AdmissionPolicy::emergencySessionLimit(config.maxConnections);
|
|
|
|
server.testing_setConnectionCounts(/*mainActive=*/hardLimit - 1, /*proxyActive=*/hardLimit);
|
|
CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());
|
|
|
|
server.testing_setConnectionCounts(/*mainActive=*/hardLimit, /*proxyActive=*/0);
|
|
CHECK((server.testing_mainSocketEmergencyGuardRejects()));
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Socket Path Resolution Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("Socket path resolution: Priority ordering", "[daemon][components][socket][utils]") {
|
|
auto tempDir = makeTempRuntimeDir("socket-utils-" + randomSuffix());
|
|
auto configBase = tempDir / "config";
|
|
auto configDir = configBase / "yams";
|
|
fs::create_directories(configDir);
|
|
|
|
ScopedEnvVar yamsConfigGuard("YAMS_CONFIG", std::nullopt);
|
|
ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET", std::nullopt);
|
|
ScopedEnvVar yamsSockCompatibilityGuard("YAMS_DAEMON_SOCKET_PATH", std::nullopt);
|
|
ScopedEnvVar xdgRuntimeGuard("XDG_RUNTIME_DIR", std::nullopt);
|
|
ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME", configBase.string());
|
|
ScopedEnvVar homeGuard("HOME", tempDir.string());
|
|
|
|
SECTION("Environment variable has highest priority") {
|
|
auto testPath = "/tmp/custom-daemon.sock";
|
|
yamsSockGuard.set(testPath);
|
|
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE((result.string() == testPath));
|
|
|
|
auto resultConfigFirst = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE((resultConfigFirst.string() == testPath));
|
|
}
|
|
|
|
#ifndef _WIN32
|
|
SECTION("XDG_RUNTIME_DIR used when env var not set") {
|
|
if (::geteuid() == 0) {
|
|
// Root runs default to /var/run per resolve_socket_path().
|
|
// This test specifically targets non-root behavior.
|
|
SKIP("XDG_RUNTIME_DIR precedence test is for non-root users");
|
|
}
|
|
yamsSockGuard.unset();
|
|
|
|
auto xdgDir = tempDir / "runtime";
|
|
fs::create_directories(xdgDir);
|
|
xdgRuntimeGuard.set(xdgDir.string());
|
|
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE((result.parent_path() == xdgDir));
|
|
REQUIRE((result.filename() == "yams-daemon.sock"));
|
|
}
|
|
|
|
SECTION("/tmp fallback for non-root when XDG not available") {
|
|
yamsSockGuard.unset();
|
|
xdgRuntimeGuard.unset();
|
|
|
|
if (::geteuid() != 0) {
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE((result.parent_path() == "/tmp"));
|
|
// Socket name includes UID for non-root users
|
|
auto uid = std::to_string(::getuid());
|
|
REQUIRE((result.filename() == "yams-daemon-" + uid + ".sock"));
|
|
}
|
|
}
|
|
#endif // !_WIN32
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(tempDir, ec);
|
|
}
|
|
|
|
TEST_CASE("Socket path resolution: Config file reading", "[daemon][components][socket][utils]") {
|
|
auto tempDir = makeTempRuntimeDir("socket-config-" + randomSuffix());
|
|
auto configBase = tempDir / "config";
|
|
auto configDir = configBase / "yams";
|
|
fs::create_directories(configDir);
|
|
|
|
auto configFile = configDir / "config.toml";
|
|
|
|
ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME", configBase.string());
|
|
ScopedEnvVar yamsConfigGuard("YAMS_CONFIG", std::nullopt);
|
|
ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET", std::nullopt);
|
|
ScopedEnvVar yamsSockCompatibilityGuard("YAMS_DAEMON_SOCKET_PATH", std::nullopt);
|
|
|
|
#ifndef _WIN32
|
|
SECTION("Explicit YAMS_CONFIG socket_path is used") {
|
|
const auto explicitConfig = tempDir / "explicit.toml";
|
|
std::ofstream out(explicitConfig);
|
|
out << "[daemon]\n";
|
|
out << "socket_path = \"/opt/yams/explicit.sock\"\n";
|
|
out.close();
|
|
|
|
yamsConfigGuard.set(explicitConfig.string());
|
|
yamsSockGuard.unset();
|
|
|
|
auto result = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE((result.string() == "/opt/yams/explicit.sock"));
|
|
}
|
|
|
|
SECTION("Config file socket_path is used") {
|
|
std::ofstream out(configFile);
|
|
out << "[daemon]\n";
|
|
out << "socket_path = \"/opt/yams/daemon.sock\"\n";
|
|
out.close();
|
|
|
|
xdgConfigGuard.set(configBase.string());
|
|
yamsSockGuard.unset();
|
|
|
|
auto result = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE((result.string() == "/opt/yams/daemon.sock"));
|
|
}
|
|
#endif // !_WIN32
|
|
|
|
SECTION("Invalid config falls back to default") {
|
|
std::ofstream out(configFile);
|
|
out << "invalid toml content ][[\n";
|
|
out.close();
|
|
|
|
xdgConfigGuard.set(configBase.string());
|
|
yamsSockGuard.unset();
|
|
|
|
// Should fall back to default resolution (doesn't crash)
|
|
auto result = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE((!result.empty()));
|
|
}
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(tempDir, ec);
|
|
}
|
|
|
|
TEST_CASE("Socket path resolution: proxy path derived from main socket",
|
|
"[daemon][components][socket][utils]") {
|
|
const fs::path mainSocket{"/tmp/yams-daemon.sock"};
|
|
const auto proxySocket = socket_utils::derive_proxy_socket_path(mainSocket);
|
|
REQUIRE((proxySocket == fs::path{"/tmp/yams-daemon.proxy.sock"}));
|
|
|
|
const fs::path nestedSocket{"/run/user/1000/custom.sock"};
|
|
const auto nestedProxy = socket_utils::derive_proxy_socket_path(nestedSocket);
|
|
REQUIRE((nestedProxy == fs::path{"/run/user/1000/custom.proxy.sock"}));
|
|
}
|
|
|
|
// =============================================================================
|
|
// Resource Pool Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("ResourcePool: Basic lifecycle", "[daemon][components][pool]") {
|
|
TestResource::resetCounters();
|
|
|
|
PoolConfig<TestResource> config;
|
|
config.minSize = 3;
|
|
config.maxSize = 5;
|
|
config.maxIdle = 3;
|
|
config.idleTimeout = 1s;
|
|
config.acquisitionTimeout = 1s;
|
|
config.validateOnAcquire = true;
|
|
config.preCreateResources = true;
|
|
|
|
SECTION("Pool pre-creates minimum resources") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms); // Allow async creation
|
|
REQUIRE((TestResource::getCreationCount() >= static_cast<int>(config.minSize)));
|
|
}
|
|
|
|
TestResource::resetCounters();
|
|
|
|
SECTION("Resource acquisition and release") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
// Acquire resource (no ID parameter, timeout is optional)
|
|
auto handleResult = pool->acquire();
|
|
REQUIRE((handleResult.has_value()));
|
|
auto& handle = handleResult.value();
|
|
REQUIRE((handle.isValid()));
|
|
REQUIRE((handle->isValid()));
|
|
|
|
// Release by destroying handle (handle goes out of scope)
|
|
}
|
|
|
|
TestResource::resetCounters();
|
|
|
|
SECTION("Invalid resources are replaced") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
// Acquire and invalidate
|
|
{
|
|
auto handleResult = pool->acquire();
|
|
REQUIRE((handleResult.has_value()));
|
|
auto& handle = handleResult.value();
|
|
handle->invalidate();
|
|
// handle goes out of scope and releases
|
|
}
|
|
|
|
// Next acquisition should get valid resource (invalid one discarded)
|
|
auto handle2Result = pool->acquire();
|
|
REQUIRE((handle2Result.has_value()));
|
|
auto& handle2 = handle2Result.value();
|
|
REQUIRE((handle2->isValid()));
|
|
}
|
|
}
|
|
|
|
TEST_CASE("ResourcePool: Concurrency", "[daemon][components][pool][concurrent]") {
|
|
TestResource::resetCounters();
|
|
|
|
PoolConfig<TestResource> config;
|
|
config.minSize = 2;
|
|
config.maxSize = 10;
|
|
config.maxIdle = 5;
|
|
config.idleTimeout = std::chrono::seconds(1);
|
|
config.acquisitionTimeout = std::chrono::seconds(1);
|
|
config.validateOnAcquire = true;
|
|
config.preCreateResources = true;
|
|
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(std::make_shared<TestResource>());
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
SECTION("Concurrent acquisition and release") {
|
|
std::vector<std::thread> threads;
|
|
std::atomic<int> successCount{0};
|
|
|
|
for (int i = 0; i < 20; ++i) {
|
|
threads.emplace_back([&pool, &successCount]() {
|
|
auto handleResult = pool->acquire();
|
|
if (handleResult.has_value() && handleResult.value().isValid()) {
|
|
auto& handle = handleResult.value();
|
|
handle->doWork();
|
|
successCount++;
|
|
}
|
|
});
|
|
}
|
|
|
|
for (auto& t : threads) {
|
|
t.join();
|
|
}
|
|
|
|
// All threads should successfully acquire resources
|
|
REQUIRE((successCount == 20));
|
|
REQUIRE((TestResource::getWorkCount() == 20));
|
|
}
|
|
}
|
|
|
|
#include <yams/daemon/client/asio_connection_pool.h>
|
|
#include <yams/daemon/client/global_io_context.h>
|
|
|
|
TEST_CASE("Daemon components: ResourceGovernor backpressure policy helpers",
|
|
"[daemon][components][governor]") {
|
|
auto& governor = ResourceGovernor::instance();
|
|
|
|
constexpr std::uint32_t currentSlots = 16;
|
|
constexpr std::uint32_t minSlots = 4;
|
|
constexpr std::uint32_t maxSlots = 64;
|
|
constexpr std::uint32_t scaleStep = 8;
|
|
|
|
// Normalize singleton policy state to avoid cross-test carryover.
|
|
governor.reportDbLockContention(0, 1);
|
|
(void)governor.recommendConnectionSlotTarget(currentSlots, currentSlots / 2, minSlots, maxSlots,
|
|
scaleStep, 1);
|
|
|
|
auto unchanged = governor.recommendConnectionSlotTarget(currentSlots, currentSlots, minSlots,
|
|
maxSlots, scaleStep, 3);
|
|
CHECK((unchanged == currentSlots));
|
|
|
|
auto shrunk =
|
|
governor.recommendConnectionSlotTarget(currentSlots, 1, minSlots, maxSlots, scaleStep, 1);
|
|
CHECK((shrunk <= currentSlots));
|
|
CHECK((shrunk >= minSlots));
|
|
|
|
governor.reportDbLockContention(100, 10);
|
|
CHECK((governor.capKgConcurrencyForDbContention(8) <= 2));
|
|
CHECK((governor.capEmbedConcurrencyForDbContention(8) <= 1));
|
|
|
|
governor.reportDbLockContention(0, 10);
|
|
CHECK((governor.capKgConcurrencyForDbContention(8) == 8));
|
|
CHECK((governor.capEmbedConcurrencyForDbContention(4) == 4));
|
|
|
|
const auto retryBaseline = governor.recommendRetryAfterMs(0, 100, 0, 1024, 1, 100, 0, 100, 500);
|
|
const auto retryOverloaded = governor.recommendRetryAfterMs(
|
|
200, 100, 4ULL * 1024 * 1024, 1024ULL * 1024, 200, 50, 100, 100, 500);
|
|
CHECK((retryOverloaded >= retryBaseline));
|
|
CHECK((retryOverloaded <= 5000));
|
|
|
|
const auto pauseIdle = governor.recommendBackpressureReadPauseMs(100);
|
|
CHECK((pauseIdle >= 1));
|
|
CHECK((pauseIdle <= 5000));
|
|
|
|
// Leave singleton policy state neutral for subsequent tests.
|
|
governor.reportDbLockContention(0, 1);
|
|
}
|