cloud/test/sync_point_test.cpp

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

// switch on syncpoint macro
#ifndef UNIT_TEST
#define UNIT_TEST
#endif
#include "cpp/sync_point.h"

#include <cassert>
#include <condition_variable>
#include <iostream>
#include <thread>

#define STDOUT (std::cout << __PRETTY_FUNCTION__ << " ")

int g_data = 12315;

void foo() {
    int a = 10086;

    STDOUT << "a: " << a << std::endl;

    TEST_SYNC_POINT_CALLBACK("foo:a_assigned", &a); // a may be accessed

    STDOUT << "a: " << a << std::endl;

    int b = 100;

    STDOUT << "b: " << b << std::endl;
    TEST_SYNC_POINT_CALLBACK("foo:b_assigned", &b); // b may be accessed

    STDOUT << "b: " << b << std::endl;

    a += b;

    STDOUT << "a: " << a << std::endl;

    if (a == 10086 + 100) {
        STDOUT << "expected branch taken, a: " << a << std::endl;
    } else {
        STDOUT << "exceptional branch taken, a: " << a << std::endl;
    }
}

void test_foo_single_thread() {
    auto sp = doris::SyncPoint::get_instance();
    std::cout << "========== default behavior ==========" << std::endl;
    foo(); // first run, nothing changed

    std::cout << "========== change default behavior ==========" << std::endl;
    sp->set_call_back("foo:a_assigned", [](void* arg) {
        int& a = *reinterpret_cast<int*>(arg); // we known what arg is
        STDOUT << "original a: " << a << std::endl;
        a = 10010;
        STDOUT << "change a: " << a << std::endl;
    });

    sp->set_call_back("foo:b_assigned", [](void* arg) {
        int& b = *reinterpret_cast<int*>(arg); // we known what arg is
        STDOUT << "original b: " << b << std::endl;
        b = 200;
        STDOUT << "change b: " << b << std::endl;
    });

    sp->enable_processing(); // turn on
    foo();

    sp->clear_all_call_backs();
    sp->disable_processing();
    sp->clear_trace();
}

void foo(int x) {
    // multi-thread data race
    g_data = x;

    TEST_SYNC_POINT("foo:assigned:1");
    // concurrent stuff may be executed by other thread here,
    // and this thread may race with it
    TEST_SYNC_POINT("foo:assigned:2");

    STDOUT << x << std::endl;
}

void bar() {
    TEST_SYNC_POINT("bar:assigned:1");
    g_data = 10086;
    TEST_SYNC_POINT("bar:assigned:2");
}

void test_foo_data_race() {
    std::cout << "========== data race ==========" << std::endl;

    std::mutex mtx;
    std::condition_variable cv;
    auto sp = doris::SyncPoint::get_instance();

    bool go = false;
    std::unique_lock<std::mutex> lk(mtx);
    g_data = 0;

    // ===========================================================================
    // FORCE threads concurrent execution with sequence:
    //
    //      thread1       thread2       thread3
    //         |             |             |
    //         |             |           foo(3)
    //         |             |             |
    //         |           foo(2)          |
    //         |             |             |
    //       foo(1)          |             |
    //         |             |             |
    //         v             v             v
    //
    std::thread th1([&] {
        {
            std::unique_lock<std::mutex> lk(mtx);
            cv.wait(lk, [&] { return go; });
        }
        TEST_SYNC_POINT("test_foo_data_race:1:start");
        assert(g_data == 2);
        foo(1);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        assert(g_data == 1);
        TEST_SYNC_POINT("test_foo_data_race:1:end");
    });
    std::thread th2([&] {
        {
            std::unique_lock<std::mutex> lk(mtx);
            cv.wait(lk, [&] { return go; });
        }
        TEST_SYNC_POINT("test_foo_data_race:2:start");
        assert(g_data == 3);
        foo(2);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        assert(g_data == 2);
        TEST_SYNC_POINT("test_foo_data_race:2:end");
    });
    std::thread th3([&] {
        {
            std::unique_lock<std::mutex> lk(mtx);
            cv.wait(lk, [&] { return go; });
        }
        TEST_SYNC_POINT("test_foo_data_race:3:start");
        assert(g_data == 0);
        foo(3);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        assert(g_data == 3);
        TEST_SYNC_POINT("test_foo_data_race:3:end");
    });

    // prepare dependency
    sp->enable_processing();
    // run foo(int) in sequence 3->2->1
    sp->load_dependency({
            {"test_foo_data_race:3:end", "test_foo_data_race:2:start"},
            {"test_foo_data_race:2:end", "test_foo_data_race:1:start"},
    });

    // set and go
    go = true;
    lk.unlock();
    cv.notify_all();

    th1.join();
    th2.join();
    th3.join();
    sp->clear_all_call_backs();
    sp->clear_trace();
    sp->disable_processing();

    // ===========================================================================
    // FORCE to run bar() in the middle of foo()
    //
    //   thread4            thread5
    //      |                  |
    //      |             foo():assigned:1
    //      |                  |
    //     bar()               |
    //      |                  |
    //      |             foo():assigned:2
    //      |                  |
    //      v                  v
    //
    lk.lock();
    go = false;

    std::thread th4([&] {
        {
            std::unique_lock<std::mutex> lk(mtx);
            cv.wait(lk, [&] { return go; });
        }
        bar();
    });
    std::thread th5([&] {
        {
            std::unique_lock<std::mutex> lk(mtx);
            cv.wait(lk, [&] { return go; });
        }
        foo(10010);              // try to set g_data to 10010
        assert(g_data == 10086); // foo() is racing with bar()
    });

    // prepare dependency
    sp->enable_processing();
    sp->load_dependency({
            {"foo:assigned:1", "bar:assigned:1"}, // no need to specify bar1->bar2,
            {"bar:assigned:2", "foo:assigned:2"}, // because they a natually sequenced
    });

    // set and go
    go = true;
    lk.unlock();
    cv.notify_all();

    th4.join();
    th5.join();
    sp->clear_all_call_backs();
    sp->clear_trace();

    sp->disable_processing();
}

