forked from Kistler-Group/sdbus-cpp
Add stress tests for sdbus-c++
This commit is contained in:
Stanislav Angelovic
Executable
+322
@@ -0,0 +1,322 @@
|
||||
/**
|
||||
* (C) 2019 KISTLER INSTRUMENTE AG, Winterthur, Switzerland
|
||||
*
|
||||
* @file stresstests.cpp
|
||||
*
|
||||
* Created on: Jan 25, 2019
|
||||
* Project: sdbus-c++
|
||||
* Description: High-level D-Bus IPC C++ library based on sd-bus
|
||||
*
|
||||
* This file is part of sdbus-c++.
|
||||
*
|
||||
* sdbus-c++ is free software; you can redistribute it and/or modify it
|
||||
* under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation, either version 2.1 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* sdbus-c++ is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with sdbus-c++. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "celsius-thermometer-adaptor.h"
|
||||
#include "celsius-thermometer-proxy.h"
|
||||
#include "fahrenheit-thermometer-adaptor.h"
|
||||
#include "fahrenheit-thermometer-proxy.h"
|
||||
#include "concatenator-adaptor.h"
|
||||
#include "concatenator-proxy.h"
|
||||
#include <sdbus-c++/sdbus-c++.h>
|
||||
#include <vector>
|
||||
#include <string>
|
||||
#include <iostream>
|
||||
#include <unistd.h>
|
||||
#include <thread>
|
||||
#include <chrono>
|
||||
#include <cassert>
|
||||
#include <atomic>
|
||||
#include <sstream>
|
||||
#include <mutex>
|
||||
#include <condition_variable>
|
||||
#include <queue>
|
||||
|
||||
using namespace std::chrono_literals;
|
||||
using namespace std::string_literals;
|
||||
|
||||
#define SERVICE_1_BUS_NAME "org.sdbuscpp.stresstest.service1"
|
||||
#define SERVICE_2_BUS_NAME "org.sdbuscpp.stresstest.service2"
|
||||
#define CELSIUS_THERMOMETER_OBJECT_PATH "/org/sdbuscpp/stresstest/celsius/thermometer"
|
||||
#define FAHRENHEIT_THERMOMETER_OBJECT_PATH "/org/sdbuscpp/stresstest/fahrenheit/thermometer"
|
||||
#define CONCATENATOR_OBJECT_PATH "/org/sdbuscpp/stresstest/concatenator"
|
||||
|
||||
class CelsiusThermometerAdaptor : public sdbus::Interfaces<org::sdbuscpp::stresstest::celsius::thermometer_adaptor>
|
||||
{
|
||||
public:
|
||||
using sdbus::Interfaces<org::sdbuscpp::stresstest::celsius::thermometer_adaptor>::Interfaces;
|
||||
|
||||
protected:
|
||||
virtual uint32_t getCurrentTemperature() override
|
||||
{
|
||||
return m_currentTemperature++;
|
||||
}
|
||||
|
||||
private:
|
||||
uint32_t m_currentTemperature{};
|
||||
};
|
||||
|
||||
class CelsiusThermometerProxy : public sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::celsius::thermometer_proxy>
|
||||
{
|
||||
public:
|
||||
using sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::celsius::thermometer_proxy>::ProxyInterfaces;
|
||||
};
|
||||
|
||||
class FahrenheitThermometerAdaptor : public sdbus::Interfaces<org::sdbuscpp::stresstest::fahrenheit::thermometer_adaptor>
|
||||
{
|
||||
public:
|
||||
FahrenheitThermometerAdaptor(sdbus::IConnection& connection, std::string objectPath)
|
||||
: sdbus::Interfaces<org::sdbuscpp::stresstest::fahrenheit::thermometer_adaptor>(connection, std::move(objectPath))
|
||||
, celsiusProxy_(connection, SERVICE_2_BUS_NAME, CELSIUS_THERMOMETER_OBJECT_PATH)
|
||||
{
|
||||
}
|
||||
|
||||
protected:
|
||||
virtual uint32_t getCurrentTemperature() override
|
||||
{
|
||||
// In this D-Bus call, make yet another D-Bus call to another service over the same connection
|
||||
return static_cast<uint32_t>(celsiusProxy_.getCurrentTemperature() * 1.8 + 32.);
|
||||
}
|
||||
|
||||
private:
|
||||
CelsiusThermometerProxy celsiusProxy_;
|
||||
};
|
||||
|
||||
class FahrenheitThermometerProxy : public sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::fahrenheit::thermometer_proxy>
|
||||
{
|
||||
public:
|
||||
using sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::fahrenheit::thermometer_proxy>::ProxyInterfaces;
|
||||
};
|
||||
|
||||
class ConcatenatorAdaptor : public sdbus::Interfaces<org::sdbuscpp::stresstest::concatenator_adaptor>
|
||||
{
|
||||
public:
|
||||
ConcatenatorAdaptor(sdbus::IConnection& connection, std::string objectPath)
|
||||
: sdbus::Interfaces<org::sdbuscpp::stresstest::concatenator_adaptor>(connection, std::move(objectPath))
|
||||
{
|
||||
unsigned int workers = std::thread::hardware_concurrency();
|
||||
if (workers < 4)
|
||||
workers = 4;
|
||||
|
||||
for (unsigned int i = 0; i < workers; ++i)
|
||||
workers_.emplace_back([this]()
|
||||
{
|
||||
//std::cout << "Created worker thread 0x" << std::hex << std::this_thread::get_id() << std::dec << std::endl;
|
||||
|
||||
while(!exit_)
|
||||
{
|
||||
// Pop a work item from the queue
|
||||
std::unique_lock<std::mutex> lock(mutex_);
|
||||
cond_.wait(lock, [this]{return !requests_.empty() || exit_;});
|
||||
if (exit_)
|
||||
break;
|
||||
auto request = std::move(requests_.front());
|
||||
requests_.pop();
|
||||
lock.unlock();
|
||||
|
||||
// Do concatenation work, return results and fire signal
|
||||
auto aString = request.input.at("key1").get<std::string>();
|
||||
auto aNumber = request.input.at("key2").get<uint32_t>();
|
||||
auto resultString = aString + " " + std::to_string(aNumber);
|
||||
|
||||
request.result.returnResults(resultString);
|
||||
|
||||
concatenatedSignal(resultString);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
~ConcatenatorAdaptor()
|
||||
{
|
||||
exit_ = true;
|
||||
cond_.notify_all();
|
||||
for (auto& worker : workers_)
|
||||
worker.join();
|
||||
}
|
||||
|
||||
protected:
|
||||
virtual void concatenate(sdbus::Result<std::string>&& result, std::map<std::string, sdbus::Variant> params) override
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(mutex_);
|
||||
requests_.push(WorkItem{std::move(params), std::move(result)});
|
||||
lock.unlock();
|
||||
cond_.notify_one();
|
||||
}
|
||||
|
||||
private:
|
||||
struct WorkItem
|
||||
{
|
||||
std::map<std::string, sdbus::Variant> input;
|
||||
sdbus::Result<std::string> result;
|
||||
};
|
||||
std::mutex mutex_;
|
||||
std::condition_variable cond_;
|
||||
std::queue<WorkItem> requests_;
|
||||
std::vector<std::thread> workers_;
|
||||
std::atomic<bool> exit_{};
|
||||
};
|
||||
|
||||
class ConcatenatorProxy : public sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::concatenator_proxy>
|
||||
{
|
||||
public:
|
||||
using sdbus::ProxyInterfaces<org::sdbuscpp::stresstest::concatenator_proxy>::ProxyInterfaces;
|
||||
|
||||
private:
|
||||
virtual void onConcatenateReply(const std::string& result, const sdbus::Error* error) override
|
||||
{
|
||||
assert(error == nullptr);
|
||||
|
||||
std::stringstream str(result);
|
||||
std::string aString;
|
||||
str >> aString;
|
||||
assert(aString == "sdbus-c++-stress-tests");
|
||||
|
||||
uint32_t aNumber;
|
||||
str >> aNumber;
|
||||
assert(aNumber >= 0);
|
||||
|
||||
++repliesReceived_;
|
||||
}
|
||||
|
||||
virtual void onConcatenatedSignal(const std::string& concatenatedString) override
|
||||
{
|
||||
std::stringstream str(concatenatedString);
|
||||
std::string aString;
|
||||
str >> aString;
|
||||
assert(aString == "sdbus-c++-stress-tests");
|
||||
|
||||
uint32_t aNumber;
|
||||
str >> aNumber;
|
||||
assert(aNumber >= 0);
|
||||
|
||||
++signalsReceived_;
|
||||
}
|
||||
|
||||
public:
|
||||
std::atomic<uint32_t> repliesReceived_;
|
||||
std::atomic<uint32_t> signalsReceived_;
|
||||
};
|
||||
|
||||
//-----------------------------------------
|
||||
int main(int /*argc*/, char */*argv*/[])
|
||||
{
|
||||
auto service2Connection = sdbus::createSystemBusConnection(SERVICE_2_BUS_NAME);
|
||||
std::thread service2Thread([&con = *service2Connection]()
|
||||
{
|
||||
CelsiusThermometerAdaptor thermometer(con, CELSIUS_THERMOMETER_OBJECT_PATH);
|
||||
con.enterProcessingLoop();
|
||||
});
|
||||
|
||||
auto service1Connection = sdbus::createSystemBusConnection(SERVICE_1_BUS_NAME);
|
||||
std::thread service1Thread([&con = *service1Connection]()
|
||||
{
|
||||
ConcatenatorAdaptor concatenator(con, CONCATENATOR_OBJECT_PATH);
|
||||
FahrenheitThermometerAdaptor thermometer(con, FAHRENHEIT_THERMOMETER_OBJECT_PATH);
|
||||
con.enterProcessingLoop();
|
||||
});
|
||||
|
||||
std::this_thread::sleep_for(100ms);
|
||||
|
||||
std::atomic<uint32_t> concatenationCallsMade{0};
|
||||
std::atomic<uint32_t> concatenationRepliesReceived{0};
|
||||
std::atomic<uint32_t> concatenationSignalsReceived{0};
|
||||
std::atomic<uint32_t> thermometerCallsMade{0};
|
||||
|
||||
auto clientConnection = sdbus::createSystemBusConnection();
|
||||
std::thread clientThread([&, &con = *clientConnection]()
|
||||
{
|
||||
std::atomic<bool> stopClients{false};
|
||||
|
||||
std::thread concatenatorThread([&]()
|
||||
{
|
||||
ConcatenatorProxy concatenator(con, SERVICE_1_BUS_NAME, CONCATENATOR_OBJECT_PATH);
|
||||
|
||||
uint32_t localCounter{};
|
||||
|
||||
// Issue async concatenate calls densely one after another
|
||||
while (!stopClients)
|
||||
{
|
||||
std::map<std::string, sdbus::Variant> param;
|
||||
param["key1"] = "sdbus-c++-stress-tests";
|
||||
param["key2"] = localCounter++;
|
||||
|
||||
concatenator.concatenate(param);
|
||||
|
||||
if ((localCounter % 10) == 0)
|
||||
{
|
||||
// Make sure the system is catching up with our async requests,
|
||||
// otherwise sleep a bit to slow down flooding the server.
|
||||
assert(localCounter >= concatenator.repliesReceived_);
|
||||
while ((localCounter - concatenator.repliesReceived_) > 20 && !stopClients)
|
||||
std::this_thread::sleep_for(2ms);
|
||||
|
||||
// Update statistics
|
||||
concatenationCallsMade = localCounter;
|
||||
concatenationRepliesReceived = (uint32_t)concatenator.repliesReceived_;
|
||||
concatenationSignalsReceived = (uint32_t)concatenator.signalsReceived_;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
std::thread thermometerThread([&]()
|
||||
{
|
||||
FahrenheitThermometerProxy thermometer(con, SERVICE_1_BUS_NAME, FAHRENHEIT_THERMOMETER_OBJECT_PATH);
|
||||
uint32_t localCounter{};
|
||||
uint32_t previousTemperature{};
|
||||
|
||||
while (!stopClients)
|
||||
{
|
||||
localCounter++;
|
||||
auto temperature = thermometer.getCurrentTemperature();
|
||||
assert(temperature >= previousTemperature); // The temperature shall rise continually
|
||||
previousTemperature = temperature;
|
||||
std::this_thread::sleep_for(5ms);
|
||||
|
||||
if ((localCounter % 10) == 0)
|
||||
thermometerCallsMade = localCounter;
|
||||
}
|
||||
});
|
||||
|
||||
con.enterProcessingLoop();
|
||||
|
||||
stopClients = true;
|
||||
concatenatorThread.join();
|
||||
thermometerThread.join();
|
||||
});
|
||||
|
||||
std::atomic<bool> exitLogger{};
|
||||
std::thread loggerThread([&]()
|
||||
{
|
||||
while (!exitLogger)
|
||||
{
|
||||
std::this_thread::sleep_for(1s);
|
||||
|
||||
std::cout << "Made " << concatenationCallsMade << " concatenation calls, received " << concatenationRepliesReceived << " replies and " << concatenationSignalsReceived << " signals so far." << std::endl;
|
||||
std::cout << "Made " << thermometerCallsMade << " thermometer calls so far." << std::endl << std::endl;
|
||||
}
|
||||
});
|
||||
|
||||
getchar();
|
||||
|
||||
exitLogger = true;
|
||||
loggerThread.join();
|
||||
clientConnection->leaveProcessingLoop();
|
||||
clientThread.join();
|
||||
service1Connection->leaveProcessingLoop();
|
||||
service1Thread.join();
|
||||
service2Connection->leaveProcessingLoop();
|
||||
service2Thread.join();
|
||||
|
||||
return 0;
|
||||
}
|
||||
Reference in New Issue
Block a user