Compare commits

..

No commits in common. "2271702e6cff23e6e5ed9aa6e66d634238c3ffc1" and "4dcbf009da48c8c063777ae64ee46394687ff5a9" have entirely different histories.

19 changed files with 7 additions and 585 deletions

View File

@ -10,12 +10,6 @@
#include "work/admin/UserDirectory.hpp" #include "work/admin/UserDirectory.hpp"
#include "work/admin/UserSession.hpp" #include "work/admin/UserSession.hpp"
#include "work/jobs/Job.hpp"
#include "work/jobs/Task.hpp""
#include "work/jobs/Payload.hpp"
#include "work/jobs/IdPayload.hpp"
#include "work/jobs/IdListPayload.hpp"
#include "matador/object/schema.hpp" #include "matador/object/schema.hpp"
#include "matador/sql/connection.hpp" #include "matador/sql/connection.hpp"
@ -26,7 +20,7 @@ using namespace matador;
using namespace work::models; using namespace work::models;
int main() { int main() {
object::schema schema("Administration"); object::schema schema;
auto result = schema.attach<admin::CollectionCenter>("collection_center") auto result = schema.attach<admin::CollectionCenter>("collection_center")
.and_then([&schema] { return schema.attach<admin::InternalUserDirectory>("internal_user_directory"); }) .and_then([&schema] { return schema.attach<admin::InternalUserDirectory>("internal_user_directory"); })
@ -38,12 +32,7 @@ int main() {
.and_then([&schema] { return schema.attach<admin::Scenario>("scenario"); }) .and_then([&schema] { return schema.attach<admin::Scenario>("scenario"); })
.and_then([&schema] { return schema.attach<admin::User>("user"); }) .and_then([&schema] { return schema.attach<admin::User>("user"); })
.and_then([&schema] { return schema.attach<admin::UserDirectory>("user_directory"); }) .and_then([&schema] { return schema.attach<admin::UserDirectory>("user_directory"); })
.and_then([&schema] { return schema.attach<admin::UserSession>("user_session"); }) .and_then([&schema] { return schema.attach<admin::UserSession>("user_session"); });
.and_then([&schema] { return schema.attach<jobs::Job>("jobs"); })
.and_then([&schema] { return schema.attach<jobs::Payload>("id_payloads"); })
.and_then([&schema] { return schema.attach<jobs::IdPayload>("id_list_payloads"); })
.and_then([&schema] { return schema.attach<jobs::IdListPayload>("payloads"); })
.and_then([&schema] { return schema.attach<jobs::Task>("tasks"); });
if (!result.is_ok()) { if (!result.is_ok()) {
return 0; return 0;
@ -62,8 +51,3 @@ int main() {
return 0; return 0;
} }
// registering table 'collection_center' (pk field: 'id')
// registering relation table 'collection_center_users' (first fk field: 'collection_center_id', second fk field: 'user_id')
// registering table 'internal_user_directory' (pk field: 'id')
//

View File

@ -38,7 +38,7 @@ struct CollectionCenter : core::Model {
field::attribute( op, "name", name, 511 ); field::attribute( op, "name", name, 511 );
field::attribute( op, "symbol", symbol ); field::attribute( op, "symbol", symbol );
field::attribute( op, "type", type ); field::attribute( op, "type", type );
field::has_many( op, "collection_center_users", users, "collection_center_id", matador::utils::fetch_type::LAZY ); field::has_many( op, "users", users, "collection_center_id", matador::utils::fetch_type::LAZY );
} }
}; };

View File

@ -1,22 +0,0 @@
#ifndef USERINFO_HPP
#define USERINFO_HPP
#include "matador/utils/access.hpp"
namespace work::core {
struct UserInfo {
unsigned long long user_session_id;
std::string user_role;
std::string database_name;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::attribute( op, "user_session_id", user_session_id );
field::attribute( op, "user_role", user_role, 255 );
field::attribute( op, "database_name", database_name, 255 );
}
};
}
#endif //USERINFO_HPP

View File

