diff options
| author | max42 <[email protected]> | 2023-07-29 00:02:16 +0300 | 
|---|---|---|
| committer | max42 <[email protected]> | 2023-07-29 00:02:16 +0300 | 
| commit | 73b89de71748a21e102d27b9f3ed1bf658766cb5 (patch) | |
| tree | 188bbd2d622fa91cdcbb1b6d6d77fbc84a0646f5 /library/cpp/threading/blocking_queue | |
| parent | 528e321bcc2a2b67b53aeba58c3bd88305a141ee (diff) | |
YT-19210: expose YQL shared library for YT.
After this, a new target libyqlplugin.so appears. in open-source cmake build.
Diff in open-source YDB repo looks like the following: https://paste.yandex-team.ru/f302bdb4-7ef2-4362-91c7-6ca45f329264
Diffstat (limited to 'library/cpp/threading/blocking_queue')
10 files changed, 475 insertions, 0 deletions
| diff --git a/library/cpp/threading/blocking_queue/CMakeLists.darwin-x86_64.txt b/library/cpp/threading/blocking_queue/CMakeLists.darwin-x86_64.txt new file mode 100644 index 00000000000..0fcf5ce481a --- /dev/null +++ b/library/cpp/threading/blocking_queue/CMakeLists.darwin-x86_64.txt @@ -0,0 +1,17 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-blocking_queue) +target_link_libraries(cpp-threading-blocking_queue PUBLIC +  contrib-libs-cxxsupp +  yutil +) +target_sources(cpp-threading-blocking_queue PRIVATE +  ${CMAKE_SOURCE_DIR}/library/cpp/threading/blocking_queue/blocking_queue.cpp +) diff --git a/library/cpp/threading/blocking_queue/CMakeLists.linux-aarch64.txt b/library/cpp/threading/blocking_queue/CMakeLists.linux-aarch64.txt new file mode 100644 index 00000000000..640f1e91440 --- /dev/null +++ b/library/cpp/threading/blocking_queue/CMakeLists.linux-aarch64.txt @@ -0,0 +1,18 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-blocking_queue) +target_link_libraries(cpp-threading-blocking_queue PUBLIC +  contrib-libs-linux-headers +  contrib-libs-cxxsupp +  yutil +) +target_sources(cpp-threading-blocking_queue PRIVATE +  ${CMAKE_SOURCE_DIR}/library/cpp/threading/blocking_queue/blocking_queue.cpp +) diff --git a/library/cpp/threading/blocking_queue/CMakeLists.linux-x86_64.txt b/library/cpp/threading/blocking_queue/CMakeLists.linux-x86_64.txt new file mode 100644 index 00000000000..640f1e91440 --- /dev/null +++ b/library/cpp/threading/blocking_queue/CMakeLists.linux-x86_64.txt @@ -0,0 +1,18 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-blocking_queue) +target_link_libraries(cpp-threading-blocking_queue PUBLIC +  contrib-libs-linux-headers +  contrib-libs-cxxsupp +  yutil +) +target_sources(cpp-threading-blocking_queue PRIVATE +  ${CMAKE_SOURCE_DIR}/library/cpp/threading/blocking_queue/blocking_queue.cpp +) diff --git a/library/cpp/threading/blocking_queue/CMakeLists.txt b/library/cpp/threading/blocking_queue/CMakeLists.txt new file mode 100644 index 00000000000..f8b31df0c11 --- /dev/null +++ b/library/cpp/threading/blocking_queue/CMakeLists.txt @@ -0,0 +1,17 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + +if (CMAKE_SYSTEM_NAME STREQUAL "Linux" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "aarch64" AND NOT HAVE_CUDA) +  include(CMakeLists.linux-aarch64.txt) +elseif (CMAKE_SYSTEM_NAME STREQUAL "Darwin" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "x86_64") +  include(CMakeLists.darwin-x86_64.txt) +elseif (WIN32 AND CMAKE_SYSTEM_PROCESSOR STREQUAL "AMD64" AND NOT HAVE_CUDA) +  include(CMakeLists.windows-x86_64.txt) +elseif (CMAKE_SYSTEM_NAME STREQUAL "Linux" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "x86_64" AND NOT HAVE_CUDA) +  include(CMakeLists.linux-x86_64.txt) +endif() diff --git a/library/cpp/threading/blocking_queue/CMakeLists.windows-x86_64.txt b/library/cpp/threading/blocking_queue/CMakeLists.windows-x86_64.txt new file mode 100644 index 00000000000..0fcf5ce481a --- /dev/null +++ b/library/cpp/threading/blocking_queue/CMakeLists.windows-x86_64.txt @@ -0,0 +1,17 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-blocking_queue) +target_link_libraries(cpp-threading-blocking_queue PUBLIC +  contrib-libs-cxxsupp +  yutil +) +target_sources(cpp-threading-blocking_queue PRIVATE +  ${CMAKE_SOURCE_DIR}/library/cpp/threading/blocking_queue/blocking_queue.cpp +) diff --git a/library/cpp/threading/blocking_queue/blocking_queue.cpp b/library/cpp/threading/blocking_queue/blocking_queue.cpp new file mode 100644 index 00000000000..db199c80be5 --- /dev/null +++ b/library/cpp/threading/blocking_queue/blocking_queue.cpp @@ -0,0 +1,3 @@ +#include "blocking_queue.h" + +// just check compilability diff --git a/library/cpp/threading/blocking_queue/blocking_queue.h b/library/cpp/threading/blocking_queue/blocking_queue.h new file mode 100644 index 00000000000..48d3762f68a --- /dev/null +++ b/library/cpp/threading/blocking_queue/blocking_queue.h @@ -0,0 +1,158 @@ +#pragma once + +#include <util/generic/deque.h> +#include <util/generic/maybe.h> +#include <util/generic/yexception.h> +#include <util/system/condvar.h> +#include <util/system/guard.h> +#include <util/system/mutex.h> + +#include <utility> + +namespace NThreading { +    /// +    /// TBlockingQueue is a queue of elements of limited or unlimited size. +    /// Queue provides Push and Pop operations that block if operation can't be executed +    /// (queue is empty or maximum size is reached). +    /// +    /// Queue can be stopped, in that case all blocked operation will return `Nothing` / false. +    /// +    /// All operations are thread safe. +    /// +    /// +    /// Example of usage: +    ///     TBlockingQueue<int> queue; +    /// +    ///     ... +    /// +    ///     // thread 1 +    ///     queue.Push(42); +    ///     queue.Push(100500); +    /// +    ///     ... +    /// +    ///     // thread 2 +    ///     while (TMaybe<int> number = queue.Pop()) { +    ///         ProcessNumber(number.GetRef()); +    ///     } +    template <class TElement> +    class TBlockingQueue { +    public: +        /// +        /// Creates blocking queue with given maxSize +        /// if maxSize == 0 then queue is unlimited +        TBlockingQueue(size_t maxSize) +            : MaxSize(maxSize == 0 ? Max<size_t>() : maxSize) +            , Stopped(false) +        { +        } + +        /// +        /// Blocks until queue has some elements or queue is stopped or deadline is reached. +        /// Returns `Nothing` if queue is stopped or deadline is reached. +        /// Returns element otherwise. +        TMaybe<TElement> Pop(TInstant deadline = TInstant::Max()) { +            TGuard<TMutex> g(Lock); + +            const auto canPop = [this]() { return CanPop(); }; +            if (!CanPopCV.WaitD(Lock, deadline, canPop)) { +                return Nothing(); +            } + +            if (Stopped && Queue.empty()) { +                return Nothing(); +            } +            TElement e = std::move(Queue.front()); +            Queue.pop_front(); +            CanPushCV.Signal(); +            return std::move(e); +        } + +        TMaybe<TElement> Pop(TDuration duration) { +            return Pop(TInstant::Now() + duration); +        } + +        /// +        /// Blocks until queue has space for new elements or queue is stopped or deadline is reached. +        /// Returns false exception if queue is stopped and push failed or deadline is reached. +        /// Pushes element to queue and returns true otherwise. +        bool Push(const TElement& e, TInstant deadline = TInstant::Max()) { +            return PushRef(e, deadline); +        } + +        bool Push(TElement&& e, TInstant deadline = TInstant::Max()) { +            return PushRef(std::move(e), deadline); +        } + +        bool Push(const TElement& e, TDuration duration) { +            return Push(e, TInstant::Now() + duration); +        } + +        bool Push(TElement&& e, TDuration duration) { +            return Push(std::move(e), TInstant::Now() + duration); +        } + +        /// +        /// Stops the queue, all blocked operations will be aborted. +        void Stop() { +            TGuard<TMutex> g(Lock); +            Stopped = true; +            CanPopCV.BroadCast(); +            CanPushCV.BroadCast(); +        } + +        /// +        /// Checks whether queue is empty. +        bool Empty() const { +            TGuard<TMutex> g(Lock); +            return Queue.empty(); +        } + +        /// +        /// Returns size of the queue. +        size_t Size() const { +            TGuard<TMutex> g(Lock); +            return Queue.size(); +        } + +        /// +        /// Checks whether queue is stopped. +        bool IsStopped() const { +            TGuard<TMutex> g(Lock); +            return Stopped; +        } + +    private: +        bool CanPush() const { +            return Queue.size() < MaxSize || Stopped; +        } + +        bool CanPop() const { +            return !Queue.empty() || Stopped; +        } + +        template <typename Ref> +        bool PushRef(Ref e, TInstant deadline) { +            TGuard<TMutex> g(Lock); +            const auto canPush = [this]() { return CanPush(); }; +            if (!CanPushCV.WaitD(Lock, deadline, canPush)) { +                return false; +            } +            if (Stopped) { +                return false; +            } +            Queue.push_back(std::forward<TElement>(e)); +            CanPopCV.Signal(); +            return true; +        } + +    private: +        TMutex Lock; +        TCondVar CanPopCV; +        TCondVar CanPushCV; +        TDeque<TElement> Queue; +        size_t MaxSize; +        bool Stopped; +    }; + +} diff --git a/library/cpp/threading/blocking_queue/blocking_queue_ut.cpp b/library/cpp/threading/blocking_queue/blocking_queue_ut.cpp new file mode 100644 index 00000000000..26e125279d2 --- /dev/null +++ b/library/cpp/threading/blocking_queue/blocking_queue_ut.cpp @@ -0,0 +1,211 @@ +#include "blocking_queue.h" + +#include <library/cpp/testing/unittest/registar.h> + +#include <util/string/builder.h> +#include <util/system/thread.h> + +namespace { +    class TFunctionThread: public ISimpleThread { +    public: +        using TFunc = std::function<void()>; + +    private: +        TFunc Func; + +    public: +        TFunctionThread(const TFunc& func) +            : Func(func) +        { +        } + +        void* ThreadProc() noexcept override { +            Func(); +            return nullptr; +        } +    }; + +} + +IOutputStream& operator<<(IOutputStream& o, const TMaybe<int>& val) { +    if (val) { +        o << "TMaybe<int>(" << val.GetRef() << ')'; +    } else { +        o << "TMaybe<int>()"; +    } +    return o; +} + +Y_UNIT_TEST_SUITE(BlockingQueueTest) { +    Y_UNIT_TEST(SimplePushPopTest) { +        const size_t limit = 100; + +        NThreading::TBlockingQueue<int> queue(100); + +        for (int i = 0; i != limit; ++i) { +            queue.Push(i); +        } + +        for (int i = 0; i != limit; ++i) { +            UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), i); +        } + +        UNIT_ASSERT(queue.Empty()); +    } + +    Y_UNIT_TEST(SimpleStopTest) { +        const size_t limit = 100; + +        NThreading::TBlockingQueue<int> queue(100); + +        for (int i = 0; i != limit; ++i) { +            queue.Push(i); +        } +        queue.Stop(); + +        bool ok = queue.Push(100500); +        UNIT_ASSERT_VALUES_EQUAL(ok, false); + +        for (int i = 0; i != limit; ++i) { +            UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), i); +        } + +        UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), TMaybe<int>()); +    } + +    Y_UNIT_TEST(BigPushPop) { +        const int limit = 100000; + +        NThreading::TBlockingQueue<int> queue(10); + +        TFunctionThread pusher([&] { +            for (int i = 0; i != limit; ++i) { +                if (!queue.Push(i)) { +                    break; +                } +            } +        }); + +        pusher.Start(); + +        try { +            for (int i = 0; i != limit; ++i) { +                size_t size = queue.Size(); +                UNIT_ASSERT_C(size <= 10, (TStringBuilder() << "Size exceeds 10: " << size).data()); +                UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), i); +            } +        } catch (...) { +            // gracefull shutdown of pusher thread if assertion fails +            queue.Stop(); +            throw; +        } + +        pusher.Join(); +    } + +    Y_UNIT_TEST(StopWhenMultiplePoppers) { +        NThreading::TBlockingQueue<int> queue(10); +        TFunctionThread popper1([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), TMaybe<int>()); +        }); +        TFunctionThread popper2([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue.Pop(), TMaybe<int>()); +        }); +        popper1.Start(); +        popper2.Start(); + +        queue.Stop(); + +        popper1.Join(); +        popper2.Join(); +    } + +    Y_UNIT_TEST(StopWhenMultiplePushers) { +        NThreading::TBlockingQueue<int> queue(1); +        queue.Push(1); +        TFunctionThread pusher1([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue.Push(2), false); +        }); +        TFunctionThread pusher2([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue.Push(2), false); +        }); +        pusher1.Start(); +        pusher2.Start(); + +        queue.Stop(); + +        pusher1.Join(); +        pusher2.Join(); +    } + +    Y_UNIT_TEST(InterruptPopByDeadline) { +        NThreading::TBlockingQueue<int> queue1(10); +        NThreading::TBlockingQueue<int> queue2(10); + +        const auto popper1DeadLine = TInstant::Now(); +        const auto popper2DeadLine = TInstant::Now() + TDuration::Seconds(2); + +        TFunctionThread popper1([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue1.Pop(popper1DeadLine), TMaybe<int>()); +            UNIT_ASSERT_VALUES_EQUAL(queue1.IsStopped(), false); +        }); + +        TFunctionThread popper2([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue2.Pop(popper2DeadLine), 2); +            UNIT_ASSERT_VALUES_EQUAL(queue2.IsStopped(), false); +        }); + +        popper1.Start(); +        popper2.Start(); + +        Sleep(TDuration::Seconds(1)); + +        queue1.Push(1); +        queue2.Push(2); + +        Sleep(TDuration::Seconds(1)); + +        queue1.Stop(); +        queue2.Stop(); + +        popper1.Join(); +        popper2.Join(); +    } + +    Y_UNIT_TEST(InterruptPushByDeadline) { +        NThreading::TBlockingQueue<int> queue1(1); +        NThreading::TBlockingQueue<int> queue2(1); + +        queue1.Push(0); +        queue2.Push(0); + +        const auto pusher1DeadLine = TInstant::Now(); +        const auto pusher2DeadLine = TInstant::Now() + TDuration::Seconds(2); + +        TFunctionThread pusher1([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue1.Push(1, pusher1DeadLine), false); +            UNIT_ASSERT_VALUES_EQUAL(queue1.IsStopped(), false); +        }); + +        TFunctionThread pusher2([&] { +            UNIT_ASSERT_VALUES_EQUAL(queue2.Push(2, pusher2DeadLine), true); +            UNIT_ASSERT_VALUES_EQUAL(queue2.IsStopped(), false); +        }); + +        pusher1.Start(); +        pusher2.Start(); + +        Sleep(TDuration::Seconds(1)); + +        queue1.Pop(); +        queue2.Pop(); + +        Sleep(TDuration::Seconds(1)); + +        queue1.Stop(); +        queue2.Stop(); + +        pusher1.Join(); +        pusher2.Join(); +    } +} diff --git a/library/cpp/threading/blocking_queue/ut/ya.make b/library/cpp/threading/blocking_queue/ut/ya.make new file mode 100644 index 00000000000..50f220d5526 --- /dev/null +++ b/library/cpp/threading/blocking_queue/ut/ya.make @@ -0,0 +1,7 @@ +UNITTEST_FOR(library/cpp/threading/blocking_queue) + +SRCS( +    blocking_queue_ut.cpp +) + +END() diff --git a/library/cpp/threading/blocking_queue/ya.make b/library/cpp/threading/blocking_queue/ya.make new file mode 100644 index 00000000000..ce1104c1c99 --- /dev/null +++ b/library/cpp/threading/blocking_queue/ya.make @@ -0,0 +1,9 @@ +LIBRARY() + +SRCS( +    blocking_queue.cpp +) + +END() + +RECURSE_FOR_TESTS(ut) | 
