// 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);

  class CallbackGuard {

  public:

    CallbackGuard() = default;

    explicit CallbackGuard(std::string point) : _point(std::move(point)) {}

    ~CallbackGuard() {

      if (!_point.empty()) {

        get_instance()->clear_call_back(_point);

      }

    }

    CallbackGuard(const CallbackGuard&) = delete;

    CallbackGuard& operator=(const CallbackGuard&) = delete;

    CallbackGuard(CallbackGuard&& other) noexcept {

      if (!_point.empty() && _point != other._point) {

        get_instance()->clear_call_back(_point);

      }

      _point = std::move(other._point);

    }

    CallbackGuard& operator=(CallbackGuard&& other) noexcept {

      if (!_point.empty() && _point != other._point) {

        get_instance()->clear_call_back(_point);

      }

      _point = std::move(other._point);

      return *this;

    };

  private:

    std::string _point;

  };

  // 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.

  // If `guard` is not nullptr, method will return a `CallbackGuard` object which will clear the

  // callback when it is destructed.

  void set_call_back(const std::string& point,

                     const std::function<void(std::vector<std::any>&&)>& callback,

                     CallbackGuard* guard = nullptr);

  // 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 = {});

  // Check if this point is registered

  bool has_point(const std::string& point);

  // Get this point if enabled

  bool get_enable();

  // 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, ...) \

{ \

  auto sync_point = doris::SyncPoint::get_instance(); \

  if (sync_point->get_enable() && sync_point->has_point(x)) { \

    std::pair ret {default_ret_val, false}; \

    std::vector<std::any> args {__VA_ARGS__}; \

    args.emplace_back(&ret); \

    sync_point->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.emplace_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

#if !defined(BE_TEST) && !defined(ENABLE_INJECTION_POINT)

# 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()

#else

# define TEST_INJECTION_POINT(x) SYNC_POINT(x);

# define TEST_IDX_INJECTION_POINT(x, index) IDX_SYNC_POINT(x, index);

# define TEST_INJECTION_POINT_CALLBACK(x, ...) SYNC_POINT_CALLBACK(x, __VA_ARGS__);

# define TEST_INJECTION_POINT_SINGLETON() SYNC_POINT_SINGLETON();

# define TEST_INJECTION_POINT_RETURN_WITH_VALUE(x, default_ret_val, ...) SYNC_POINT_RETURN_WITH_VALUE(x, default_ret_val, __VA_ARGS__);

# define TEST_INJECTION_POINT_RETURN_WITH_VOID(x, ...) SYNC_POINT_RETURN_WITH_VOID(x, __VA_ARGS__);

#endif // ENABLE_INJECTION_POINT

// clang-format on

// vim: et tw=80 ts=2 sw=2 cc=80: