github-actions[bot] commented on code in PR #28924: URL: https://github.com/apache/doris/pull/28924#discussion_r1435673135
########## be/src/common/sync_point.cpp: ########## @@ -0,0 +1,236 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +// clang-format off +#include "sync_point.h" + +#include <atomic> +#include <condition_variable> +#include <functional> +#include <mutex> +#include <random> +#include <string> +#include <thread> +#include <unordered_map> +#include <unordered_set> + +namespace doris { + +struct SyncPoint::Data { // impl +public: + Data() : enabled_(false) { } + virtual ~Data() {} + void process(const std::string& point, std::vector<std::any>&& cb_args); + void load_dependency(const std::vector<SyncPointPair>& dependencies); + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + bool predecessors_all_cleared(const std::string& point); + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + void clear_call_back(const std::string& point); + void clear_all_call_backs(); + void enable_processing(); + void disable_processing(); + void clear_trace(); +private: + bool disable_by_marker(const std::string& point, std::thread::id thread_id); +private: + // successor/predecessor map loaded from load_dependency + std::unordered_map<std::string, std::vector<std::string>> successors_; + std::unordered_map<std::string, std::vector<std::string>> predecessors_; + std::unordered_map<std::string, std::function<void(std::vector<std::any>&&)>> callbacks_; + std::unordered_map<std::string, std::vector<std::string>> markers_; + std::unordered_map<std::string, std::thread::id> marked_thread_id_; + std::mutex mutex_; + std::condition_variable cv_; + // sync points that have been passed through + std::unordered_set<std::string> cleared_points_; + std::atomic<bool> enabled_; + int num_callbacks_running_ = 0; +}; + +SyncPoint* SyncPoint::get_instance() { + static SyncPoint sync_point; + return &sync_point; +} +SyncPoint::SyncPoint() : + impl_(new Data) { +} +SyncPoint:: ~SyncPoint() { + delete impl_; +} +void SyncPoint::load_dependency(const std::vector<SyncPointPair>& dependencies) { + impl_->load_dependency(dependencies); +} +void SyncPoint::load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers) { + impl_->load_dependency_and_markers(dependencies, markers); +} +void SyncPoint::set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback) { + impl_->set_call_back(point, callback); +} +void SyncPoint::clear_call_back(const std::string& point) { + impl_->clear_call_back(point); +} +void SyncPoint::clear_all_call_backs() { + impl_->clear_all_call_backs(); +} +void SyncPoint::enable_processing() { + impl_->enable_processing(); +} +void SyncPoint::disable_processing() { + impl_->disable_processing(); +} +void SyncPoint::clear_trace() { + impl_->clear_trace(); +} +void SyncPoint::process(const std::string& point, std::vector<std::any>&& cb_arg) { + impl_->process(point, std::move(cb_arg)); +} + +// ============================================================================= +// SyncPoint implementation +// ============================================================================= + +void SyncPoint::Data::load_dependency( + const std::vector<SyncPointPair>& dependencies) { + std::lock_guard lock(mutex_); + successors_.clear(); + predecessors_.clear(); + cleared_points_.clear(); + for (const auto& dependency : dependencies) { + successors_[dependency.predecessor].push_back(dependency.successor); + predecessors_[dependency.successor].push_back(dependency.predecessor); + } + cv_.notify_all(); +} + +/** + * Markers are also dependency descriptions + */ +void SyncPoint::Data::load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers) { + std::lock_guard lock(mutex_); + successors_.clear(); + predecessors_.clear(); + cleared_points_.clear(); + markers_.clear(); + marked_thread_id_.clear(); + for (const auto& dependency : dependencies) { + successors_[dependency.predecessor].push_back(dependency.successor); + predecessors_[dependency.successor].push_back(dependency.predecessor); + } + for (const auto& marker : markers) { + successors_[marker.predecessor].push_back(marker.successor); + predecessors_[marker.successor].push_back(marker.predecessor); + markers_[marker.predecessor].push_back(marker.successor); + } + cv_.notify_all(); +} + +bool SyncPoint::Data::predecessors_all_cleared(const std::string& point) { + for (const auto& pred : predecessors_[point]) { + if (cleared_points_.count(pred) == 0) { Review Comment: warning: use 'contains' to check for membership [readability-container-contains] ```suggestion if (!cleared_points_.contains(pred)) { ``` ########## be/src/common/sync_point.h: ########## @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +#pragma once +// clang-format off +#include <functional> +#include <iostream> +#include <string> +#include <vector> +#include <any> + +namespace doris { + +#define SYNC_POINT_HOOK_RETURN_VALUE(expr, point_name, ...) \ + [&]() mutable { \ + TEST_SYNC_POINT_RETURN_WITH_VALUE(point_name, decltype((expr)) {}, __VA_ARGS__); \ + return (expr); \ + }() + +// This class provides facility to reproduce race conditions deterministically +// in unit tests. +// Developer could specify sync points in the codebase via TEST_SYNC_POINT. +// Each sync point represents a position in the execution stream of a thread. +// In the unit test, 'Happens After' relationship among sync points could be +// setup via SyncPoint::load_dependency, to reproduce a desired interleave of +// threads execution. +// Refer to (DBTest,TransactionLogIteratorRace), for an example use case. +class SyncPoint { +public: + static SyncPoint* get_instance(); + SyncPoint(const SyncPoint&) = delete; + SyncPoint& operator=(const SyncPoint&) = delete; + ~SyncPoint(); + struct SyncPointPair { + std::string predecessor; + std::string successor; + }; + + // call once at the beginning of a test to setup the dependency between + // sync points + // + // Example: + // load_dependency({{"point1", "point2"}, + // {"point2", "point3"}, + // {"point3", "point4"}}); + // + // test case thread thread for object being tested + // | | + // | | + // | \-------------0-------------\ | + // | \-> x sync point1 set in code + // | /----------1----------------/ | + // point2 o <-/ /-> x sync point4 set in code + // | / | + // z / | + // z /---------2-----------/ | there may be nothing + // | / | between point1 point4 + // ponit3 o --/ | they are for sync + // | | between test case and object + // v v + // + // vertical arrow means the procedure of each thread, the running order will + // be: + // test case thread -> point1 -> point2 -> point3 -> point4 -> object being + // tested + // + // we may do a lot of things between point2 and point3, say, change the + // object's status, call another method, propagate data race and etc. + void load_dependency(const std::vector<SyncPointPair>& dependencies); + + // call once at the beginning of a test to setup the dependency between + // sync points and setup markers indicating the successor is only enabled + // when it is processed on the same thread as the predecessor. + // When adding a marker, it implicitly adds a dependency for the marker pair. + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + + // The argument to the callback is passed through from + // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or + // TEST_IDX_SYNC_POINT was used. + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + + // Clear callback function by point + void clear_call_back(const std::string& point); + + // Clear all call back functions. + void clear_all_call_backs(); + + // Enable sync point processing (disabled on startup) + void enable_processing(); + + // Disable sync point processing + void disable_processing(); + + // Remove the execution trace of all sync points + void clear_trace(); + + // Triggered by TEST_SYNC_POINT, blocking execution until all predecessors + // are executed. + // And/or call registered callback function, with argument `cb_args` + void process(const std::string& point, std::vector<std::any>&& cb_args = {}); + + // TODO: it might be useful to provide a function that blocks until all + // sync points are cleared. + // We want this to be public so we can subclass the implementation + struct Data; + +private: + // Singleton + SyncPoint(); + Data* impl_; // impletation which is hidden in cpp file +}; + +template <class T> +T try_any_cast(const std::any& a) { + try { + return std::any_cast<T>(a); + } catch (const std::bad_any_cast& e) { + std::cerr << e.what() << " expected=" << typeid(T).name() << " actual=" << a.type().name() << std::endl; + throw e; + } +} + +template <typename T> +auto try_any_cast_ret(std::vector<std::any>& any) { + return try_any_cast<std::pair<T, bool>*>(any.back()); +} + +} // namespace doris + +#define SYNC_POINT(x) doris::SyncPoint::get_instance()->process(x) +#define IDX_SYNC_POINT(x, index) \ + doris::SyncPoint::get_instance()->process(x + std::to_string(index)) +#define SYNC_POINT_CALLBACK(x, ...) doris::SyncPoint::get_instance()->process(x, {__VA_ARGS__}) +#define SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) \ +{ \ + std::pair ret {default_ret_val, false}; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&ret); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (ret.second) return std::move(ret.first); \ +} +#define SYNC_POINT_RETURN_WITH_VOID(x, ...) \ +{ \ + bool pred = false; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&pred); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (pred) return; \ +} +#define SYNC_POINT_SINGLETON() (void)doris::SyncPoint::get_instance() + +// TEST_SYNC_POINT is no op in release build. +// Turn on this feature by defining the macro +#ifndef BE_TEST +# define TEST_SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) +// seldom called +# define TEST_SYNC_POINT_SINGLETON() +#else +// Use TEST_SYNC_POINT to specify sync points inside code base. +// Sync points can have happens-after depedency on other sync points, +// configured at runtime via SyncPoint::load_dependency. This could be +// utilized to re-produce race conditions between threads. +# define TEST_SYNC_POINT(x) SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__) +# define TEST_SYNC_POINT_SINGLETON() SYNC_POINT_SINGLETON() + +/** + * Inject return points for testing. + * + * Currently we can only insert more points to get context from tested thread + * and process in testing thread, e.g. + * + * tested thread: + * ... + * TEST_SYNC_POINT_RETURN_WITH_VALUE("point_ret", int(0), ctx0); + * ... + * + * testing thread: + * sync_point->add("point_ret", [](auto&& args) { + * auto ctx0 = try_any_cast<bool>(args[0]); + * auto pair = try_any_cast<std::pair<int, bool>*>(args.back()); + * pair->first = ...; + * pair->second = ctx0; }); + * + * See sync_piont_test.cpp for more details. + */ +#pragma GCC diagnostic ignored "-Waddress" +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__) + +#endif // BE_TEST + +// TODO: define injection point in production env. +// the `if` expr can be live configure of the application +#ifndef ENABLE_INJECTION_POINT +# define TEST_INJECTION_POINT(x) +# define TEST_IDX_INJECTION_POINT(x, index) +# define TEST_INJECTION_POINT_CALLBACK(x, ...) +# define TEST_INJECTION_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) Review Comment: warning: macro is not used [clang-diagnostic-unused-macros] ```cpp # define TEST_INJECTION_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) ^ ``` ########## be/src/common/sync_point.h: ########## @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +#pragma once +// clang-format off +#include <functional> +#include <iostream> +#include <string> +#include <vector> +#include <any> + +namespace doris { + +#define SYNC_POINT_HOOK_RETURN_VALUE(expr, point_name, ...) \ + [&]() mutable { \ + TEST_SYNC_POINT_RETURN_WITH_VALUE(point_name, decltype((expr)) {}, __VA_ARGS__); \ + return (expr); \ + }() + +// This class provides facility to reproduce race conditions deterministically +// in unit tests. +// Developer could specify sync points in the codebase via TEST_SYNC_POINT. +// Each sync point represents a position in the execution stream of a thread. +// In the unit test, 'Happens After' relationship among sync points could be +// setup via SyncPoint::load_dependency, to reproduce a desired interleave of +// threads execution. +// Refer to (DBTest,TransactionLogIteratorRace), for an example use case. +class SyncPoint { +public: + static SyncPoint* get_instance(); + SyncPoint(const SyncPoint&) = delete; + SyncPoint& operator=(const SyncPoint&) = delete; + ~SyncPoint(); + struct SyncPointPair { + std::string predecessor; + std::string successor; + }; + + // call once at the beginning of a test to setup the dependency between + // sync points + // + // Example: + // load_dependency({{"point1", "point2"}, + // {"point2", "point3"}, + // {"point3", "point4"}}); + // + // test case thread thread for object being tested + // | | + // | | + // | \-------------0-------------\ | + // | \-> x sync point1 set in code + // | /----------1----------------/ | + // point2 o <-/ /-> x sync point4 set in code + // | / | + // z / | + // z /---------2-----------/ | there may be nothing + // | / | between point1 point4 + // ponit3 o --/ | they are for sync + // | | between test case and object + // v v + // + // vertical arrow means the procedure of each thread, the running order will + // be: + // test case thread -> point1 -> point2 -> point3 -> point4 -> object being + // tested + // + // we may do a lot of things between point2 and point3, say, change the + // object's status, call another method, propagate data race and etc. + void load_dependency(const std::vector<SyncPointPair>& dependencies); + + // call once at the beginning of a test to setup the dependency between + // sync points and setup markers indicating the successor is only enabled + // when it is processed on the same thread as the predecessor. + // When adding a marker, it implicitly adds a dependency for the marker pair. + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + + // The argument to the callback is passed through from + // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or + // TEST_IDX_SYNC_POINT was used. + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + + // Clear callback function by point + void clear_call_back(const std::string& point); + + // Clear all call back functions. + void clear_all_call_backs(); + + // Enable sync point processing (disabled on startup) + void enable_processing(); + + // Disable sync point processing + void disable_processing(); + + // Remove the execution trace of all sync points + void clear_trace(); + + // Triggered by TEST_SYNC_POINT, blocking execution until all predecessors + // are executed. + // And/or call registered callback function, with argument `cb_args` + void process(const std::string& point, std::vector<std::any>&& cb_args = {}); + + // TODO: it might be useful to provide a function that blocks until all + // sync points are cleared. + // We want this to be public so we can subclass the implementation + struct Data; + +private: + // Singleton + SyncPoint(); + Data* impl_; // impletation which is hidden in cpp file +}; + +template <class T> +T try_any_cast(const std::any& a) { + try { + return std::any_cast<T>(a); + } catch (const std::bad_any_cast& e) { + std::cerr << e.what() << " expected=" << typeid(T).name() << " actual=" << a.type().name() << std::endl; + throw e; + } +} + +template <typename T> +auto try_any_cast_ret(std::vector<std::any>& any) { + return try_any_cast<std::pair<T, bool>*>(any.back()); +} + +} // namespace doris + +#define SYNC_POINT(x) doris::SyncPoint::get_instance()->process(x) +#define IDX_SYNC_POINT(x, index) \ + doris::SyncPoint::get_instance()->process(x + std::to_string(index)) +#define SYNC_POINT_CALLBACK(x, ...) doris::SyncPoint::get_instance()->process(x, {__VA_ARGS__}) +#define SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) \ +{ \ + std::pair ret {default_ret_val, false}; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&ret); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (ret.second) return std::move(ret.first); \ +} +#define SYNC_POINT_RETURN_WITH_VOID(x, ...) \ +{ \ + bool pred = false; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&pred); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (pred) return; \ +} +#define SYNC_POINT_SINGLETON() (void)doris::SyncPoint::get_instance() + +// TEST_SYNC_POINT is no op in release build. +// Turn on this feature by defining the macro +#ifndef BE_TEST +# define TEST_SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) +// seldom called +# define TEST_SYNC_POINT_SINGLETON() +#else +// Use TEST_SYNC_POINT to specify sync points inside code base. +// Sync points can have happens-after depedency on other sync points, +// configured at runtime via SyncPoint::load_dependency. This could be +// utilized to re-produce race conditions between threads. +# define TEST_SYNC_POINT(x) SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__) +# define TEST_SYNC_POINT_SINGLETON() SYNC_POINT_SINGLETON() + +/** + * Inject return points for testing. + * + * Currently we can only insert more points to get context from tested thread + * and process in testing thread, e.g. + * + * tested thread: + * ... + * TEST_SYNC_POINT_RETURN_WITH_VALUE("point_ret", int(0), ctx0); + * ... + * + * testing thread: + * sync_point->add("point_ret", [](auto&& args) { + * auto ctx0 = try_any_cast<bool>(args[0]); + * auto pair = try_any_cast<std::pair<int, bool>*>(args.back()); + * pair->first = ...; + * pair->second = ctx0; }); + * + * See sync_piont_test.cpp for more details. + */ +#pragma GCC diagnostic ignored "-Waddress" +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__) + +#endif // BE_TEST + +// TODO: define injection point in production env. +// the `if` expr can be live configure of the application +#ifndef ENABLE_INJECTION_POINT +# define TEST_INJECTION_POINT(x) +# define TEST_IDX_INJECTION_POINT(x, index) Review Comment: warning: macro is not used [clang-diagnostic-unused-macros] ```cpp # define TEST_IDX_INJECTION_POINT(x, index) ^ ``` ########## be/src/common/sync_point.h: ########## @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +#pragma once +// clang-format off +#include <functional> +#include <iostream> +#include <string> +#include <vector> +#include <any> + +namespace doris { + +#define SYNC_POINT_HOOK_RETURN_VALUE(expr, point_name, ...) \ + [&]() mutable { \ + TEST_SYNC_POINT_RETURN_WITH_VALUE(point_name, decltype((expr)) {}, __VA_ARGS__); \ + return (expr); \ + }() + +// This class provides facility to reproduce race conditions deterministically +// in unit tests. +// Developer could specify sync points in the codebase via TEST_SYNC_POINT. +// Each sync point represents a position in the execution stream of a thread. +// In the unit test, 'Happens After' relationship among sync points could be +// setup via SyncPoint::load_dependency, to reproduce a desired interleave of +// threads execution. +// Refer to (DBTest,TransactionLogIteratorRace), for an example use case. +class SyncPoint { +public: + static SyncPoint* get_instance(); + SyncPoint(const SyncPoint&) = delete; + SyncPoint& operator=(const SyncPoint&) = delete; + ~SyncPoint(); + struct SyncPointPair { + std::string predecessor; + std::string successor; + }; + + // call once at the beginning of a test to setup the dependency between + // sync points + // + // Example: + // load_dependency({{"point1", "point2"}, + // {"point2", "point3"}, + // {"point3", "point4"}}); + // + // test case thread thread for object being tested + // | | + // | | + // | \-------------0-------------\ | + // | \-> x sync point1 set in code + // | /----------1----------------/ | + // point2 o <-/ /-> x sync point4 set in code + // | / | + // z / | + // z /---------2-----------/ | there may be nothing + // | / | between point1 point4 + // ponit3 o --/ | they are for sync + // | | between test case and object + // v v + // + // vertical arrow means the procedure of each thread, the running order will + // be: + // test case thread -> point1 -> point2 -> point3 -> point4 -> object being + // tested + // + // we may do a lot of things between point2 and point3, say, change the + // object's status, call another method, propagate data race and etc. + void load_dependency(const std::vector<SyncPointPair>& dependencies); + + // call once at the beginning of a test to setup the dependency between + // sync points and setup markers indicating the successor is only enabled + // when it is processed on the same thread as the predecessor. + // When adding a marker, it implicitly adds a dependency for the marker pair. + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + + // The argument to the callback is passed through from + // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or + // TEST_IDX_SYNC_POINT was used. + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + + // Clear callback function by point + void clear_call_back(const std::string& point); + + // Clear all call back functions. + void clear_all_call_backs(); + + // Enable sync point processing (disabled on startup) + void enable_processing(); + + // Disable sync point processing + void disable_processing(); + + // Remove the execution trace of all sync points + void clear_trace(); + + // Triggered by TEST_SYNC_POINT, blocking execution until all predecessors + // are executed. + // And/or call registered callback function, with argument `cb_args` + void process(const std::string& point, std::vector<std::any>&& cb_args = {}); + + // TODO: it might be useful to provide a function that blocks until all + // sync points are cleared. + // We want this to be public so we can subclass the implementation + struct Data; + +private: + // Singleton + SyncPoint(); + Data* impl_; // impletation which is hidden in cpp file +}; + +template <class T> +T try_any_cast(const std::any& a) { + try { + return std::any_cast<T>(a); + } catch (const std::bad_any_cast& e) { + std::cerr << e.what() << " expected=" << typeid(T).name() << " actual=" << a.type().name() << std::endl; + throw e; + } +} + +template <typename T> +auto try_any_cast_ret(std::vector<std::any>& any) { + return try_any_cast<std::pair<T, bool>*>(any.back()); +} + +} // namespace doris + +#define SYNC_POINT(x) doris::SyncPoint::get_instance()->process(x) +#define IDX_SYNC_POINT(x, index) \ + doris::SyncPoint::get_instance()->process(x + std::to_string(index)) +#define SYNC_POINT_CALLBACK(x, ...) doris::SyncPoint::get_instance()->process(x, {__VA_ARGS__}) +#define SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) \ +{ \ + std::pair ret {default_ret_val, false}; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&ret); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (ret.second) return std::move(ret.first); \ +} +#define SYNC_POINT_RETURN_WITH_VOID(x, ...) \ +{ \ + bool pred = false; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&pred); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (pred) return; \ +} +#define SYNC_POINT_SINGLETON() (void)doris::SyncPoint::get_instance() + +// TEST_SYNC_POINT is no op in release build. +// Turn on this feature by defining the macro +#ifndef BE_TEST +# define TEST_SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) +// seldom called +# define TEST_SYNC_POINT_SINGLETON() +#else +// Use TEST_SYNC_POINT to specify sync points inside code base. +// Sync points can have happens-after depedency on other sync points, +// configured at runtime via SyncPoint::load_dependency. This could be +// utilized to re-produce race conditions between threads. +# define TEST_SYNC_POINT(x) SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__) +# define TEST_SYNC_POINT_SINGLETON() SYNC_POINT_SINGLETON() + +/** + * Inject return points for testing. + * + * Currently we can only insert more points to get context from tested thread + * and process in testing thread, e.g. + * + * tested thread: + * ... + * TEST_SYNC_POINT_RETURN_WITH_VALUE("point_ret", int(0), ctx0); + * ... + * + * testing thread: + * sync_point->add("point_ret", [](auto&& args) { + * auto ctx0 = try_any_cast<bool>(args[0]); + * auto pair = try_any_cast<std::pair<int, bool>*>(args.back()); + * pair->first = ...; + * pair->second = ctx0; }); + * + * See sync_piont_test.cpp for more details. + */ +#pragma GCC diagnostic ignored "-Waddress" +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__) + +#endif // BE_TEST + +// TODO: define injection point in production env. +// the `if` expr can be live configure of the application +#ifndef ENABLE_INJECTION_POINT +# define TEST_INJECTION_POINT(x) +# define TEST_IDX_INJECTION_POINT(x, index) +# define TEST_INJECTION_POINT_CALLBACK(x, ...) +# define TEST_INJECTION_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_INJECTION_POINT_RETURN_WITH_VOID(x, ...) Review Comment: warning: macro is not used [clang-diagnostic-unused-macros] ```cpp # define TEST_INJECTION_POINT_RETURN_WITH_VOID(x, ...) ^ ``` ########## be/src/common/sync_point.cpp: ########## @@ -0,0 +1,236 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +// clang-format off +#include "sync_point.h" + +#include <atomic> +#include <condition_variable> +#include <functional> +#include <mutex> +#include <random> +#include <string> +#include <thread> +#include <unordered_map> +#include <unordered_set> + +namespace doris { + +struct SyncPoint::Data { // impl +public: + Data() : enabled_(false) { } + virtual ~Data() {} + void process(const std::string& point, std::vector<std::any>&& cb_args); + void load_dependency(const std::vector<SyncPointPair>& dependencies); + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + bool predecessors_all_cleared(const std::string& point); + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + void clear_call_back(const std::string& point); + void clear_all_call_backs(); + void enable_processing(); + void disable_processing(); + void clear_trace(); +private: + bool disable_by_marker(const std::string& point, std::thread::id thread_id); +private: + // successor/predecessor map loaded from load_dependency + std::unordered_map<std::string, std::vector<std::string>> successors_; + std::unordered_map<std::string, std::vector<std::string>> predecessors_; + std::unordered_map<std::string, std::function<void(std::vector<std::any>&&)>> callbacks_; + std::unordered_map<std::string, std::vector<std::string>> markers_; + std::unordered_map<std::string, std::thread::id> marked_thread_id_; + std::mutex mutex_; + std::condition_variable cv_; + // sync points that have been passed through + std::unordered_set<std::string> cleared_points_; + std::atomic<bool> enabled_; + int num_callbacks_running_ = 0; +}; + +SyncPoint* SyncPoint::get_instance() { + static SyncPoint sync_point; + return &sync_point; +} +SyncPoint::SyncPoint() : + impl_(new Data) { +} +SyncPoint:: ~SyncPoint() { + delete impl_; +} +void SyncPoint::load_dependency(const std::vector<SyncPointPair>& dependencies) { + impl_->load_dependency(dependencies); +} +void SyncPoint::load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers) { + impl_->load_dependency_and_markers(dependencies, markers); +} +void SyncPoint::set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback) { + impl_->set_call_back(point, callback); +} +void SyncPoint::clear_call_back(const std::string& point) { + impl_->clear_call_back(point); +} +void SyncPoint::clear_all_call_backs() { + impl_->clear_all_call_backs(); +} +void SyncPoint::enable_processing() { + impl_->enable_processing(); +} +void SyncPoint::disable_processing() { + impl_->disable_processing(); +} +void SyncPoint::clear_trace() { + impl_->clear_trace(); +} +void SyncPoint::process(const std::string& point, std::vector<std::any>&& cb_arg) { + impl_->process(point, std::move(cb_arg)); +} + +// ============================================================================= +// SyncPoint implementation +// ============================================================================= + +void SyncPoint::Data::load_dependency( + const std::vector<SyncPointPair>& dependencies) { + std::lock_guard lock(mutex_); + successors_.clear(); + predecessors_.clear(); + cleared_points_.clear(); + for (const auto& dependency : dependencies) { + successors_[dependency.predecessor].push_back(dependency.successor); + predecessors_[dependency.successor].push_back(dependency.predecessor); + } + cv_.notify_all(); +} + +/** + * Markers are also dependency descriptions + */ +void SyncPoint::Data::load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers) { + std::lock_guard lock(mutex_); + successors_.clear(); + predecessors_.clear(); + cleared_points_.clear(); + markers_.clear(); + marked_thread_id_.clear(); + for (const auto& dependency : dependencies) { + successors_[dependency.predecessor].push_back(dependency.successor); + predecessors_[dependency.successor].push_back(dependency.predecessor); + } + for (const auto& marker : markers) { + successors_[marker.predecessor].push_back(marker.successor); + predecessors_[marker.successor].push_back(marker.predecessor); + markers_[marker.predecessor].push_back(marker.successor); + } + cv_.notify_all(); +} + +bool SyncPoint::Data::predecessors_all_cleared(const std::string& point) { + for (const auto& pred : predecessors_[point]) { Review Comment: warning: replace loop by 'std::ranges::all_of()' [readability-use-anyofallof] ```cpp for (const auto& pred : predecessors_[point]) { ^ ``` ########## be/src/common/sync_point.cpp: ########## @@ -0,0 +1,236 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +// clang-format off +#include "sync_point.h" + +#include <atomic> +#include <condition_variable> +#include <functional> +#include <mutex> +#include <random> +#include <string> +#include <thread> +#include <unordered_map> +#include <unordered_set> + +namespace doris { + +struct SyncPoint::Data { // impl +public: + Data() : enabled_(false) { } + virtual ~Data() {} + void process(const std::string& point, std::vector<std::any>&& cb_args); + void load_dependency(const std::vector<SyncPointPair>& dependencies); + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + bool predecessors_all_cleared(const std::string& point); + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + void clear_call_back(const std::string& point); + void clear_all_call_backs(); + void enable_processing(); + void disable_processing(); + void clear_trace(); +private: + bool disable_by_marker(const std::string& point, std::thread::id thread_id); +private: Review Comment: warning: redundant access specifier has the same accessibility as the previous access specifier [readability-redundant-access-specifiers] ```suggestion ``` <details> <summary>Additional context</summary> **be/src/common/sync_point.cpp:52:** previously declared here ```cpp private: ^ ``` </details> ########## be/src/common/sync_point.h: ########## @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +#pragma once +// clang-format off +#include <functional> +#include <iostream> +#include <string> +#include <vector> +#include <any> + +namespace doris { + +#define SYNC_POINT_HOOK_RETURN_VALUE(expr, point_name, ...) \ + [&]() mutable { \ + TEST_SYNC_POINT_RETURN_WITH_VALUE(point_name, decltype((expr)) {}, __VA_ARGS__); \ + return (expr); \ + }() + +// This class provides facility to reproduce race conditions deterministically +// in unit tests. +// Developer could specify sync points in the codebase via TEST_SYNC_POINT. +// Each sync point represents a position in the execution stream of a thread. +// In the unit test, 'Happens After' relationship among sync points could be +// setup via SyncPoint::load_dependency, to reproduce a desired interleave of +// threads execution. +// Refer to (DBTest,TransactionLogIteratorRace), for an example use case. +class SyncPoint { +public: + static SyncPoint* get_instance(); + SyncPoint(const SyncPoint&) = delete; + SyncPoint& operator=(const SyncPoint&) = delete; + ~SyncPoint(); + struct SyncPointPair { + std::string predecessor; + std::string successor; + }; + + // call once at the beginning of a test to setup the dependency between + // sync points + // + // Example: + // load_dependency({{"point1", "point2"}, + // {"point2", "point3"}, + // {"point3", "point4"}}); + // + // test case thread thread for object being tested + // | | + // | | + // | \-------------0-------------\ | + // | \-> x sync point1 set in code + // | /----------1----------------/ | + // point2 o <-/ /-> x sync point4 set in code + // | / | + // z / | + // z /---------2-----------/ | there may be nothing + // | / | between point1 point4 + // ponit3 o --/ | they are for sync + // | | between test case and object + // v v + // + // vertical arrow means the procedure of each thread, the running order will + // be: + // test case thread -> point1 -> point2 -> point3 -> point4 -> object being + // tested + // + // we may do a lot of things between point2 and point3, say, change the + // object's status, call another method, propagate data race and etc. + void load_dependency(const std::vector<SyncPointPair>& dependencies); + + // call once at the beginning of a test to setup the dependency between + // sync points and setup markers indicating the successor is only enabled + // when it is processed on the same thread as the predecessor. + // When adding a marker, it implicitly adds a dependency for the marker pair. + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + + // The argument to the callback is passed through from + // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or + // TEST_IDX_SYNC_POINT was used. + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + + // Clear callback function by point + void clear_call_back(const std::string& point); + + // Clear all call back functions. + void clear_all_call_backs(); + + // Enable sync point processing (disabled on startup) + void enable_processing(); + + // Disable sync point processing + void disable_processing(); + + // Remove the execution trace of all sync points + void clear_trace(); + + // Triggered by TEST_SYNC_POINT, blocking execution until all predecessors + // are executed. + // And/or call registered callback function, with argument `cb_args` + void process(const std::string& point, std::vector<std::any>&& cb_args = {}); + + // TODO: it might be useful to provide a function that blocks until all + // sync points are cleared. + // We want this to be public so we can subclass the implementation + struct Data; + +private: + // Singleton + SyncPoint(); + Data* impl_; // impletation which is hidden in cpp file +}; + +template <class T> +T try_any_cast(const std::any& a) { + try { + return std::any_cast<T>(a); + } catch (const std::bad_any_cast& e) { + std::cerr << e.what() << " expected=" << typeid(T).name() << " actual=" << a.type().name() << std::endl; + throw e; + } +} + +template <typename T> +auto try_any_cast_ret(std::vector<std::any>& any) { + return try_any_cast<std::pair<T, bool>*>(any.back()); +} + +} // namespace doris + +#define SYNC_POINT(x) doris::SyncPoint::get_instance()->process(x) +#define IDX_SYNC_POINT(x, index) \ + doris::SyncPoint::get_instance()->process(x + std::to_string(index)) +#define SYNC_POINT_CALLBACK(x, ...) doris::SyncPoint::get_instance()->process(x, {__VA_ARGS__}) +#define SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) \ +{ \ + std::pair ret {default_ret_val, false}; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&ret); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (ret.second) return std::move(ret.first); \ +} +#define SYNC_POINT_RETURN_WITH_VOID(x, ...) \ +{ \ + bool pred = false; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&pred); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (pred) return; \ +} +#define SYNC_POINT_SINGLETON() (void)doris::SyncPoint::get_instance() + +// TEST_SYNC_POINT is no op in release build. +// Turn on this feature by defining the macro +#ifndef BE_TEST +# define TEST_SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) +// seldom called +# define TEST_SYNC_POINT_SINGLETON() +#else +// Use TEST_SYNC_POINT to specify sync points inside code base. +// Sync points can have happens-after depedency on other sync points, +// configured at runtime via SyncPoint::load_dependency. This could be +// utilized to re-produce race conditions between threads. +# define TEST_SYNC_POINT(x) SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__) +# define TEST_SYNC_POINT_SINGLETON() SYNC_POINT_SINGLETON() + +/** + * Inject return points for testing. + * + * Currently we can only insert more points to get context from tested thread + * and process in testing thread, e.g. + * + * tested thread: + * ... + * TEST_SYNC_POINT_RETURN_WITH_VALUE("point_ret", int(0), ctx0); + * ... + * + * testing thread: + * sync_point->add("point_ret", [](auto&& args) { + * auto ctx0 = try_any_cast<bool>(args[0]); + * auto pair = try_any_cast<std::pair<int, bool>*>(args.back()); + * pair->first = ...; + * pair->second = ctx0; }); + * + * See sync_piont_test.cpp for more details. + */ +#pragma GCC diagnostic ignored "-Waddress" +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__) + +#endif // BE_TEST + +// TODO: define injection point in production env. +// the `if` expr can be live configure of the application +#ifndef ENABLE_INJECTION_POINT +# define TEST_INJECTION_POINT(x) +# define TEST_IDX_INJECTION_POINT(x, index) +# define TEST_INJECTION_POINT_CALLBACK(x, ...) Review Comment: warning: macro is not used [clang-diagnostic-unused-macros] ```cpp # define TEST_INJECTION_POINT_CALLBACK(x, ...) ^ ``` ########## be/src/common/sync_point.h: ########## @@ -0,0 +1,240 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Most code of this file is copied from rocksdb SyncPoint. +// https://github.com/facebook/rocksdb + +#pragma once +// clang-format off +#include <functional> +#include <iostream> +#include <string> +#include <vector> +#include <any> + +namespace doris { + +#define SYNC_POINT_HOOK_RETURN_VALUE(expr, point_name, ...) \ + [&]() mutable { \ + TEST_SYNC_POINT_RETURN_WITH_VALUE(point_name, decltype((expr)) {}, __VA_ARGS__); \ + return (expr); \ + }() + +// This class provides facility to reproduce race conditions deterministically +// in unit tests. +// Developer could specify sync points in the codebase via TEST_SYNC_POINT. +// Each sync point represents a position in the execution stream of a thread. +// In the unit test, 'Happens After' relationship among sync points could be +// setup via SyncPoint::load_dependency, to reproduce a desired interleave of +// threads execution. +// Refer to (DBTest,TransactionLogIteratorRace), for an example use case. +class SyncPoint { +public: + static SyncPoint* get_instance(); + SyncPoint(const SyncPoint&) = delete; + SyncPoint& operator=(const SyncPoint&) = delete; + ~SyncPoint(); + struct SyncPointPair { + std::string predecessor; + std::string successor; + }; + + // call once at the beginning of a test to setup the dependency between + // sync points + // + // Example: + // load_dependency({{"point1", "point2"}, + // {"point2", "point3"}, + // {"point3", "point4"}}); + // + // test case thread thread for object being tested + // | | + // | | + // | \-------------0-------------\ | + // | \-> x sync point1 set in code + // | /----------1----------------/ | + // point2 o <-/ /-> x sync point4 set in code + // | / | + // z / | + // z /---------2-----------/ | there may be nothing + // | / | between point1 point4 + // ponit3 o --/ | they are for sync + // | | between test case and object + // v v + // + // vertical arrow means the procedure of each thread, the running order will + // be: + // test case thread -> point1 -> point2 -> point3 -> point4 -> object being + // tested + // + // we may do a lot of things between point2 and point3, say, change the + // object's status, call another method, propagate data race and etc. + void load_dependency(const std::vector<SyncPointPair>& dependencies); + + // call once at the beginning of a test to setup the dependency between + // sync points and setup markers indicating the successor is only enabled + // when it is processed on the same thread as the predecessor. + // When adding a marker, it implicitly adds a dependency for the marker pair. + void load_dependency_and_markers( + const std::vector<SyncPointPair>& dependencies, + const std::vector<SyncPointPair>& markers); + + // The argument to the callback is passed through from + // TEST_SYNC_POINT_CALLBACK(); nullptr if TEST_SYNC_POINT or + // TEST_IDX_SYNC_POINT was used. + void set_call_back(const std::string& point, + const std::function<void(std::vector<std::any>&&)>& callback); + + // Clear callback function by point + void clear_call_back(const std::string& point); + + // Clear all call back functions. + void clear_all_call_backs(); + + // Enable sync point processing (disabled on startup) + void enable_processing(); + + // Disable sync point processing + void disable_processing(); + + // Remove the execution trace of all sync points + void clear_trace(); + + // Triggered by TEST_SYNC_POINT, blocking execution until all predecessors + // are executed. + // And/or call registered callback function, with argument `cb_args` + void process(const std::string& point, std::vector<std::any>&& cb_args = {}); + + // TODO: it might be useful to provide a function that blocks until all + // sync points are cleared. + // We want this to be public so we can subclass the implementation + struct Data; + +private: + // Singleton + SyncPoint(); + Data* impl_; // impletation which is hidden in cpp file +}; + +template <class T> +T try_any_cast(const std::any& a) { + try { + return std::any_cast<T>(a); + } catch (const std::bad_any_cast& e) { + std::cerr << e.what() << " expected=" << typeid(T).name() << " actual=" << a.type().name() << std::endl; + throw e; + } +} + +template <typename T> +auto try_any_cast_ret(std::vector<std::any>& any) { + return try_any_cast<std::pair<T, bool>*>(any.back()); +} + +} // namespace doris + +#define SYNC_POINT(x) doris::SyncPoint::get_instance()->process(x) +#define IDX_SYNC_POINT(x, index) \ + doris::SyncPoint::get_instance()->process(x + std::to_string(index)) +#define SYNC_POINT_CALLBACK(x, ...) doris::SyncPoint::get_instance()->process(x, {__VA_ARGS__}) +#define SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) \ +{ \ + std::pair ret {default_ret_val, false}; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&ret); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (ret.second) return std::move(ret.first); \ +} +#define SYNC_POINT_RETURN_WITH_VOID(x, ...) \ +{ \ + bool pred = false; \ + std::vector<std::any> args {__VA_ARGS__}; \ + args.push_back(&pred); \ + doris::SyncPoint::get_instance()->process(x, std::move(args)); \ + if (pred) return; \ +} +#define SYNC_POINT_SINGLETON() (void)doris::SyncPoint::get_instance() + +// TEST_SYNC_POINT is no op in release build. +// Turn on this feature by defining the macro +#ifndef BE_TEST +# define TEST_SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) +// seldom called +# define TEST_SYNC_POINT_SINGLETON() +#else +// Use TEST_SYNC_POINT to specify sync points inside code base. +// Sync points can have happens-after depedency on other sync points, +// configured at runtime via SyncPoint::load_dependency. This could be +// utilized to re-produce race conditions between threads. +# define TEST_SYNC_POINT(x) SYNC_POINT(x) +# define TEST_IDX_SYNC_POINT(x, index) IDX_SYNC_POINT(x, index) +# define TEST_SYNC_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__) +# define TEST_SYNC_POINT_SINGLETON() SYNC_POINT_SINGLETON() + +/** + * Inject return points for testing. + * + * Currently we can only insert more points to get context from tested thread + * and process in testing thread, e.g. + * + * tested thread: + * ... + * TEST_SYNC_POINT_RETURN_WITH_VALUE("point_ret", int(0), ctx0); + * ... + * + * testing thread: + * sync_point->add("point_ret", [](auto&& args) { + * auto ctx0 = try_any_cast<bool>(args[0]); + * auto pair = try_any_cast<std::pair<int, bool>*>(args.back()); + * pair->first = ...; + * pair->second = ctx0; }); + * + * See sync_piont_test.cpp for more details. + */ +#pragma GCC diagnostic ignored "-Waddress" +# define TEST_SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__) +# define TEST_SYNC_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__) + +#endif // BE_TEST + +// TODO: define injection point in production env. +// the `if` expr can be live configure of the application +#ifndef ENABLE_INJECTION_POINT +# define TEST_INJECTION_POINT(x) +# define TEST_IDX_INJECTION_POINT(x, index) +# define TEST_INJECTION_POINT_CALLBACK(x, ...) +# define TEST_INJECTION_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) +# define TEST_INJECTION_POINT_RETURN_WITH_VOID(x, ...) +# define TEST_INJECTION_POINT_SINGLETON() Review Comment: warning: macro is not used [clang-diagnostic-unused-macros] ```cpp # define TEST_INJECTION_POINT_SINGLETON() ^ ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