@ -1,20 +0,0 @@
#ifndef ID_LIST_PAYLOAD_HPP
#define ID_LIST_PAYLOAD_HPP
#include "Payload.hpp"
#include "matador/object/collection.hpp"
namespace work::models::jobs {
struct IdListPayload : Payload {
matador::object::collection<unsigned long long> ids;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::process( op, *matador::base_class<Payload>( this ) );
field::has_many( op, "payload_ids", ids );
}
};
}
#endif //ID_LIST_PAYLOAD_HPP

View File

@ -1,18 +0,0 @@
#ifndef ID_PAYLOAD_HPP
#define ID_PAYLOAD_HPP
#include "Payload.hpp"
namespace work::models::jobs {
struct IdPayload : Payload {
unsigned long long id;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::process( op, *matador::base_class<Payload>( this ) );
field::attribute( op, "id", id );
}
};
}
#endif //ID_PAYLOAD_HPP

View File

@ -1,46 +0,0 @@
#ifndef JOB_HPP
#define JOB_HPP
#include "JobMode.hpp"
#include "JobState.hpp"
#include "Payload.hpp"
#include "Task.hpp"
#include "../core/Model.hpp"
#include "../core/Types.hpp"
#include "../core/UserInfo.hpp"
#include "matador/object/object_ptr.hpp"
#include "matador/utils/base_class.hpp"
namespace work::models::jobs {
struct Job : core::Model {
std::string name;
std::string description;
JobState state;
JobMode mode;
core::timestamp created_at;
matador::object::object_ptr<Payload> payload;
matador::object::object_ptr<Task> task;
core::UserInfo user_info;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::process( op, *matador::base_class<Model>( this ) );
field::attribute( op, "name", name, 511 );
field::attribute( op, "description", description, 511 );
field::attribute( op, "state", state );
field::attribute( op, "mode", mode );
field::attribute( op, "created_at", created_at );
field::belongs_to( op, "payload", payload, matador::utils::default_foreign_attributes );
field::belongs_to( op, "task", task, matador::utils::default_foreign_attributes );
field::attribute( op, "user_info", user_info );
}
};
}
#endif //JOB_HPP

View File

@ -1,20 +0,0 @@
#ifndef JOB_MODE_HPP
#define JOB_MODE_HPP
#include "matador/utils/enum_mapper.hpp"
#include <cstdint>
namespace work::models::jobs {
enum class JobMode : uint8_t {
Background,
Foreground
};
static const matador::utils::enum_mapper<JobMode> JobModeEnum({
{JobMode::Background, "Background"},
{JobMode::Foreground, "Foreground"}
});
}
#endif //JOB_MODE_HPP

View File

@ -1,28 +0,0 @@
#ifndef JOB_STATE_HPP
#define JOB_STATE_HPP
#include "matador/utils/enum_mapper.hpp"
#include <cstdint>
namespace work::models::jobs {
enum class JobState : uint8_t {
Pending,
Running,
Succeeded,
Failed,
Cancelled
};
static const matador::utils::enum_mapper<JobState> JobStateEnum({
{JobState::Pending, "Pending"},
{JobState::Running, "Running"},
{JobState::Succeeded, "Succeeded"},
{JobState::Failed, "Failed"},
{JobState::Cancelled, "Cancelled"}
});
}
#endif //JOB_STATE_HPP

View File

@ -1,20 +0,0 @@
#ifndef PAYLOAD_HPP
#define PAYLOAD_HPP
#include "../core/Model.hpp"
#include "matador/utils/base_class.hpp"
namespace work::models::jobs {
struct Payload : core::Model {
std::string type;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::process( op, *matador::base_class<Model>( this ) );
field::attribute( op, "type", type, 255 );
}
};
}
#endif //PAYLOAD_HPP

View File

