diff --git a/src/spider/scheduler/SchedulerMessage.hpp b/src/spider/scheduler/SchedulerMessage.hpp index b7b61e1bb..b3d1e07a0 100644 --- a/src/spider/scheduler/SchedulerMessage.hpp +++ b/src/spider/scheduler/SchedulerMessage.hpp @@ -3,7 +3,6 @@ #include #include -#include #include #include @@ -55,28 +54,19 @@ class ScheduleTaskResponse { public: ScheduleTaskResponse() = default; - ScheduleTaskResponse( - boost::uuids::uuid const task_id, - boost::uuids::uuid const task_instance_id - ) - : m_task_ids{std::make_tuple(task_id, task_instance_id)} {} - - explicit ScheduleTaskResponse(std::tuple const& task_ids - ) - : m_task_ids{task_ids} {} + explicit ScheduleTaskResponse(boost::uuids::uuid const task_id) : m_task_id{task_id} {} - [[nodiscard]] auto has_task_id() const -> bool { return m_task_ids.has_value(); } + [[nodiscard]] auto has_task_id() const -> bool { return m_task_id.has_value(); } - [[nodiscard]] auto get_task_ids( - ) const -> std::tuple const& { + [[nodiscard]] auto get_task_id() const -> boost::uuids::uuid const& { // NOLINTNEXTLINE(bugprone-unchecked-optional-access) - return m_task_ids.value(); + return m_task_id.value(); } - MSGPACK_DEFINE_ARRAY(m_task_ids); + MSGPACK_DEFINE_ARRAY(m_task_id); private: - std::optional> m_task_ids = std::nullopt; + std::optional m_task_id = std::nullopt; }; } // namespace spider::scheduler diff --git a/src/spider/scheduler/SchedulerServer.cpp b/src/spider/scheduler/SchedulerServer.cpp index 48b0278c8..c59009986 100644 --- a/src/spider/scheduler/SchedulerServer.cpp +++ b/src/spider/scheduler/SchedulerServer.cpp @@ -12,7 +12,6 @@ #include #include "../core/Error.hpp" -#include "../core/Task.hpp" #include "../io/BoostAsio.hpp" // IWYU pragma: keep #include "../io/MsgPack.hpp" // IWYU pragma: keep #include "../io/msgpack_message.hpp" @@ -160,17 +159,7 @@ auto SchedulerServer::process_message(boost::asio::ip::tcp::socket socket = m_policy->schedule_next(request.get_worker_id(), request.get_worker_addr()); ScheduleTaskResponse response{}; if (task_id.has_value()) { - core::TaskInstance const instance{task_id.value()}; - core::StorageErr const err = m_metadata_store->create_task_instance(m_conn, instance); - if (err.success()) { - response = ScheduleTaskResponse{task_id.value(), instance.id}; - } else { - spdlog::error( - "Cannot create task instance {}: {}", - boost::uuids::to_string(task_id.value()), - err.description - ); - } + response = ScheduleTaskResponse{task_id.value()}; } msgpack::sbuffer response_buffer; msgpack::pack(response_buffer, response); diff --git a/src/spider/worker/WorkerClient.cpp b/src/spider/worker/WorkerClient.cpp index e59402245..b504bdc68 100644 --- a/src/spider/worker/WorkerClient.cpp +++ b/src/spider/worker/WorkerClient.cpp @@ -17,6 +17,7 @@ #include "../core/Driver.hpp" #include "../core/Error.hpp" +#include "../core/Task.hpp" #include "../io/BoostAsio.hpp" // IWYU pragma: keep #include "../io/MsgPack.hpp" // IWYU pragma: keep #include "../io/msgpack_message.hpp" @@ -112,7 +113,23 @@ auto WorkerClient::get_next_task(std::optional const& fail_t if (!response.has_task_id()) { return std::nullopt; } - return response.get_task_ids(); + boost::uuids::uuid const task_id = response.get_task_id(); + std::variant conn_result + = core::MySqlConnection::create(m_metadata_store->get_url()); + if (std::holds_alternative(conn_result)) { + spdlog::error( + "Failed to connect to storage: {}", + std::get(conn_result).description + ); + return std::nullopt; + } + auto& conn = std::get(conn_result); + core::TaskInstance const instance{task_id}; + core::StorageErr const err = m_metadata_store->create_task_instance(conn, instance); + if (!err.success()) { + return std::nullopt; + } + return std::make_tuple(task_id, instance.id); } catch (boost::system::system_error const& e) { return std::nullopt; } catch (std::runtime_error const& e) { diff --git a/tests/scheduler/test-SchedulerPolicy.cpp b/tests/scheduler/test-SchedulerPolicy.cpp index 66d78a640..eca670f5c 100644 --- a/tests/scheduler/test-SchedulerPolicy.cpp +++ b/tests/scheduler/test-SchedulerPolicy.cpp @@ -66,16 +66,28 @@ TEMPLATE_LIST_TEST_CASE( spider::scheduler::FifoPolicy policy{metadata_store, data_store, conn}; - // Scheduler the earlier task - std::optional const optional_task_id = policy.schedule_next(gen(), ""); + // Schedule the earlier task + std::optional optional_task_id = policy.schedule_next(gen(), ""); REQUIRE(optional_task_id.has_value()); if (optional_task_id.has_value()) { boost::uuids::uuid const& task_id = optional_task_id.value(); REQUIRE(task_id == task_1.get_id()); } + // Schedule the later task + optional_task_id = policy.schedule_next(gen(), ""); + REQUIRE(optional_task_id.has_value()); + if (optional_task_id.has_value()) { + boost::uuids::uuid const& task_id = optional_task_id.value(); + REQUIRE(task_id == task_2.get_id()); + } + REQUIRE(metadata_store->remove_job(conn, job_id_1).success()); REQUIRE(metadata_store->remove_job(conn, job_id_2).success()); + + // Schedule when no task available + optional_task_id = policy.schedule_next(gen(), ""); + REQUIRE(!optional_task_id.has_value()); } TEMPLATE_LIST_TEST_CASE( diff --git a/tests/scheduler/test-SchedulerServer.cpp b/tests/scheduler/test-SchedulerServer.cpp index 9233d3c1c..d0905393c 100644 --- a/tests/scheduler/test-SchedulerServer.cpp +++ b/tests/scheduler/test-SchedulerServer.cpp @@ -108,7 +108,7 @@ TEMPLATE_LIST_TEST_CASE( spider::scheduler::ScheduleTaskResponse const res = object.as(); REQUIRE(res.has_task_id()); - REQUIRE(std::get<0>(res.get_task_ids()) == parent_task.get_id()); + REQUIRE(res.get_task_id() == parent_task.get_id()); } socket.close(); server.stop();