Commit 0ab5fad6 authored by Dominik Charousset's avatar Dominik Charousset

Port to latest bounded buffer API

parent 6aca257b
...@@ -14,7 +14,7 @@ namespace caf::net { ...@@ -14,7 +14,7 @@ namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new /// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing. /// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer> template <class Buffer>
class consumer_adapter : public async::consumer, public ref_counted { class consumer_adapter final : public ref_counted, public async::consumer {
public: public:
using buf_ptr = intrusive_ptr<Buffer>; using buf_ptr = intrusive_ptr<Buffer>;
...@@ -36,11 +36,14 @@ public: ...@@ -36,11 +36,14 @@ public:
this->deref(); this->deref();
} }
template <class Policy, class OnNext, class OnError = unit_t> template <class Policy, class Observer>
bool consume(Policy policy, size_t demand, OnNext&& on_next, std::pair<bool, size_t> pull(Policy policy, size_t demand, Observer& dst) {
OnError on_error = OnError{}) { return buf_->pull(policy, demand, dst);
return buf_->consume(policy, demand, std::forward<OnNext>(on_next), }
std::move(on_error));
void cancel() {
buf_->cancel();
buf_ = nullptr;
} }
bool has_data() const noexcept { bool has_data() const noexcept {
...@@ -83,7 +86,7 @@ private: ...@@ -83,7 +86,7 @@ private:
} }
void on_wakeup() { void on_wakeup() {
if (has_data()) if (buf_ && buf_->has_consumer_event())
mgr_->mpx().register_writing(mgr_); mgr_->mpx().register_writing(mgr_);
} }
......
...@@ -93,7 +93,7 @@ public: ...@@ -93,7 +93,7 @@ public:
} }
bool handle_write_event(endpoint_manager& manager) override { bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id) CAF_LOG_TRACE(CAF_ARG2("socket", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size())); << CAF_ARG2("queue-size", packet_queue_.size()));
auto fetch_next_message = [&] { auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) { if (auto msg = manager.next_message()) {
......
...@@ -16,6 +16,7 @@ ...@@ -16,6 +16,7 @@
namespace caf::net { namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource.
template <class Buffer> template <class Buffer>
class producer_adapter final : public ref_counted, public async::producer { class producer_adapter final : public ref_counted, public async::producer {
public: public:
...@@ -127,8 +128,11 @@ private: ...@@ -127,8 +128,11 @@ private:
} }
void on_cancel() { void on_cancel() {
if (buf_) if (buf_) {
mgr_->mpx().shutdown_reading(mgr_); mgr_->mpx().shutdown_reading(mgr_);
buf_ = nullptr;
mgr_ = nullptr;
}
} }
auto strong_this() { auto strong_this() {
......
...@@ -155,7 +155,7 @@ public: ...@@ -155,7 +155,7 @@ public:
template <class ParentPtr> template <class ParentPtr>
bool handle_read_event(ParentPtr parent) { bool handle_read_event(ParentPtr parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", parent->handle().id));
auto fail = [this, parent](auto reason) { auto fail = [this, parent](auto reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason)); CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent->abort_reason(std::move(reason)); parent->abort_reason(std::move(reason));
...@@ -256,7 +256,7 @@ public: ...@@ -256,7 +256,7 @@ public:
template <class ParentPtr> template <class ParentPtr>
bool handle_write_event(ParentPtr parent) { bool handle_write_event(ParentPtr parent) {
CAF_LOG_TRACE(CAF_ARG2("handle", parent->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", parent->handle().id));
auto fail = [this, parent](sec reason) { auto fail = [this, parent](sec reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason)); CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
parent->abort_reason(reason); parent->abort_reason(reason);
......
...@@ -18,6 +18,7 @@ namespace caf::net { ...@@ -18,6 +18,7 @@ namespace caf::net {
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
void close(socket fd) { void close(socket fd) {
CAF_LOG_DEBUG("close" << CAF_ARG2("socket", fd.id));
closesocket(fd.id); closesocket(fd.id);
} }
...@@ -131,6 +132,7 @@ error nonblocking(socket x, bool new_value) { ...@@ -131,6 +132,7 @@ error nonblocking(socket x, bool new_value) {
#else // CAF_WINDOWS #else // CAF_WINDOWS
void close(socket fd) { void close(socket fd) {
CAF_LOG_DEBUG("close" << CAF_ARG2("socket", fd.id));
::close(fd.id); ::close(fd.id);
} }
......
...@@ -70,55 +70,73 @@ public: ...@@ -70,55 +70,73 @@ public:
using adapter_type = adapter_ptr::element_type; using adapter_type = adapter_ptr::element_type;
explicit app_t(resource_type input) : input_(std::move(input)) { explicit app_t(resource_type input) : input(std::move(input)) {
// nop // nop
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
error init(net::socket_manager* mgr, LowerLayerPtr, const settings&) { error init(net::socket_manager* mgr, LowerLayerPtr, const settings&) {
if (auto ptr = adapter_type::try_open(mgr, std::move(input_))) { if (auto ptr = adapter_type::try_open(mgr, std::move(input))) {
adapter_ = std::move(ptr); adapter = std::move(ptr);
return none; return none;
} else { } else {
FAIL("unable to open the resource"); FAIL("unable to open the resource");
} }
} }
template <class LowerLayerPtr>
struct send_helper {
app_t* thisptr;
LowerLayerPtr down;
bool on_next_called;
bool aborted;
void on_next(span<const int32_t> items) {
REQUIRE_EQ(items.size(), 1u);
auto val = items[0];
thisptr->written_values.emplace_back(val);
auto offset = thisptr->written_bytes.size();
binary_serializer sink{nullptr, thisptr->written_bytes};
if (!sink.apply(val))
FAIL("sink.apply failed: " << sink.get_error());
auto bytes = make_span(thisptr->written_bytes).subspan(offset);
down->begin_output();
auto& buf = down->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output();
}
void on_complete() {
// nop
}
void on_error(const error&) {
aborted = true;
}
};
template <class LowerLayerPtr> template <class LowerLayerPtr>
bool prepare_send(LowerLayerPtr down) { bool prepare_send(LowerLayerPtr down) {
bool run = !done_; if (done)
while (run && down->can_send_more()) { return true;
bool on_next_called = false; auto helper = send_helper<LowerLayerPtr>{this, down, false, false};
auto fin = adapter_->consume( while (down->can_send_more()) {
async::ignore_errors, 1, auto [ok, consumed] = adapter->pull(async::delay_errors, 1, helper);
[this, down, &on_next_called](span<const int32_t> items) { if (!ok) {
REQUIRE_EQ(items.size(), 1u);
auto val = items[0];
written_values_.emplace_back(val);
auto offset = written_bytes_.size();
binary_serializer sink{nullptr, written_bytes_};
if (!sink.apply(val))
FAIL("sink.apply failed: " << sink.get_error());
auto bytes = make_span(written_bytes_).subspan(offset);
down->begin_output();
auto& buf = down->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end());
down->end_output();
on_next_called = true;
});
if (fin) {
MESSAGE("adapter signaled end-of-buffer"); MESSAGE("adapter signaled end-of-buffer");
done_ = true; done = true;
break;
} else if (consumed == 0) {
break;
} }
run = !done_ && on_next_called;
} }
MESSAGE(written_bytes_.size() << " bytes written"); MESSAGE(written_bytes.size() << " bytes written");
return true; return true;
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
bool done_sending(LowerLayerPtr) { bool done_sending(LowerLayerPtr) {
return done_ || !adapter_->has_data(); return done || !adapter->has_data();
} }
template <class LowerLayerPtr> template <class LowerLayerPtr>
...@@ -136,20 +154,11 @@ public: ...@@ -136,20 +154,11 @@ public:
FAIL("app::consume called: unexpected data"); FAIL("app::consume called: unexpected data");
} }
auto& written_values() { bool done = false;
return written_values_; std::vector<int32_t> written_values;
} std::vector<byte> written_bytes;
adapter_ptr adapter;
auto& written_bytes() { resource_type input;
return written_bytes_;
}
private:
bool done_ = false;
std::vector<int32_t> written_values_;
std::vector<byte> written_bytes_;
adapter_ptr adapter_;
resource_type input_;
}; };
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
...@@ -199,10 +208,10 @@ SCENARIO("subscriber adapters wake up idle socket managers") { ...@@ -199,10 +208,10 @@ SCENARIO("subscriber adapters wake up idle socket managers") {
run(); run();
rd.read_some(); rd.read_some();
} }
CHECK_EQ(app.written_values(), std::vector<int32_t>(num_items, 42)); CHECK_EQ(app.written_values, std::vector<int32_t>(num_items, 42));
CHECK_EQ(app.written_bytes().size(), num_items * sizeof(int32_t)); CHECK_EQ(app.written_bytes.size(), num_items * sizeof(int32_t));
CHECK_EQ(rd.buf().size(), num_items * sizeof(int32_t)); CHECK_EQ(rd.buf().size(), num_items * sizeof(int32_t));
CHECK_EQ(app.written_bytes(), rd.buf()); CHECK_EQ(app.written_bytes, rd.buf());
} }
} }
} }
......
...@@ -174,10 +174,9 @@ SCENARIO("calling suspend_reading removes message apps temporarily") { ...@@ -174,10 +174,9 @@ SCENARIO("calling suspend_reading removes message apps temporarily") {
CHECK_EQ(state.inputs[1], "second"); CHECK_EQ(state.inputs[1], "second");
CHECK_EQ(state.inputs[2], "pause"); CHECK_EQ(state.inputs[2], "pause");
} }
THEN("users can resume it via register_reading ") { THEN("users can resume it via continue_reading ") {
mpx.register_reading(mgr); mgr->continue_reading();
CHECK_EQ(mgr->mask(), net::operation::read); CHECK_EQ(mgr->mask(), net::operation::read);
//mgr->register_reading();
while (mpx.num_socket_managers() > 1u) while (mpx.num_socket_managers() > 1u)
mpx.poll_once(true); mpx.poll_once(true);
if (CHECK_EQ(state.inputs.size(), 5u)) { if (CHECK_EQ(state.inputs.size(), 5u)) {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment