diff --git a/src/spider/storage/mysql/mysql_stmt.hpp b/src/spider/storage/mysql/mysql_stmt.hpp index db5c9f2b4..81856febd 100644 --- a/src/spider/storage/mysql/mysql_stmt.hpp +++ b/src/spider/storage/mysql/mysql_stmt.hpp @@ -67,10 +67,10 @@ std::string const cCreateOutputTaskTable = R"(CREATE TABLE IF NOT EXISTS output_ std::string const cCreateTaskInputTable = R"(CREATE TABLE IF NOT EXISTS `task_inputs` ( `task_id` BINARY(16) NOT NULL, `position` INT UNSIGNED NOT NULL, - `type` VARCHAR(64) NOT NULL, + `type` VARCHAR(999) NOT NULL, `output_task_id` BINARY(16), `output_task_position` INT UNSIGNED, - `value` VARBINARY(64), -- Use VARBINARY for all types of values + `value` VARBINARY(999), -- Use VARBINARY for all types of values `data_id` BINARY(16), CONSTRAINT `input_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `input_task_output_match` FOREIGN KEY (`output_task_id`, `output_task_position`) REFERENCES task_outputs (`task_id`, `position`) ON UPDATE NO ACTION ON DELETE SET NULL, @@ -81,8 +81,8 @@ std::string const cCreateTaskInputTable = R"(CREATE TABLE IF NOT EXISTS `task_in std::string const cCreateTaskOutputTable = R"(CREATE TABLE IF NOT EXISTS `task_outputs` ( `task_id` BINARY(16) NOT NULL, `position` INT UNSIGNED NOT NULL, - `type` VARCHAR(64) NOT NULL, - `value` VARBINARY(64), + `type` VARCHAR(999) NOT NULL, + `value` VARBINARY(999), `data_id` BINARY(16), CONSTRAINT `output_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `output_data_id` FOREIGN KEY (`data_id`) REFERENCES `data` (`id`) ON UPDATE NO ACTION ON DELETE NO ACTION, @@ -108,7 +108,7 @@ std::string const cCreateTaskInstanceTable = R"(CREATE TABLE IF NOT EXISTS `task std::string const cCreateDataTable = R"(CREATE TABLE IF NOT EXISTS `data` ( `id` BINARY(16) NOT NULL, - `value` VARBINARY(256) NOT NULL, + `value` VARBINARY(999) NOT NULL, `hard_locality` BOOL DEFAULT FALSE, `persisted` BOOL DEFAULT FALSE, PRIMARY KEY (`id`) @@ -141,14 +141,14 @@ std::string const cCreateDataRefTaskTable = R"(CREATE TABLE IF NOT EXISTS `data_ std::string const cCreateClientKVDataTable = R"(CREATE TABLE IF NOT EXISTS `client_kv_data` ( `kv_key` VARCHAR(64) NOT NULL, - `value` VARBINARY(128) NOT NULL, + `value` VARBINARY(999) NOT NULL, `client_id` BINARY(16) NOT NULL, PRIMARY KEY (`client_id`, `kv_key`) ))"; std::string const cCreateTaskKVDataTable = R"(CREATE TABLE IF NOT EXISTS `task_kv_data` ( `kv_key` VARCHAR(64) NOT NULL, - `value` VARBINARY(128) NOT NULL, + `value` VARBINARY(999) NOT NULL, `task_id` BINARY(16) NOT NULL, PRIMARY KEY (`task_id`, `kv_key`), CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE diff --git a/src/spider/worker/TaskExecutorMessage.hpp b/src/spider/worker/TaskExecutorMessage.hpp index 234fbfc47..4fa8de1b8 100644 --- a/src/spider/worker/TaskExecutorMessage.hpp +++ b/src/spider/worker/TaskExecutorMessage.hpp @@ -37,29 +37,46 @@ enum class TaskExecutorRequestType : std::uint8_t { Resume, }; -inline auto get_request_type(msgpack::sbuffer const& buffer) -> TaskExecutorRequestType { - // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) - msgpack::object_handle const handle = msgpack::unpack(buffer.data(), buffer.size()); - msgpack::object const object = handle.get(); - if (object.type != msgpack::type::ARRAY || object.via.array.size < 2) { - return TaskExecutorRequestType::Unknown; +class TaskExecutorRequestParser { +public: + /** + * @param buffer + * @throw std::bad_cast if the buffer does not store a valid msgpack object + */ + explicit TaskExecutorRequestParser(msgpack::sbuffer const& buffer) + : m_obj(msgpack::unpack(buffer.data(), buffer.size())) {} + + /** + * @return The type of the message. + */ + [[nodiscard]] auto get_type() const -> TaskExecutorRequestType { + // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) + msgpack::object const object = m_obj.get(); + if (object.type != msgpack::type::ARRAY || object.via.array.size < 2) { + return TaskExecutorRequestType::Unknown; + } + msgpack::object const header = object.via.array.ptr[0]; + try { + return header.as(); + } catch (msgpack::type_error const&) { + return TaskExecutorRequestType::Unknown; + } + // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) } - msgpack::object const header = object.via.array.ptr[0]; - try { - return header.as(); - } catch (msgpack::type_error const&) { - return TaskExecutorRequestType::Unknown; + + /** + * @return The body of the message. Cannot outlive the `TaskExecutorRequestParser` object. + */ + [[nodiscard]] auto get_body() const -> msgpack::object { + // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) + msgpack::object const object = m_obj.get(); + return object.via.array.ptr[1]; + // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) } - // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) -} -inline auto get_message_body(msgpack::sbuffer const& buffer) -> msgpack::object { - // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) - msgpack::object_handle const handle = msgpack::unpack(buffer.data(), buffer.size()); - msgpack::object const object = handle.get(); - return object.via.array.ptr[1]; - // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic) -} +private: + msgpack::object_handle m_obj; +}; } // namespace spider::worker diff --git a/src/spider/worker/task_executor.cpp b/src/spider/worker/task_executor.cpp index f92a7fd6b..1b15c5d6b 100644 --- a/src/spider/worker/task_executor.cpp +++ b/src/spider/worker/task_executor.cpp @@ -142,13 +142,12 @@ auto main(int const argc, char** argv) -> int { return cFuncArgParseErr; } msgpack::sbuffer const& request_buffer = request_buffer_option.value(); - if (spider::worker::TaskExecutorRequestType::Arguments - != spider::worker::get_request_type(request_buffer)) - { + spider::worker::TaskExecutorRequestParser const request_parser{request_buffer}; + if (spider::worker::TaskExecutorRequestType::Arguments != request_parser.get_type()) { spdlog::error("Expect args request."); return cFuncArgParseErr; } - msgpack::object const args_object = spider::worker::get_message_body(request_buffer); + msgpack::object const args_object = request_parser.get_body(); msgpack::sbuffer args_buffer; msgpack::packer packer{args_buffer}; packer.pack(args_object); diff --git a/tests/client/client-test.cpp b/tests/client/client-test.cpp index 201973c34..a390e6160 100644 --- a/tests/client/client-test.cpp +++ b/tests/client/client-test.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -179,6 +180,33 @@ auto test_graph_batch_submission(spider::Driver& driver) -> int { return 0; } +auto test_large_input_output( + spider::Driver& driver, + size_t const input_size_1, + size_t const input_size_2 +) -> int { + std::string input_1(input_size_1, 'a'); + std::string input_2(input_size_2, 'b'); + + spider::Job job = driver.start(&join_string_test, input_1, input_2); + job.wait_complete(); + if (job.get_status() != spider::JobStatus::Succeeded) { + spdlog::error("Large input job failed"); + return cJobFailed; + } + if (job.get_result() != input_1 + input_2) { + spdlog::error( + "Large input job wrong result. Expect {}. Get {}.", + input_1 + input_2, + job.get_result() + ); + return cJobFailed; + } + return 0; +} + +constexpr size_t cLargeInputSize = 300; + } // namespace // NOLINTNEXTLINE(bugprone-exception-escape) @@ -243,5 +271,10 @@ auto main(int argc, char** argv) -> int { return result; } + result = test_large_input_output(driver, cLargeInputSize, cLargeInputSize); + if (0 != result) { + return result; + } + return 0; } diff --git a/tests/worker/test-TaskExecutor.cpp b/tests/worker/test-TaskExecutor.cpp index e48b0c4bc..bbea86504 100644 --- a/tests/worker/test-TaskExecutor.cpp +++ b/tests/worker/test-TaskExecutor.cpp @@ -208,6 +208,43 @@ TEMPLATE_LIST_TEST_CASE( REQUIRE(data_storage->remove_data(*conn, data.get_id()).success()); } +constexpr int cLargeInputSize = 300; + +TEMPLATE_LIST_TEST_CASE( + "Task execute large input&output", + "[worker][storage]", + spider::test::StorageFactoryTypeList +) { + absl::flat_hash_map< + boost::process::v2::environment::key, + boost::process::v2::environment::value> const environment_variable + = get_environment_variable(); + + boost::asio::io_context context; + + boost::uuids::random_generator gen; + + std::string const input_1(cLargeInputSize, 'a'); + std::string const input_2(cLargeInputSize, 'b'); + + spider::worker::TaskExecutor executor{ + context, + "join_string_test", + gen(), + spider::test::get_storage_url(), + get_libraries(), + environment_variable, + input_1, + input_2 + }; + context.run(); + executor.wait(); + REQUIRE(executor.succeed()); + std::optional const result_option = executor.get_result(); + REQUIRE(result_option.has_value()); + REQUIRE(input_1 + input_2 == result_option.value_or("")); +} + } // namespace // NOLINTEND(cert-err58-cpp,cppcoreguidelines-avoid-do-while,readability-function-cognitive-complexity,cppcoreguidelines-avoid-non-const-global-variables,cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays,clang-analyzer-unix.BlockInCriticalSection) diff --git a/tests/worker/worker-test.cpp b/tests/worker/worker-test.cpp index ed5840fca..1daf725f2 100644 --- a/tests/worker/worker-test.cpp +++ b/tests/worker/worker-test.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -61,6 +62,14 @@ auto create_task_test(spider::TaskContext& context, int x, int y) -> int { return job.get_result(); } +auto join_string_test( + spider::TaskContext& /*context*/, + std::string const& input_1, + std::string const& input_2 +) -> std::string { + return input_1 + input_2; +} + // NOLINTBEGIN(cert-err58-cpp) SPIDER_REGISTER_TASK(sum_test); SPIDER_REGISTER_TASK(swap_test); @@ -69,4 +78,5 @@ SPIDER_REGISTER_TASK(data_test); SPIDER_REGISTER_TASK(random_fail_test); SPIDER_REGISTER_TASK(create_data_test); SPIDER_REGISTER_TASK(create_task_test); +SPIDER_REGISTER_TASK(join_string_test); // NOLINTEND(cert-err58-cpp) diff --git a/tests/worker/worker-test.hpp b/tests/worker/worker-test.hpp index 691378642..f1624586a 100644 --- a/tests/worker/worker-test.hpp +++ b/tests/worker/worker-test.hpp @@ -1,6 +1,7 @@ #ifndef SPIDER_TEST_WORKER_TEST_HPP #define SPIDER_TEST_WORKER_TEST_HPP +#include #include #include @@ -20,4 +21,10 @@ auto create_data_test(spider::TaskContext& context, int x) -> spider::Data; auto create_task_test(spider::TaskContext& context, int x, int y) -> int; +auto join_string_test( + spider::TaskContext& context, + std::string const& input_1, + std::string const& input_2 +) -> std::string; + #endif diff --git a/tools/scripts/storage/init_db.sql b/tools/scripts/storage/init_db.sql index f002bb01c..8b4f3fd9f 100644 --- a/tools/scripts/storage/init_db.sql +++ b/tools/scripts/storage/init_db.sql @@ -58,7 +58,7 @@ CREATE TABLE IF NOT EXISTS output_tasks CREATE TABLE IF NOT EXISTS `data` ( `id` BINARY(16) NOT NULL, - `value` VARBINARY(256) NOT NULL, + `value` VARBINARY(999) NOT NULL, `hard_locality` BOOL DEFAULT FALSE, `persisted` BOOL DEFAULT FALSE, PRIMARY KEY (`id`) @@ -67,8 +67,8 @@ CREATE TABLE IF NOT EXISTS `task_outputs` ( `task_id` BINARY(16) NOT NULL, `position` INT UNSIGNED NOT NULL, - `type` VARCHAR(64) NOT NULL, - `value` VARBINARY(64), + `type` VARCHAR(999) NOT NULL, + `value` VARBINARY(999), `data_id` BINARY(16), CONSTRAINT `output_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `output_data_id` FOREIGN KEY (`data_id`) REFERENCES `data` (`id`) ON UPDATE NO ACTION ON DELETE NO ACTION, @@ -78,10 +78,10 @@ CREATE TABLE IF NOT EXISTS `task_inputs` ( `task_id` BINARY(16) NOT NULL, `position` INT UNSIGNED NOT NULL, - `type` VARCHAR(64) NOT NULL, + `type` VARCHAR(999) NOT NULL, `output_task_id` BINARY(16), `output_task_position` INT UNSIGNED, - `value` VARBINARY(64), -- Use VARBINARY for all types of values + `value` VARBINARY(999), -- Use VARBINARY for all types of values `data_id` BINARY(16), CONSTRAINT `input_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE, CONSTRAINT `input_task_output_match` FOREIGN KEY (`output_task_id`, `output_task_position`) REFERENCES task_outputs (`task_id`, `position`) ON UPDATE NO ACTION ON DELETE SET NULL, @@ -135,14 +135,14 @@ CREATE TABLE IF NOT EXISTS `data_ref_task` CREATE TABLE IF NOT EXISTS `client_kv_data` ( `kv_key` VARCHAR(64) NOT NULL, - `value` VARBINARY(128) NOT NULL, + `value` VARBINARY(999) NOT NULL, `client_id` BINARY(16) NOT NULL, PRIMARY KEY (`client_id`, `kv_key`) ); CREATE TABLE IF NOT EXISTS `task_kv_data` ( `kv_key` VARCHAR(64) NOT NULL, - `value` VARBINARY(128) NOT NULL, + `value` VARBINARY(999) NOT NULL, `task_id` BINARY(16) NOT NULL, PRIMARY KEY (`task_id`, `kv_key`), CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE