ircd::json: Add flush watermarking to json::stack.

include/ircd/json/stack.h

@@ -50,6 +50,8 @@ struct ircd::json::stack
 	window_buffer buf;
 	flush_callback flusher;
 	std::exception_ptr eptr;
+	size_t hiwat;                      ///< autoflush watermark
+	size_t lowat;                      ///< flush(false) call min watermark
 
 	object *co {nullptr};              ///< The root object instance.
 	array *ca {nullptr};               ///< Could be union with top_object but
@@ -63,13 +65,17 @@ struct ircd::json::stack
 	bool closed() const;               ///< No stacking in progress.
 	bool clean() const;                ///< Never opened.
 	bool done() const;                 ///< Opened and closed.
-
 	size_t remaining() const;
 	const_buffer completed() const;
-	bool flush();
+
+	bool flush(const bool &force = false);
 	void clear();
 
-	stack(const mutable_buffer &, flush_callback = {});
+	stack(const mutable_buffer &,
+	      flush_callback           = {},
+	      const size_t &hiwat      = -1,
+	      const size_t &lowat      = 0);
+
 	stack(stack &&) noexcept;
 	stack(const stack &) = delete;
 	~stack() noexcept;

ircd/json.cc

@@ -416,9 +416,13 @@ ircd::json::input<it>::throws_exceeded()
 //
 
 ircd::json::stack::stack(const mutable_buffer &buf,
-                         flush_callback flusher)
+                         flush_callback flusher,
+                         const size_t &hiwat,
+                         const size_t &lowat)
 :buf{buf}
 ,flusher{std::move(flusher)}
+,hiwat{hiwat}
+,lowat{lowat}
 {
 }
 
@@ -427,6 +431,8 @@ noexcept
 :buf{std::move(other.buf)}
 ,flusher{std::move(other.flusher)}
 ,eptr{std::move(other.eptr)}
+,hiwat{std::move(other.hiwat)}
+,lowat{std::move(other.lowat)}
 ,co{std::move(other.co)}
 ,ca{std::move(other.ca)}
 {
@@ -467,7 +473,7 @@ try
 			size(buf.base)
 		};
 
-		if(!flush())
+		if(!flush(true))
 			return;
 
 		if(unlikely(expect > buf.remaining())) throw print_error
@@ -483,6 +489,8 @@ try
 	});
 
 	if(!buf.remaining())
+		flush(true);
+	else if(buf.consumed() >= hiwat)
 		flush();
 }
 catch(...)
@@ -499,7 +507,7 @@ ircd::json::stack::rethrow_exception()
 }
 
 bool
-ircd::json::stack::flush()
+ircd::json::stack::flush(const bool &force)
 try
 {
 	if(!flusher)
@@ -508,6 +516,9 @@ try
 	if(unlikely(eptr))
 		return false;
 
+	if(!force && buf.consumed() < lowat)
+		return false;
+
 	// The user returns the portion of the buffer they were able to flush
 	// rather than forcing them to wait on their sink to flush the whole
 	// thing, they can continue with us for a little while more.
@@ -644,7 +655,7 @@ noexcept
 	assert(cm == nullptr);
 	s->append("}"_sv);
 
-	if(pm)
+	if(pm) // branch taken if member of object
 	{
 		assert(pa == nullptr);
 		assert(pm->ca == nullptr);
@@ -653,7 +664,7 @@ noexcept
 		return;
 	}
 
-	if(pa)
+	if(pa) // branch taken if value in array
 	{
 		assert(pm == nullptr);
 		assert(pa->ca == nullptr);
@@ -663,12 +674,13 @@ noexcept
 		return;
 	}
 
+	// branch taken if top of stack:: (w/ final flush)
 	assert(s->co == this);
 	assert(s->ca == nullptr);
 	assert(pm == nullptr && pa == nullptr);
 	s->co = nullptr;
 	assert(s->done());
-	s->flush();
+	s->flush(true);
 }
 
 //
@@ -741,7 +753,7 @@ noexcept
 	assert(ca == nullptr);
 	s->append("]"_sv);
 
-	if(pm)
+	if(pm) // branch taken if member of object
 	{
 		assert(pa == nullptr);
 		assert(pm->ca == this);
@@ -750,7 +762,7 @@ noexcept
 		return;
 	}
 
-	if(pa)
+	if(pa) // branch taken if value in array
 	{
 		assert(pm == nullptr);
 		assert(pa->ca == this);
@@ -760,12 +772,13 @@ noexcept
 		return;
 	}
 
+	// branch taken if top of stack:: (w/ final flush)
 	assert(s->ca == this);
 	assert(s->co == nullptr);
 	assert(pm == nullptr && pa == nullptr);
 	s->ca = nullptr;
 	assert(s->done());
-	s->flush();
+	s->flush(true);
 }
 
 void