@ -1,42 +0,0 @@
#ifndef TASK_HPP
#define TASK_HPP
#include "TaskState.hpp"
#include "../core/Model.hpp"
#include "../core/Types.hpp"
#include "matador/object/object_ptr.hpp"
#include "matador/utils/base_class.hpp"
namespace work::models::jobs {
struct Task : core::Model {
std::string name;
std::string description;
std::string job_name;
TaskState state;
matador::object::object_ptr<Payload> payload;
JobMode job_mode;
core::timestamp start_delay;
core::timestamp interval;
unsigned long long user_session_id;
template<typename Operator>
void process( Operator& op ) {
namespace field = matador::access;
field::process( op, *matador::base_class<Model>( this ) );
field::attribute( op, "name", name, 511 );
field::attribute( op, "description", description, 511 );
field::attribute( op, "job_name", job_name, 511 );
field::attribute( op, "state", state );
field::belongs_to( op, "payload", payload, matador::utils::default_foreign_attributes );
field::attribute( op, "job_mode", job_mode );
field::attribute( op, "start_delay", start_delay );
field::attribute( op, "interval", interval );
field::attribute( op, "user_session_id", user_session_id );
}
};
}
#endif //TASK_HPP

View File

@ -1,22 +0,0 @@
#ifndef TASK_STATE_HPP
#define TASK_STATE_HPP
#include "matador/utils/enum_mapper.hpp"
#include <cstdint>
namespace work::models::jobs {
enum class TaskState : uint8_t {
Active,
Inactive
};
static const matador::utils::enum_mapper<TaskState> TaskStateEnum({
{TaskState::Active, "Active"},
{TaskState::Inactive, "Inactive"}
});
}
#endif //TASK_STATE_HPP

View File

@ -1,7 +1,7 @@
#ifndef SCHEMA_HPP #ifndef SCHEMA_HPP
#define SCHEMA_HPP #define SCHEMA_HPP
#include "matador/object/primary_key_resolver.hpp" #include "primary_key_resolver.hpp"
#include "matador/object/error_code.hpp" #include "matador/object/error_code.hpp"
#include "matador/object/schema_node.hpp" #include "matador/object/schema_node.hpp"
#include "matador/object/schema_node_iterator.hpp" #include "matador/object/schema_node_iterator.hpp"

View File