int foo_return_with_value() {
    int ctx = 1;
    (void)ctx;
    // for those return values are not explictly declared
    {
        // for ctx capture
        TEST_SYNC_POINT_CALLBACK("foo_return_with_value1_ctx", &ctx);

        int tmp_ret = -1;
        (void)tmp_ret; // supress `unused` warning when build in release mode
        TEST_SYNC_POINT_RETURN_WITH_VALUE("foo_return_with_value1", &tmp_ret);
    }

    // for those return valuse are explicitly declared
    {
        int ret = -1;
        TEST_SYNC_POINT_RETURN_WITH_VALUE("foo_return_with_value2", &ret);
        return ret;
    }
}

void foo_return_with_void(int* in) {
    *in = 10000;
    TEST_SYNC_POINT_RETURN_WITH_VOID("foo_return_with_void1");
    *in = 10010;
    TEST_SYNC_POINT_RETURN_WITH_VOID("foo_return_with_void2");
    *in = 10086;
}

void test_return_point() {
    auto sp = doris::SyncPoint::get_instance();
    sp->clear_all_call_backs();
    sp->enable_processing();

    // test pred == false, nothing happens
    {
        sp->clear_all_call_backs();

        sp->set_call_back("foo_return_with_value1",
                          [](void* ret) { *reinterpret_cast<int*>(ret) = -1; });
        sp->set_call_back("foo_return_with_value1::pred",
                          [](void* pred) { *reinterpret_cast<bool*>(pred) = false; });

        sp->set_call_back("foo_return_with_void1::pred",
                          [](void* pred) { *reinterpret_cast<bool*>(pred) = false; });

        int ret = foo_return_with_value();
        assert(ret == -1);

        foo_return_with_void(&ret);
        assert(ret == 10086);
    }

    // test pred == true, get the value and return point we want
    {
        sp->clear_all_call_backs();
        sp->set_call_back("foo_return_with_value2",
                          [](void* ret) { *reinterpret_cast<int*>(ret) = 10086; });
        sp->set_call_back("foo_return_with_value2::pred",
                          [](void* pred) { *reinterpret_cast<bool*>(pred) = true; });

        sp->set_call_back("foo_return_with_void2::pred",
                          [](void* pred) { *reinterpret_cast<bool*>(pred) = true; });

        int ret = foo_return_with_value();
        assert(ret == 10086);

        foo_return_with_void(&ret);
        assert(ret == 10010);
    }

    // pred depends on tested thread's context
    {
        sp->clear_all_call_backs();
        int* ctx;
        // "steal" context from tested thread to this testing thread in order to
        // change behaviors of the tested thread without changing it's code
        sp->set_call_back("foo_return_with_value1_ctx",
                          [&ctx](void* tested_ctx) { ctx = reinterpret_cast<int*>(tested_ctx); });
        sp->set_call_back("foo_return_with_value1",
                          [](void* ret) { *reinterpret_cast<int*>(ret) = 10086; });
        // use the context from tested thread to do more checks and modifications
        sp->set_call_back("foo_return_with_value1::pred", [&ctx](void* pred) {
            // we can change the return logic of the tested thread
            *reinterpret_cast<bool*>(pred) = (*ctx > 0); // true
        });

        [[maybe_unused]] int ret = foo_return_with_value();
        assert(ret == 10086);
    }
}

void test() {
    //   test_foo_single_thread();
    //   test_foo_data_race();
    test_return_point();
}

int main() {
    test();
    return 0;
}