thread: PIGrabberBase and PIPipelineThread on PIConditionVariable instead of delays

- PIGrabberBase: replace piMSleep(200)/piMinSleep() polling with interruptible
  PIConditionVariable::waitFor waits; add wakeUp() handler for event-driven
  polling and setPollDelay()/pollDelay() for the fallback poll interval
- PIPipelineThread: single mutex + cv_not_empty/cv_not_full (bounded queue
  pattern), terminating in wait predicates so stop/clear wake blocked
  producers/consumers; drop bogus mutex unlocks in stopCalc()
- fix potential bug: calculated() was emitted with 2 of 3 arguments
- fix potential bug: connectTo() ABI mismatch for POD Tout (event dispatch
  passes first arg by value, handler took it by reference) - route through
  a by-value fromCalculated() adapter
- PIOBJECT_SUBCLASS/PIOBJECT_PARENT made variadic so template parents with
  commas (PIPipelineThread<int, int>) can be passed directly
- add gtest suites for both classes (tests/thread/)

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)                   \

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<T> 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

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<typename T>
 	void connectTo(PIPipelineThread<Tout, T> * 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;

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 <atomic>
+#include <thread>
+
+namespace {
+
+class FakeGrabber: public PIGrabberBase<int> {
+	PIOBJECT_SUBCLASS(FakeGrabber, PIGrabberBase<int>)
+
+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<int> & 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.);
+}

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 <atomic>
+#include <functional>
+#include <thread>
+
+namespace {
+
+class Stage: public PIPipelineThread<int, int> {
+	PIOBJECT_SUBCLASS(Stage, PIPipelineThread<int, int>)
+
+public:
+	std::function<int(int)> 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<int, int> {
+	PIOBJECT_SUBCLASS(SlowStage, PIPipelineThread<int, int>)
+
+public:
+	int calc_ms = 300;
+
+protected:
+	int calc(int & v, bool & ok) override {
+		ok = true;
+		piMSleep(calc_ms);
+		return v;
+	}
+};
+
+bool waitForCount(const PIPipelineThread<int, int> & 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());
+}