0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2024-11-29 18:22:50 +01:00

ircd::server: Add preliminary write loop; minor cleanup.

This commit is contained in:
Jason Volk 2018-01-14 20:51:39 -08:00
parent d62afeb0a7
commit 368838d5a5
4 changed files with 125 additions and 41 deletions

View file

@ -36,9 +36,9 @@ struct ircd::server::link
bool busy() const;
protected:
void discard_read();
const_buffer process_read_next(const const_buffer &, struct request::tag &, bool &done);
bool process_read(const_buffer &);
void discard_read();
void handle_readable_success();
void handle_readable(const error_code &);
void wait_readable();

View file

@ -85,16 +85,21 @@ struct ircd::server::request::tag
{
server::request *request;
ctx::promise<http::code> p;
size_t head_written {0};
size_t content_written {0};
size_t head_read {0};
size_t content_read {0};
mutable_buffer make_content_buffer() const;
mutable_buffer make_head_buffer() const;
mutable_buffer make_read_content_buffer() const;
mutable_buffer make_read_head_buffer() const;
const_buffer read_content(const const_buffer &, bool &done);
const_buffer read_head(const const_buffer &, bool &done);
public:
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);

View file

@ -2198,17 +2198,16 @@ noexcept try
{
using namespace boost::system::errc;
reply_set = false;
switch(ec.value())
{
case operation_canceled:
return;
case success:
reply_set = false;
set_handle();
break;
case operation_canceled:
log::debug("Resolver leaving");
return;
default:
throw boost::system::system_error(ec);
}

View file

@ -214,8 +214,47 @@ const
{
assert(request);
return !request? mutable_buffer{}:
!request->in.head.status? make_head_buffer():
make_content_buffer();
!request->in.head.status? make_read_head_buffer():
make_read_content_buffer();
}
void
ircd::server::request::tag::wrote_buffer(const const_buffer &buffer)
{
assert(request);
const auto &req{*request};
if(head_written < size(req.out.head))
{
assert(data(buffer) == data(req.out.head));
head_written += size(buffer);
}
else if(content_written < size(req.out.content))
{
assert(data(buffer) == data(req.out.content));
content_written += size(buffer);
}
}
ircd::const_buffer
ircd::server::request::tag::make_write_buffer()
const
{
assert(request);
const auto &req{*request};
if(head_written < size(req.out.head))
{
const size_t remain{size(req.out.head) - head_written};
const const_buffer window{data(req.out.head) + head_written, remain};
return window;
}
else if(content_written < size(req.out.content))
{
const size_t remain{size(req.out.content) - content_written};
const const_buffer window{data(req.out.content) + content_written, remain};
return window;
}
return {};
}
ircd::const_buffer
@ -346,7 +385,7 @@ ircd::server::request::tag::read_content(const const_buffer &buffer,
}
ircd::mutable_buffer
ircd::server::request::tag::make_head_buffer()
ircd::server::request::tag::make_read_head_buffer()
const
{
assert(request);
@ -372,7 +411,7 @@ const
}
ircd::mutable_buffer
ircd::server::request::tag::make_content_buffer()
ircd::server::request::tag::make_read_content_buffer()
const
{
assert(request);
@ -624,8 +663,7 @@ ircd::server::link::submit(request &request)
queue.emplace(end(queue), request)
};
write(*socket, request.out.head);
write(*socket, request.out.content);
wait_writable();
}
void
@ -707,7 +745,49 @@ ircd::server::link::wait_writable()
void
ircd::server::link::handle_writable(const error_code &ec)
{
std::cout << this << " writable: " << string(ec) << std::endl;
using namespace boost::system::errc;
using boost::system::system_category;
switch(ec.value())
{
case success:
break;
case operation_canceled:
break;
default:
throw boost::system::system_error{ec};
}
for(auto &tag : queue)
{
const const_buffer buffer
{
tag.make_write_buffer()
};
if(empty(buffer))
continue;
const size_t bytes
{
write_any(*socket, buffer)
};
const const_buffer written
{
data(buffer), bytes
};
tag.wrote_buffer(written);
if(bytes < size(buffer))
{
wait_writable();
break;
}
}
}
void
@ -772,32 +852,6 @@ ircd::server::link::handle_readable_success()
while(!queue.empty());
}
void
ircd::server::link::discard_read()
{
const size_t discard
{
available(*socket)
};
const size_t discarded
{
discard_any(*socket, discard)
};
// Shouldn't ever be hit because the read() within discard() throws
// the pending error like an eof.
log::warning("Link discarded %zu of %zu unexpected bytes",
discard,
discarded);
assert(0);
// for non-assert builds just in case; so this doesn't get loopy with
// discarding zero with an empty queue...
if(unlikely(!discard || !discarded))
throw assertive("Queue is empty and nothing to discard.");
}
/// Process as many read operations for one tag as possible
bool
ircd::server::link::process_read(const_buffer &overrun)
@ -873,6 +927,32 @@ ircd::server::link::process_read_next(const const_buffer &underrun,
return overrun;
}
void
ircd::server::link::discard_read()
{
const size_t discard
{
available(*socket)
};
const size_t discarded
{
discard_any(*socket, discard)
};
// Shouldn't ever be hit because the read() within discard() throws
// the pending error like an eof.
log::warning("Link discarded %zu of %zu unexpected bytes",
discard,
discarded);
assert(0);
// for non-assert builds just in case; so this doesn't get loopy with
// discarding zero with an empty queue...
if(unlikely(!discard || !discarded))
throw assertive("Queue is empty and nothing to discard.");
}
bool
ircd::server::link::busy()
const