ircd::server: Add stats accumulation; develop preliminary dispatch; various.

include/ircd/server/link.h

@@ -30,11 +30,8 @@ struct ircd::server::link
 	std::shared_ptr<net::socket> socket;         ///< link's socket
 	std::deque<tag> queue;                       ///< link's work queue
 
-	bool connected() const noexcept;
-	bool ready() const;
-	bool busy() const;
+	template<class F> size_t accumulate_tags(F&&) const;
 
-  protected:
 	void discard_read();
 	const_buffer process_read_next(const const_buffer &, tag &, bool &done);
 	bool process_read(const_buffer &);
@@ -52,12 +49,40 @@ struct ircd::server::link
 	void handle_open(std::exception_ptr);
 
   public:
+	// config related
+	size_t tag_max() const;
+	size_t tag_commit_max() const;
+
+	// indicator lights
+	bool connected() const noexcept;
+	bool ready() const;
+	bool busy() const;
+
+	// stats for upload-side bytes across all tags
+	size_t write_total() const;
+	size_t write_completed() const;
+	size_t write_remaining() const;
+
+	// stats for download-side bytes ~across all tags~; note: this is not
+	// accurate except for the one tag at the front of the queue having
+	// its response processed.
+	size_t read_total() const;         // see: tag::read_total() notes
+	size_t read_completed() const;     // see: tag::read_completed() notes
+	size_t read_remaining() const;     // see: tag::read_remaining() notes
+
+	// stats for tags
+	size_t tag_total() const;
+	size_t tag_committed() const;
+	size_t tag_uncommitted() const;
+
+	// request panel
+	tag cancel(request &);
+	void submit(request &);
+
+	// control panel
 	bool close(const net::close_opts &);
 	bool open(const net::open_opts &);
 
-	void cancel(request &);
-	void submit(request &);
-
 	link(server::node &);
 	~link() noexcept;
 };

include/ircd/server/node.h

@@ -28,7 +28,9 @@ struct ircd::server::node
 	std::exception_ptr eptr;
 	net::remote remote;
 	std::list<link> links;
-	ctx::dock dock;
+
+	template<class F> size_t accumulate_links(F&&) const;
+	template<class F> size_t accumulate_tags(F&&) const;
 
 	void handle_resolve(std::weak_ptr<node>, std::exception_ptr, const ipport &);
 	void resolve(const hostport &);
@@ -39,17 +41,39 @@ struct ircd::server::node
 	void handle_open(link &, std::exception_ptr);
 
   public:
-	size_t num_tags() const;
-	size_t num_links() const;
-	size_t num_links_busy() const;
-	size_t num_links_ready() const;
+	// config related
+	size_t link_max() const;
 
+	// stats for all links in node
+	size_t link_total() const;
+	size_t link_busy() const;
+	size_t link_ready() const;
+
+	// stats for all tags in all links in node
+	size_t tag_total() const;
+	size_t tag_committed() const;
+	size_t tag_uncommitted() const;
+
+	// stats for all upload-side bytes in all tags in all links
+	size_t write_total() const;
+	size_t write_completed() const;
+	size_t write_remaining() const;
+
+	// stats for download-side bytes in all tags in all links (note:
+	// see notes in link.h/tag.h about inaccuracy here).
+	size_t read_total() const;
+	size_t read_completed() const;
+	size_t read_remaining() const;
+
+	// link control panel
 	link &link_add(const size_t &num = 1);
-	link &link_get();
+	link &link_get(const request &);
 
+	// request panel
 	void cancel(request &);
 	void submit(request &);
 
+	// control panel
 	void interrupt();
 	void close();
 

include/ircd/server/request.h

@@ -27,6 +27,11 @@ namespace ircd::server
 	struct in;
 	struct out;
 	struct request;
+
+	size_t size(const in &);
+	size_t size(const out &);
+
+	void submit(const hostport &, request &);
 }
 
 /// Request data and options related to transmitting the request. This
@@ -76,3 +81,61 @@ struct ircd::server::request
 	request &operator=(const request &) = delete;
 	~request() noexcept;
 };