@ -11,7 +11,7 @@ namespace matador::utils {
template < typename ValueType, typename ErrorType > template < typename ValueType, typename ErrorType >
class [[nodiscard]] result; class [[nodiscard]] result;
template <typename> template <typename ValueType>
struct is_result : std::false_type {}; struct is_result : std::false_type {};
template < typename ValueType, typename ErrorType > template < typename ValueType, typename ErrorType >

View File

@ -1,145 +0,0 @@
#ifndef MATADOR_THREAD_POOL_HPP
#define MATADOR_THREAD_POOL_HPP
#include "matador/utils/result.hpp"
#include <thread>
#include <vector>
#include <future>
#include <atomic>
#include <functional>
#include <deque>
#include <mutex>
#include <condition_variable>
namespace matador::utils {
class cancel_token {
public:
cancel_token() = default;
void request_cancel() {
cancelled_.store(true, std::memory_order_relaxed);
}
bool is_cancelled() const {
return cancelled_.load(std::memory_order_relaxed);
}
private:
std::atomic_bool cancelled_{false};
};
template<typename T>
using result_fut = result<std::future<T>, std::string>;
/**
* Simple thread pool class. The pool
* consists of a given number of threads and
* a queue of tasks to be executed.
*/
class thread_pool {
public:
thread_pool(const thread_pool&) = delete;
thread_pool& operator=(const thread_pool&) = delete;
/**
* Creating a new thread pool with given
* numbers of threads.
*
* @param size Number of provided threads.
*/
explicit thread_pool(std::size_t size);
~thread_pool();
/**
* Push a task into the thread pool
* to be executed once a thread is available
*
* @tparam F Type of threaded function
* @tparam Args Types of the arguments
* @param func Function to be executed in the next available thread.
* @param args Passed arguments
*/
template <typename F, typename... Args>
auto schedule(F&& func, Args&&... args) -> result_fut<std::result_of_t<F(cancel_token&, Args...)>> {
using return_type = std::result_of_t<F(cancel_token&, Args...)>;
const auto token = std::make_shared<cancel_token>();
auto task_ptr = std::make_shared<std::packaged_task<return_type()>>(
std::bind(std::forward<F>(func), std::ref(*token), std::forward<Args>(args)...)
);
std::future<return_type> res = task_ptr->get_future();
{
std::lock_guard lock(mutex_);
if (!running_) {
return failure(std::string("Thread pool is shut down, cannot schedule new tasks."));
}
tasks_.emplace_back([task_ptr, token] {
try { (*task_ptr)(); }
catch (...) { /* Prevent exception escape */ }
}, token);
}
condition_task_.notify_one();
return ok(std::move(res));
}
/**
* Returns the number of available threads.
*
* @return Number of threads.
*/
std::size_t size() const;
/**
* Shuts the thread pool down.
*/
void shutdown();
/**
* Waits until all threads are finished.
*/
void wait();
void cancel_all_in_progress() const {
std::lock_guard lock(mutex_);
for (auto& t : tasks_) {
if (t.token)
t.token->request_cancel();
}
}
/**
* Returns the number of pending tasks.
*
* @return Number of pending tasks.
*/
std::size_t pending() const;
private:
/*
* wait for a task to execute
*/
void execute();
private:
struct task {
std::function<void()> func;
std::shared_ptr<cancel_token> token;
task(std::function<void()> f, std::shared_ptr<cancel_token> t) : func(std::move(f)), token(std::move(t)) {}
};
std::deque<task> tasks_;
std::vector<std::thread> threads_;
mutable std::mutex mutex_;
std::condition_variable condition_task_;
std::condition_variable condition_finished_;
std::atomic_uint busy_{0};
bool running_{true};
};
}
#endif //MATADOR_THREAD_POOL_HPP

View File

@ -39,7 +39,6 @@ add_library(matador-core STATIC
../../include/matador/utils/result.hpp ../../include/matador/utils/result.hpp
../../include/matador/utils/singleton.hpp ../../include/matador/utils/singleton.hpp
../../include/matador/utils/string.hpp ../../include/matador/utils/string.hpp
../../include/matador/utils/thread_pool.hpp
../../include/matador/utils/types.hpp ../../include/matador/utils/types.hpp
../../include/matador/utils/uuid.hpp ../../include/matador/utils/uuid.hpp
../../include/matador/utils/value.hpp ../../include/matador/utils/value.hpp
@ -62,7 +61,6 @@ add_library(matador-core STATIC
utils/library.cpp utils/library.cpp
utils/os.cpp utils/os.cpp
utils/string.cpp utils/string.cpp
utils/thread_pool.cpp
utils/types.cpp utils/types.cpp
utils/uuid.cpp utils/uuid.cpp
utils/value.cpp utils/value.cpp

View File

@ -1,84 +0,0 @@
#include <thread>
#include <algorithm>
#include "matador/utils/thread_pool.hpp"
namespace matador::utils {
thread_pool::thread_pool(const std::size_t size) {
if (size == 0) {
throw std::invalid_argument("Thread pool size must be positive");
}
for (std::size_t i = 0; i < size; ++i) {
threads_.emplace_back([this] { this->execute(); });
}
}
thread_pool::~thread_pool() {
shutdown();
wait();
}
std::size_t thread_pool::size() const {
return threads_.size();
}
void thread_pool::shutdown() {
{
std::lock_guard lock(mutex_);
running_ = false;
}
condition_task_.notify_all();
}
void thread_pool::wait() {
for (std::thread& t : threads_) {
if (t.joinable()) {
t.join();
}
}
threads_.clear();
}
std::size_t thread_pool::pending() const {
const std::lock_guard l(mutex_);
return tasks_.size();
}
void thread_pool::execute()
{
while (true) {
task t{nullptr, nullptr};
{
std::unique_lock l(mutex_);
condition_task_.wait(l, [this] {
return !running_ ||!tasks_.empty();
});
if (!running_ && tasks_.empty()) {
break;
}
if (!tasks_.empty()) {
t = std::move(tasks_.front());
tasks_.pop_front();
++busy_;
if (t.token) {
t.token->request_cancel(); // or signal task as "started"
}
} else {
continue;
}
}
if (t.func) {
try { t.func(); } catch (...) {}
}
{
std::lock_guard lock(mutex_);
--busy_;
if (tasks_.empty() && busy_ == 0)
condition_finished_.notify_all();
}
}
}
}

View File

@ -11,7 +11,6 @@ add_executable(CoreTests
utils/ResultTest.cpp utils/ResultTest.cpp
utils/FieldAttributeTest.cpp utils/FieldAttributeTest.cpp
utils/VersionTest.cpp utils/VersionTest.cpp
utils/ThreadPoolTest.cpp
utils/StringTest.cpp utils/StringTest.cpp
object/AttributeDefinitionGeneratorTest.cpp object/AttributeDefinitionGeneratorTest.cpp
object/PrimaryKeyResolverTest.cpp object/PrimaryKeyResolverTest.cpp

View File

@ -1,92 +0,0 @@
#include <catch2/catch_test_macros.hpp>
#include "matador/utils/thread_pool.hpp"
using namespace matador::utils;
TEST_CASE("thread_pool creates and rejects zero-sized construction") {
REQUIRE_THROWS(thread_pool(0));
const thread_pool tp(2);
REQUIRE(tp.size() == 2);
}
TEST_CASE("thread_pool schedules and runs simple tasks") {
thread_pool pool(3);
auto fut = pool.schedule([](cancel_token&, const int x) { return x * 10; }, 7);
REQUIRE(fut);
REQUIRE(fut.value().get() == 70);
}
TEST_CASE("thread_pool parallel computation and results") {
thread_pool pool(4);
std::vector<std::future<int>> futs;
for (int i = 1; i <= 20; ++i) {
auto res = pool.schedule([](cancel_token&, const int y) { return y + 1; }, i);
REQUIRE(res);
futs.push_back(std::move(res.value()));
}
int sum = 0;
for (auto& f : futs) {
sum += f.get();
}
REQUIRE(sum == 230); // (2+3+...+21) = (21*22/2)-(1) = 231-1
}
TEST_CASE("thread_pool rejects scheduling after shutdown") {
thread_pool pool(2);
pool.shutdown();
auto result = pool.schedule([](cancel_token&) { return 1; });
REQUIRE_FALSE(result);
REQUIRE(result.err().find("shut down") != std::string::npos);
}
TEST_CASE("thread_pool supports cancel_token for not-yet-run task") {
thread_pool pool(1);
std::atomic ran{false};
// Fill the only worker
auto fut1 = std::move(pool.schedule([](cancel_token&, const int delay) {
std::this_thread::sleep_for(std::chrono::milliseconds(delay));
return 123;
}, 250).value());
// This will wait in the queue
std::shared_ptr<cancel_token> ct;
auto result = pool.schedule([&ran]( const cancel_token& token) {
ran = true;
for (int i = 0; i < 10; ++i) {
if (token.is_cancelled())
return -1;
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
return 100;
});
REQUIRE(result);
// ct = std::make_shared<cancel_token>(); // Dummy; real token is internal
pool.cancel_all_in_progress();
// Give time for first task to finish and second to start
int v = result.value().get();
REQUIRE((v == -1 || v == 100)); // Cancel best-effort, may not run at all
// If cancelled while still not started, should be -1
// If started before cancel, may be 100 (rare but possible)
}
TEST_CASE("thread_pool: shutdown finishes all running tasks") {
thread_pool pool(2);
std::atomic counter{0};
std::vector<std::future<void>> futs;
for (int i = 0; i < 4; ++i) {
auto fut = pool.schedule([&counter](cancel_token&) {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
++counter;
});
REQUIRE(fut);
futs.push_back(std::move(fut.value()));
}
pool.shutdown();
pool.wait();
for (auto& f : futs) { f.get(); }
REQUIRE(counter == 4);
}

View File

@ -42,7 +42,7 @@ TEST_CASE("Create sql insert for entity with eager has one", "[query][entity][in
auto b737 = object_ptr(new airplane{0, "Boeing", "737"}); auto b737 = object_ptr(new airplane{0, "Boeing", "737"});
flight f1{0, b737, "F828"}; flight f1{0, b737, "F828"};
// auto data = eib.build(f1); auto data = eib.build(f1);
// REQUIRE(data.is_ok()); REQUIRE(data.is_ok());
} }