mpsc_queue: replace recursion with loop

in try_consume_batch. In case producers constantly produce
new data, try_consume_batch will be called over and over again
leading to stack overflow.

Author: igchor

include/libpmemobj++/experimental/mpsc_queue.hpp

@@ -263,50 +263,57 @@ mpsc_queue::try_consume_batch(Function &&f)
 		throw pmem::transaction_scope_error(
 			"Function called inside a transaction scope.");
 
-	size_t offset;
-	size_t len = ringbuf_consume(ring_buffer.get(), &offset);
+	bool consumed = false;
+
+	/* Need to call try_consume twice, as some data may be at the end
+	 * of buffer, and some may be at the beginning. Ringbuffer does not
+	 * merge those two pats into one try_consume. If all data was
+	 * consumed during first try_consume, second would fail. */
+	for (int i = 0; i < 2; i++) {
+		size_t offset;
+		size_t len = ringbuf_consume(ring_buffer.get(), &offset);
 
 #if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
-	ANNOTATE_HAPPENS_AFTER(ring_buffer.get());
+		ANNOTATE_HAPPENS_AFTER(ring_buffer.get());
 #endif
 
-	if (!len)
-		return false;
-
-	auto data = buf + offset;
+		if (!len)
+			return consumed;
 
-	auto begin = iterator(data, data + len);
-	auto end = iterator(data + len, data + len);
+		auto data = buf + offset;
+		auto begin = iterator(data, data + len);
+		auto end = iterator(data + len, data + len);
 
-	auto elements_to_consume = begin != end;
+		pmem::obj::flat_transaction::run(pop, [&] {
+			if (begin != end) {
+				consumed = true;
+				f(batch_type(begin, end));
+			}
 
-	pmem::obj::flat_transaction::run(pop, [&] {
-		if (elements_to_consume)
-			f(batch_type(begin, end));
+			auto b = reinterpret_cast<first_block *>(data);
+			clear_cachelines(b, len);
 
-		auto b = reinterpret_cast<first_block *>(data);
-		clear_cachelines(b, len);
-
-		if (offset + len < buff_size_)
-			pmem->written = offset + len;
-		else if (offset + len == buff_size_)
-			pmem->written = 0;
-		else
-			assert(false);
-	});
+			if (offset + len < buff_size_)
+				pmem->written = offset + len;
+			else if (offset + len == buff_size_)
+				pmem->written = 0;
+			else
+				assert(false);
+		});
 
 #if LIBPMEMOBJ_CPP_VG_HELGRIND_ENABLED
-	ANNOTATE_HAPPENS_BEFORE(ring_buffer.get());
+		ANNOTATE_HAPPENS_BEFORE(ring_buffer.get());
 #endif
 
-	ringbuf_release(ring_buffer.get(), len);
+		ringbuf_release(ring_buffer.get(), len);
+
+		/* XXX: it would be better to call f once - hide
+		 * wraparound behind iterators */
+		/* XXX: add param to ringbuf_consume and do not
+		 * call store_explicit in consume */
+	}
 
-	/* XXX: it would be better to call f once - hide
-	 * wraparound behind iterators */
-	/* XXX: add param to ringbuf_consume and do not
-	 * call store_explicit in consume */
-	return try_consume_batch(std::forward<Function>(f)) ||
-		elements_to_consume;
+	return consumed;
 }
 
 inline mpsc_queue::worker::worker(mpsc_queue *q)