+
+inline
+ircd::server::request::request(const net::hostport &hostport,
+                               server::out out,
+                               server::in in)
+:tag{nullptr}
+,out{std::move(out)}
+,in{std::move(in)}
+{
+	submit(hostport, *this);
+}
+
+inline
+ircd::server::request::request(request &&o)
+noexcept
+:ctx::future<http::code>{std::move(o)}
+,tag{std::move(o.tag)}
+,out{std::move(o.out)}
+,in{std::move(o.in)}
+{
+	if(tag)
+		associate(*this, *tag, std::move(o));
+}
+
+inline ircd::server::request &
+ircd::server::request::operator=(request &&o)
+noexcept
+{
+	ctx::future<http::code>::operator=(std::move(o));
+	out = std::move(o.out);
+	in = std::move(o.in);
+	tag = std::move(o.tag);
+
+	if(tag)
+		associate(*this, *tag, std::move(o));
+
+	return *this;
+}
+
+inline
+ircd::server::request::~request()
+noexcept
+{
+	if(tag)
+		disassociate(*this, *tag);
+}
+
+inline size_t
+ircd::server::size(const in &in)
+{
+	return size(in.head_buffer) + size(in.content_buffer);
+}
+
+inline size_t
+ircd::server::size(const out &out)
+{
+	return size(out.head) + size(out.content);
+}

include/ircd/server/server.h

@@ -35,6 +35,9 @@ namespace ircd::server
 	extern ircd::log::log log;
 	extern std::map<string_view, std::shared_ptr<node>> nodes;
 
+	size_t link_total();
+	size_t tag_total();
+
 	bool exists(const net::hostport &);
 	node &find(const net::hostport &);
 	node &get(const net::hostport &);

include/ircd/server/tag.h

@@ -23,16 +23,20 @@
 namespace ircd::server
 {
 	struct tag;
+
+	void associate(request &, tag &);
+	void associate(request &, tag &, tag &&);
+	void associate(request &, tag &, request &&);
+	void disassociate(request &, tag &);
 }
 
 /// Internal portion of the request
 ///
 struct ircd::server::tag
 {
-	server::request *request;
+	server::request *request {nullptr};
 	ctx::promise<http::code> p;
-	size_t head_written {0};
-	size_t content_written {0};
+	size_t written {0};
 	size_t head_read {0};
 	size_t content_read {0};
 
@@ -43,12 +47,21 @@ struct ircd::server::tag
 	const_buffer read_head(const const_buffer &, bool &done);
 
   public:
+	size_t write_total() const;
+	size_t write_completed() const;
+	size_t write_remaining() const;
+
+	size_t read_total() const;               // not accurate until content-length known
+	size_t read_completed() const;           // reports all received so far
+	size_t read_remaining() const;           // not accurate until content-length known
+
 	const_buffer make_write_buffer() const;
 	void wrote_buffer(const const_buffer &);
 
 	mutable_buffer make_read_buffer() const;
 	const_buffer read_buffer(const const_buffer &, bool &done);
 
+	tag() = default;
 	tag(server::request &);
 	tag(tag &&) noexcept;
 	tag(const tag &) = delete;
@@ -56,3 +69,46 @@ struct ircd::server::tag
 	tag &operator=(const tag &) = delete;
 	~tag() noexcept;
 };
+
+inline
+ircd::server::tag::tag(server::request &request)
+{
+	associate(request, *this);
+}
+
+inline
+ircd::server::tag::tag(tag &&o)
+noexcept
+:request{std::move(o.request)}
+,p{std::move(o.p)}
+,written{std::move(o.written)}
+,head_read{std::move(o.head_read)}
+,content_read{std::move(o.content_read)}
+{
+	if(request)
+		associate(*request, *this, std::move(o));
+}
+
+inline ircd::server::tag &
+ircd::server::tag::operator=(tag &&o)
+noexcept
+{
+	request = std::move(o.request);
+	p = std::move(o.p);
+	written = std::move(o.written);
+	head_read = std::move(o.head_read);
+	content_read = std::move(o.content_read);
+
+	if(request)
+		associate(*request, *this, std::move(o));
+
+	return *this;
+}
+
+inline
+ircd::server::tag::~tag()
+noexcept
+{
+	if(request)
+		disassociate(*request, *this);
+}