diff --git a/libs/main/core/piobject_macros.h b/libs/main/core/piobject_macros.h index 7a238254..1622346c 100644 --- a/libs/main/core/piobject_macros.h +++ b/libs/main/core/piobject_macros.h @@ -50,7 +50,7 @@ //! \~english Put this macro inside a %PIObject subclass definition to inherit registered methods and class scope from "parent". //! \~russian Поместите этот макрос внутрь объявления наследника %PIObject, чтобы унаследовать зарегистрированные методы и цепочку //! классов от "parent". -# define PIOBJECT_SUBCLASS(name, parent) +# define PIOBJECT_SUBCLASS(name, ...) //! \relatesalso PIObject @@ -456,11 +456,11 @@ }; \ __BaseInitializer__ __base_init__; -# define PIOBJECT_PARENT(name) \ +# define PIOBJECT_PARENT(...) \ class __ParentInitializer__ { \ public: \ __ParentInitializer__() { \ - uint pid = name::__classNameIDS(); \ + uint pid = __VA_ARGS__::__classNameIDS(); \ if (pid == 0) return; \ uint id = __classNameIDS(); \ PIMutexLocker ml(__meta_mutex()); \ @@ -478,13 +478,13 @@ \ public: \ const char * parentClassName() const override { \ - return #name; \ + return #__VA_ARGS__; \ } \ - typedef name __Parent__; \ + typedef __VA_ARGS__ __Parent__; \ \ private: -# define PIOBJECT_SUBCLASS(name, parent) PIOBJECT(name) PIOBJECT_PARENT(parent) +# define PIOBJECT_SUBCLASS(name, ...) PIOBJECT(name) PIOBJECT_PARENT(__VA_ARGS__) # define __EH_INIT_BASE__(ret, name) \ diff --git a/libs/main/thread/pigrabberbase.h b/libs/main/thread/pigrabberbase.h index 0e78cbdf..5c54e977 100644 --- a/libs/main/thread/pigrabberbase.h +++ b/libs/main/thread/pigrabberbase.h @@ -5,7 +5,7 @@ //! \~russian Базовый класс потока-граббера /* PIP - Platform Independent Primitives - Grabber thread base class + Grabber thread base class Andrey Bychkov work.a.b@yandex.ru This program is free software: you can redistribute it and/or modify @@ -25,9 +25,9 @@ #ifndef PIGRABBERBASE_H #define PIGRABBERBASE_H +#include "piconditionvar.h" #include "pidiagnostics.h" #include "pithread.h" -#include "pitime.h" //! \~\ingroup Thread //! \~\brief @@ -48,6 +48,7 @@ public: PIGrabberBase() { is_opened = false; is_recording = false; + poll_delay_ = PISystemTime::fromMilliseconds(PIP_MIN_MSLEEP); } //! \~english Stops the grabber thread and releases recording/open state. @@ -139,6 +140,7 @@ public: void stopGrabber(bool wait_forever = true) { if (isRunning()) { stop(); + cv.notifyAll(); if (wait_forever) waitForFinish(); else { @@ -210,6 +212,51 @@ public: EVENT(closed); //! \} + //! \handlers + //! \{ + + //! \~\fn void wakeUp() + //! \~\brief + //! \~english Wakes the grabber loop from its wait between polls. + //! \~russian Пробуждает цикл граббера из ожидания между опросами. + //! \~\details + //! \~english + //! Call this (directly or via event connection) from the device layer when new data may be available. + //! \~It is safe to call from any thread. The next \a get() poll happens immediately instead of after the + //! \a pollDelay() interval. + //! \~russian + //! Вызывайте это (напрямую или через подключение события) из устройства, когда могут появиться новые данные. + //! \~Потокбезопасно из любого потока. Следующий опрос \a get() произойдет немедленно, а не через интервал + //! \a pollDelay(). + EVENT_HANDLER0(void, wakeUp) { cv.notifyAll(); } + + //! \} + //! \~english Returns the fallback interval between polls when no wake-up was requested. + //! \~russian Возвращает резервный интервал между опросами, если не было запрошено пробуждения. + PISystemTime pollDelay() const { + PIMutexLocker locker(wait_mutex); + return poll_delay_; + } + + //! \~english Sets the fallback interval between polls when no wake-up was requested. + //! \~russian Устанавливает резервный интервал между опросами, если не было запрошено пробуждения. + //! \~\details + //! \~english + //! The grabber loop waits on a condition variable for up to this interval between \a get() polls and wakes + //! immediately on \a wakeUp() or stop. A larger value saves CPU when data is rare; a smaller value reduces + //! worst-case capture latency. Default is \a PIP_MIN_MSLEEP milliseconds. + //! \~russian + //! Цикл граббера ожидает на переменной условия не дольше этого интервала между опросами \a get() и + //! пробуждается немедленно по \a wakeUp() или при остановке. Большее значение экономит CPU при редких + //! данных; меньшее уменьшает худшую задержку захвата. По умолчанию \a PIP_MIN_MSLEEP миллисекунд. + //! \~\note + //! \~english If the grabber is currently waiting, it is woken up to apply the new interval. + //! \~russian Если граббер сейчас ожидает, он пробуждается для применения нового интервала. + void setPollDelay(PISystemTime d) { + PIMutexLocker locker(wait_mutex); + poll_delay_ = d; + cv.notifyAll(); + } protected: //! \~english Virtual method executed once when the thread starts, before polling begins. @@ -251,7 +298,12 @@ private: if (!isOpened()) { open(); diag_.reset(); - if (!is_opened) piMSleep(200); + if (!is_opened) { + wait_mutex.lock(); + cv.waitFor(wait_mutex, PISystemTime::fromMilliseconds(200)); + wait_mutex.unlock(); + return; + } } if (isOpened()) { T c; @@ -261,7 +313,9 @@ private: return; } if (ret > 0) { - piMinSleep(); + wait_mutex.lock(); + cv.waitFor(wait_mutex, poll_delay_); + wait_mutex.unlock(); return; } diag_.received(1); @@ -289,7 +343,9 @@ private: T last_; PIQueue que; PIDiagnostics diag_; - mutable PIMutex que_mutex, last_mutex; + PISystemTime poll_delay_; + PIConditionVariable cv; + mutable PIMutex que_mutex, last_mutex, wait_mutex; }; #endif // PIGRABBERBASE_H diff --git a/libs/main/thread/pipipelinethread.h b/libs/main/thread/pipipelinethread.h index 6cfae3fa..0eade22f 100644 --- a/libs/main/thread/pipipelinethread.h +++ b/libs/main/thread/pipipelinethread.h @@ -55,9 +55,13 @@ public: //! \~english Stops the stage thread and may terminate it forcibly if it does not finish in time. //! \~russian Останавливает поток стадии и может принудительно завершить его, если он не завершится вовремя. + //! \~\note + //! \~english This is the correct way to stop a stage: it also wakes any thread blocked in \a enqueue(). + //! \~russian Это корректный способ остановки стадии: дополнительно будит поток, заблокированный в \a enqueue(). ~PIPipelineThread() { stop(); - cv.notifyAll(); + cv_not_empty.notifyAll(); + cv_not_full.notifyAll(); if (!waitForFinish(1000)) { piCoutObj << "terminating self thread"; terminate(); @@ -68,7 +72,7 @@ public: //! \~russian Подключает к следующему этапу конвейера через event-уведомления template void connectTo(PIPipelineThread * next) { - CONNECT3(void, Tout, bool, bool *, this, calculated, next, enqueue); + CONNECT3(void, Tout, bool, bool *, this, calculated, next, fromCalculated); } //! \~\handlers @@ -86,16 +90,11 @@ public: EVENT_HANDLER3(void, enqueue, const Tin &, v, bool, wait, bool *, overload) { mutex.lock(); // piCoutObj << "enque" << overload; - if (wait && max_size != 0) { - mutex_wait.lock(); - while (in.size() >= max_size) - cv_wait.wait(mutex_wait); - mutex_wait.unlock(); - } - if (max_size == 0 || in.size() < max_size) { + if (wait && max_size != 0) cv_not_full.wait(mutex, [this] { return terminating || in.size() < max_size; }); + if (!terminating && (max_size == 0 || in.size() < max_size)) { in.enqueue(v); - cv.notifyAll(); if (overload) *overload = false; + cv_not_empty.notifyOne(); } else { if (overload) *overload = true; } @@ -155,27 +154,30 @@ public: //! \~english Clear input queue //! \~russian Очищает входную очередь + //! \~\note + //! \~english Wakes producer threads blocked in \a enqueue() with \a wait equal to \c true. + //! \~russian Будит потоки-продюсеры, заблокированные в \a enqueue() с \a wait равным \c true. void clear() { mutex.lock(); - mutex_wait.lock(); in.clear(); - cv_wait.notifyAll(); - mutex_wait.unlock(); + cv_not_full.notifyAll(); mutex.unlock(); } //! \~english Stops calculation and waits for thread finish //! \~russian Останавливает вычисления и ожидает завершения потока + //! \~\details + //! \~english This is the correct way to stop a stage: it wakes the consumer loop and any producer blocked in \a enqueue(). + //! \~russian Это корректный способ остановки стадии: будит цикл потребителя и любой поток, заблокированный в \a enqueue(). //! \~\note //! \~english If the stage does not stop within \a wait_delay, it may be terminated forcibly. //! \~russian Если стадия не остановится за \a wait_delay, она может быть принудительно завершена. void stopCalc(int wait_delay = 100) { if (isRunning()) { stop(); - cv.notifyAll(); + cv_not_empty.notifyAll(); + cv_not_full.notifyAll(); if (!waitForFinish(wait_delay)) { - mutex_last.unlock(); - mutex.unlock(); terminate(); } } @@ -212,6 +214,7 @@ public: mutex.lock(); max_size = count; if (max_size > 0 && in.size() > max_size) in.resize(max_size); + cv_not_full.notifyAll(); mutex.unlock(); } @@ -246,20 +249,29 @@ protected: private: void begin() override { cnt = 0; } + //! \~english Internal adapter for \a connectTo(): forwards \a calculated() into \a enqueue(). + //! \~russian Внутренний адаптер для \a connectTo(): пересылает \a calculated() в \a enqueue(). + //! \~\details + //! \~english + //! Event dispatch passes event arguments by value, while \a enqueue() takes its data by reference. + //! \~For POD \a Tout types the direct connection would read the value as a pointer, so the connection + //! \~goes through this by-value adapter. + //! \~russian + //! Диспетчер событий передает аргументы события по значению, тогда как \a enqueue() принимает данные + //! \~по ссылке. Для POD-типов \a Tout прямое соединение прочитало бы значение как указатель, поэтому + //! \~подключение идет через этот адаптер, принимающий аргумент по значению. + EVENT_HANDLER3(void, fromCalculated, Tout, v, bool, wait, bool *, overload) { enqueue(v, wait, overload); } + void run() override { mutex.lock(); - while (in.isEmpty()) { - cv.wait(mutex); - if (terminating) { - mutex.unlock(); - return; - } + cv_not_empty.wait(mutex, [this] { return terminating || !in.isEmpty(); }); + if (terminating) { + mutex.unlock(); + return; } - mutex_wait.lock(); Tin t = in.dequeue(); mutex.unlock(); - cv_wait.notifyAll(); - mutex_wait.unlock(); + cv_not_full.notifyOne(); bool ok = true; Tout r = calc(t, ok); if (ok) { @@ -268,13 +280,14 @@ private: mutex_last.unlock(); cnt++; // piCoutObj << "calc ok"; - calculated(r, wait_next_pipe); + calculated(r, wait_next_pipe, nullptr); } // piCoutObj << "run ok"; } - PIMutex mutex, mutex_wait; - PIConditionVariable cv, cv_wait; + PIMutex mutex; + PIConditionVariable cv_not_empty; + PIConditionVariable cv_not_full; PIMutex mutex_last; bool wait_next_pipe; ullong cnt; diff --git a/tests/thread/pigrabberbase_test.cpp b/tests/thread/pigrabberbase_test.cpp new file mode 100644 index 00000000..1199501d --- /dev/null +++ b/tests/thread/pigrabberbase_test.cpp @@ -0,0 +1,218 @@ +#include "pibase_macros.h" +#include "pigrabberbase.h" +#include "pisystemtime.h" +#include "pithread.h" +#include "pitime.h" + +#include "gtest/gtest.h" +#include +#include + +namespace { + +class FakeGrabber: public PIGrabberBase { + PIOBJECT_SUBCLASS(FakeGrabber, PIGrabberBase) + +public: + std::atomic_int open_count{0}; + std::atomic_bool open_ok{true}; + int open_fail_times = 0; // fail first N open attempts + + std::atomic_bool data_available{false}; + int ready_value = 7; + std::atomic_int poll_count{0}; + std::atomic_int record_count{0}; + +protected: + bool openInternal() override { + open_count++; + if (open_fail_times > 0) { + open_fail_times--; + return false; + } + return open_ok; + } + + void closeInternal() override {} + + int get(int & val) override { + poll_count++; + if (!data_available) return 1; // no item ready yet + val = ready_value; + return 0; + } + + void record(const int & val) override { + (void)val; + record_count++; + } +}; + +bool waitForQueSize(PIGrabberBase & g, int size, double timeout_s = 5.) { + PITimeMeasurer tm; + while (g.queSize() < size && tm.elapsed_s() < timeout_s) + piMSleep(1); + return g.queSize() >= size; +} + +} // namespace + + +TEST(PIGrabberBase_Basic, CaptureAndLast) { + FakeGrabber g; + g.data_available = true; + g.ready_value = 7; + + std::atomic_int ready_events(0); + CONNECTL(&g, dataReady, [&ready_events]() { ready_events++; }); + + g.start(); + ASSERT_TRUE(waitForQueSize(g, 5)); + + EXPECT_EQ(g.last(), 7); + EXPECT_GE(ready_events.load(), 5); + + g.stopGrabber(); + EXPECT_FALSE(g.isRunning()); + EXPECT_FALSE(g.isOpened()); + + // queue is frozen now, counters must match exactly + EXPECT_EQ(g.diag().state().received_packets, (ullong)g.queSize()); + EXPECT_EQ(g.dequeue(), 7); +} + + +TEST(PIGrabberBase_Wake, WakeUpBeatsPollDelay) { + FakeGrabber g; + g.data_available = false; + g.setPollDelay(PISystemTime::fromMilliseconds(1000)); + + g.start(); + piMSleep(100); + ASSERT_EQ(g.queSize(), 0); // loop is in its 1 second wait + + g.ready_value = 5; + g.data_available = true; + g.wakeUp(); + + PITimeMeasurer tm; + while (g.queSize() < 1 && tm.elapsed_s() < 2.) + piMSleep(1); + EXPECT_GE(g.queSize(), 1); + EXPECT_EQ(g.last(), 5); + EXPECT_LT(tm.elapsed_m(), 500.); // woke up long before the 1 second poll interval + + g.stopGrabber(); + EXPECT_FALSE(g.isRunning()); +} + + +TEST(PIGrabberBase_Stop, FastStopDuringLongPollWait) { + FakeGrabber g; + g.data_available = false; + g.setPollDelay(PISystemTime::fromMilliseconds(2000)); + + g.start(); + piMSleep(100); // loop is now waiting up to 2 seconds + + PITimeMeasurer tm; + g.stopGrabber(); + EXPECT_LT(tm.elapsed_m(), 500.); // stop interrupts the wait, does not wait for it + EXPECT_FALSE(g.isRunning()); +} + + +TEST(PIGrabberBase_Open, RetryUntilSuccess) { + FakeGrabber g; + g.open_fail_times = 2; // first two opens fail, the third succeeds + g.data_available = true; + + g.start(); + + PITimeMeasurer tm; + while (!g.isOpened() && tm.elapsed_s() < 3.) + piMSleep(1); + EXPECT_TRUE(g.isOpened()); + EXPECT_LE(tm.elapsed_s(), 1.); // two 200 ms retry waits + EXPECT_EQ(g.open_count.load(), 3); + + ASSERT_TRUE(waitForQueSize(g, 1)); + EXPECT_EQ(g.last(), g.ready_value); + + g.stopGrabber(); + EXPECT_FALSE(g.isRunning()); +} + + +TEST(PIGrabberBase_Open, FastStopDuringOpenRetry) { + FakeGrabber g; + g.open_ok = false; // opens always fail, loop retries every 200 ms + + g.start(); + piMSleep(100); // loop is in the 200 ms open-retry wait + + PITimeMeasurer tm; + g.stopGrabber(); + EXPECT_LT(tm.elapsed_m(), 150.); // stop interrupts the retry wait + EXPECT_FALSE(g.isRunning()); + EXPECT_FALSE(g.isOpened()); +} + + +TEST(PIGrabberBase_Recording, StartStopRecord) { + FakeGrabber g; + g.data_available = true; + + g.start(); + ASSERT_TRUE(waitForQueSize(g, 2)); + + EXPECT_FALSE(g.isRecording()); + g.startRecord("test"); + ASSERT_TRUE(g.isRecording()); + + PITimeMeasurer tm; + while (g.record_count.load() < 1 && tm.elapsed_s() < 5.) + piMSleep(1); + EXPECT_GE(g.record_count.load(), 1); + + g.stopRecord(); + EXPECT_FALSE(g.isRecording()); + + g.stopGrabber(); + EXPECT_FALSE(g.isRunning()); +} + + +TEST(PIGrabberBase_Close, RestartReopens) { + FakeGrabber g; + g.data_available = true; + + g.start(); + ASSERT_TRUE(waitForQueSize(g, 1)); + EXPECT_EQ(g.last(), g.ready_value); + + // stop the data flow, then let the loop settle before clearing + g.data_available = false; + piMSleep(20); + g.restart(); // clear queue and close + EXPECT_FALSE(g.isOpened()); + EXPECT_TRUE(g.isEmpty()); + + // the running loop must reopen the grabber on its own + PITimeMeasurer tm; + while (!g.isOpened() && tm.elapsed_s() < 3.) + piMSleep(1); + EXPECT_TRUE(g.isOpened()); + + g.stopGrabber(); + EXPECT_FALSE(g.isRunning()); +} + + +TEST(PIGrabberBase_PollDelay, GetterSetter) { + FakeGrabber g; + EXPECT_DOUBLE_EQ(g.pollDelay().toMilliseconds(), (double)PIP_MIN_MSLEEP); + + g.setPollDelay(PISystemTime::fromMilliseconds(55)); + EXPECT_DOUBLE_EQ(g.pollDelay().toMilliseconds(), 55.); +} diff --git a/tests/thread/pipipelinethread_test.cpp b/tests/thread/pipipelinethread_test.cpp new file mode 100644 index 00000000..f7ccb5c8 --- /dev/null +++ b/tests/thread/pipipelinethread_test.cpp @@ -0,0 +1,232 @@ +#include "pipipelinethread.h" +#include "pisystemtime.h" +#include "pithread.h" +#include "pitime.h" + +#include "gtest/gtest.h" +#include +#include +#include + +namespace { + +class Stage: public PIPipelineThread { + PIOBJECT_SUBCLASS(Stage, PIPipelineThread) + +public: + std::function f; + bool drop_odd = false; + + Stage(): f([](int v) { return v; }) {} + +protected: + int calc(int & v, bool & ok) override { + ok = !drop_odd || !(v & 1); + return f(v); + } +}; + +// Stage with slow calculation, used to keep the consumer busy +class SlowStage: public PIPipelineThread { + PIOBJECT_SUBCLASS(SlowStage, PIPipelineThread) + +public: + int calc_ms = 300; + +protected: + int calc(int & v, bool & ok) override { + ok = true; + piMSleep(calc_ms); + return v; + } +}; + +bool waitForCount(const PIPipelineThread & stage, ullong count, double timeout_s = 5.) { + PITimeMeasurer tm; + while (stage.counter() < count && tm.elapsed_s() < timeout_s) + piMSleep(1); + return stage.counter() >= count; +} + +} // namespace + + +TEST(PIPipelineThread_Basic, ProcessItems) { + Stage stage; + stage.f = [](int v) { return v * 2; }; + stage.start(); + + const int N = 100; + for (int i = 0; i < N; ++i) + stage.enqueue(i); + + EXPECT_TRUE(waitForCount(stage, N)); + EXPECT_EQ(stage.counter(), N); + EXPECT_EQ(stage.queSize(), 0); + EXPECT_TRUE(stage.isEmpty()); + EXPECT_EQ(stage.getLast(), (N - 1) * 2); + + stage.stopCalc(2000); + EXPECT_FALSE(stage.isRunning()); +} + + +TEST(PIPipelineThread_Basic, DropByCalc) { + Stage stage; + stage.drop_odd = true; + stage.start(); + + for (int i = 0; i < 20; ++i) + stage.enqueue(i); + + EXPECT_TRUE(waitForCount(stage, 10)); + EXPECT_EQ(stage.counter(), 10); // only even values were published + EXPECT_EQ(stage.queSize(), 0); + EXPECT_EQ(stage.getLast(), 18); + + stage.stopCalc(2000); + EXPECT_FALSE(stage.isRunning()); +} + + +TEST(PIPipelineThread_Chain, TwoStages) { + Stage s1; + Stage s2; + s1.f = [](int v) { return v * 2; }; + s2.f = [](int v) { return v + 10; }; + s1.connectTo(&s2); + + s1.start(); + s2.start(); + + const int N = 50; + for (int i = 0; i < N; ++i) + s1.enqueue(i); + + EXPECT_TRUE(waitForCount(s1, N)); + EXPECT_TRUE(waitForCount(s2, N)); + EXPECT_EQ(s2.queSize(), 0); + EXPECT_EQ(s2.getLast(), (N - 1) * 2 + 10); + + s2.stopCalc(2000); + s1.stopCalc(2000); + EXPECT_FALSE(s1.isRunning()); + EXPECT_FALSE(s2.isRunning()); +} + + +TEST(PIPipelineThread_Bounded, DropWhenFull) { + Stage stage; + stage.setMaxQueSize(2); + EXPECT_EQ(stage.maxQueSize(), 2u); + + bool ov1 = false; + bool ov2 = false; + bool ov3 = true; + stage.enqueue(1, false, &ov1); + stage.enqueue(2, false, &ov2); + stage.enqueue(3, false, &ov3); + + EXPECT_FALSE(ov1); + EXPECT_FALSE(ov2); + EXPECT_TRUE(ov3); + EXPECT_EQ(stage.queSize(), 2); +} + + +TEST(PIPipelineThread_Bounded, ResizeTrimsQueue) { + Stage stage; + stage.setMaxQueSize(10); + for (int i = 0; i < 5; ++i) + stage.enqueue(i); + EXPECT_EQ(stage.queSize(), 5); + + stage.setMaxQueSize(2); + EXPECT_EQ(stage.queSize(), 2); +} + + +TEST(PIPipelineThread_Bounded, WaitUnblockedByClear) { + Stage stage; + stage.setMaxQueSize(1); + stage.enqueue(1, false); + ASSERT_EQ(stage.queSize(), 1); + + std::atomic_bool done(false); + std::atomic_bool producer_overload(true); + std::thread producer([&stage, &done, &producer_overload] { + bool overload = true; + stage.enqueue(2, true, &overload); + producer_overload = overload; + done = true; + }); + + piMSleep(100); + EXPECT_FALSE(done); + EXPECT_EQ(stage.queSize(), 1); // 2 is still waiting for space + + stage.clear(); // frees space and wakes the producer + + producer.join(); + EXPECT_FALSE(producer_overload); + EXPECT_EQ(stage.queSize(), 1); // only 2 remains in the queue +} + + +TEST(PIPipelineThread_Bounded, WaitUnblockedByStopCalc) { + SlowStage stage; + stage.setMaxQueSize(1); + stage.start(); + + // First item goes to the slow calc, then the second item fills the queue + stage.enqueue(1, false); + PITimeMeasurer tm; + while (stage.queSize() > 0 && tm.elapsed_s() < 5.) + piMSleep(1); // wait until the consumer takes item 1 into the slow calc + stage.enqueue(2, false); + ASSERT_EQ(stage.queSize(), 1); + + std::atomic_bool done(false); + bool overload = false; + std::thread producer([&stage, &done, &overload] { + stage.enqueue(3, true, &overload); + done = true; + }); + + piMSleep(100); + EXPECT_FALSE(done); // blocked on the full queue + + stage.stopCalc(2000); + + // producer must be woken by stopCalc, not hang forever + producer.join(); + EXPECT_TRUE(done); + EXPECT_TRUE(overload); // item was dropped because the stage is stopping + EXPECT_FALSE(stage.isRunning()); +} + + +TEST(PIPipelineThread_Stop, FastStopOnEmptyQueue) { + Stage stage; + stage.start(); + piMSleep(50); // let the thread reach the wait + + PITimeMeasurer tm; + stage.stopCalc(1000); + EXPECT_LT(tm.elapsed_m(), 500.); + EXPECT_FALSE(stage.isRunning()); +} + + +TEST(PIPipelineThread_Stop, StopWithPendingItems) { + Stage stage; + stage.start(); + for (int i = 0; i < 10; ++i) + stage.enqueue(i); + piMSleep(50); + + PITimeMeasurer tm; + stage.stopCalc(2000); + EXPECT_LT(tm.elapsed_m(), 1500.); + EXPECT_FALSE(stage.isRunning()); +}