Keep track of timestamps and skip old logs

native/jni/utils/include/BlockingQueue.h

@@ -6,10 +6,10 @@
 template<typename T>
 class BlockingQueue {
 	std::deque<T> deque{};
-	pthread_mutex_t lock;
-	pthread_cond_t cond;
+	pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
+	pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
+	bool cancelled = false;
 public:
-	BlockingQueue() : lock(PTHREAD_MUTEX_INITIALIZER), cond(PTHREAD_COND_INITIALIZER) {}
 	~BlockingQueue();
 	T take();
 	T &front();
@@ -19,8 +19,15 @@ public:
 	template< class... Args >
 	void emplace_back(Args&&... args);
 	void clear();
+	void cancel();
 };
 
+#define run_and_notify(block) \
+pthread_mutex_lock(&this->lock); \
+block \
+pthread_cond_signal(&this->cond); \
+pthread_mutex_unlock(&this->lock);
+
 template<typename T>
 BlockingQueue<T>::~BlockingQueue() {
 	pthread_mutex_destroy(&lock);
@@ -30,8 +37,11 @@ BlockingQueue<T>::~BlockingQueue() {
 template<typename T>
 T BlockingQueue<T>::take() {
 	pthread_mutex_lock(&lock);
-	while (deque.empty())
+	cancelled = false;
+	while (deque.empty() && !cancelled)
 		pthread_cond_wait(&cond, &lock);
+	if (cancelled)
+		pthread_exit(nullptr);
 	T ret(std::move(deque.front()));
 	deque.pop_front();
 	pthread_mutex_unlock(&lock);
@@ -40,41 +50,49 @@ T BlockingQueue<T>::take() {
 
 template<typename T>
 void BlockingQueue<T>::put(const T &s) {
-	pthread_mutex_lock(&lock);
-	deque.push_back(s);
-	pthread_cond_signal(&cond);
-	pthread_mutex_unlock(&lock);
+	run_and_notify({ deque.push_back(s); })
 }
 
 template<typename T>
 void BlockingQueue<T>::put(T &&s) {
-	pthread_mutex_lock(&lock);
-	deque.push_back(std::move(s));
-	pthread_cond_signal(&cond);
-	pthread_mutex_unlock(&lock);
+	run_and_notify({ deque.push_back(std::move(s)); })
 }
 
 template<typename T>
 T &BlockingQueue<T>::front() {
-	return deque.front();
+	pthread_mutex_lock(&lock);
+	auto &ret = deque.front();
+	pthread_mutex_unlock(&lock);
+	return ret;
 }
 
 template<typename T>
 T &BlockingQueue<T>::back() {
-	return deque.back();
+	pthread_mutex_lock(&lock);
+	auto &ret = deque.back();
+	pthread_mutex_unlock(&lock);
+	return ret;
 }
 
 template<typename T>
 template<class... Args>
 void BlockingQueue<T>::emplace_back(Args &&... args) {
-	pthread_mutex_lock(&lock);
-	deque.emplace_back(std::forward<Args>(args)...);
-	pthread_cond_signal(&cond);
-	pthread_mutex_unlock(&lock);
+	run_and_notify({ deque.emplace_back(std::forward<Args>(args)...); })
 }
 
 template<typename T>
 void BlockingQueue<T>::clear() {
+	pthread_mutex_lock(&lock);
 	std::deque<T> t;
 	deque.swap(t);
+	pthread_mutex_unlock(&lock);
+}
+
+template<typename T>
+void BlockingQueue<T>::cancel() {
+	run_and_notify({
+		cancelled = true;
+		std::deque<T> t;
+		deque.swap(t);
+	})
 }