Commit 4f917888 authored by Dominik Charousset's avatar Dominik Charousset

Allow stream managers to close individual slots

parent 58b281ec
......@@ -74,6 +74,12 @@ public:
return central_buf + max_path_buf;
}
size_t buffered(stream_slot slot) const noexcept override {
auto i = state_map_.find(slot);
return this->buf_.size()
+ (i != state_map_.end() ? i->second.buf.size() : 0u);
}
/// Sets the filter for `slot` to `filter`. Inserts a new element if `slot`
/// is a new path.
void set_filter(stream_slot slot, filter_type filter) {
......@@ -220,31 +226,30 @@ private:
return;
// Calculate the chunk size, i.e., how many more items we can put to our
// caches at the most.
struct get_credit {
inline size_t operator()(typename map_type::value_type& x) {
return static_cast<size_t>(x.second->open_credit);
}
auto not_closing = [&](typename map_type::value_type& x,
typename state_map_type::value_type&) {
return !x.second->closing;
};
struct get_cache_size {
inline size_t operator()(typename state_map_type::value_type& x) {
return x.second.buf.size();
}
};
auto f = [](size_t x, size_t credit, size_t cache_size) {
auto y = credit > cache_size ? credit - cache_size : 0u;
return x < y ? x : y;
// Returns the minimum of `interim` and `get_credit(x) - get_cache_size(y)`.
auto f = [&](size_t interim, typename map_type::value_type& x,
typename state_map_type::value_type& y) {
auto credit = static_cast<size_t>(x.second->open_credit);
auto cache_size = y.second.buf.size();
return std::min(interim, credit > cache_size ? credit - cache_size : 0u);
};
auto chunk_size = detail::zip_fold(
f, std::numeric_limits<size_t>::max(),
detail::make_container_view<get_credit>(this->paths_.container()),
detail::make_container_view<get_cache_size>(state_map_.container()));
auto chunk_size = detail::zip_fold_if(f, not_closing,
std::numeric_limits<size_t>::max(),
this->paths_.container(),
state_map_.container());
auto chunk = this->get_chunk(chunk_size);
if (chunk.empty()) {
auto g = [&](typename map_type::value_type& x,
typename state_map_type::value_type& y) {
x.second->emit_batches(this->self(), y.second.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
detail::zip_foreach_if(g, not_closing, this->paths_.container(),
state_map_.container());
} else {
auto g = [&](typename map_type::value_type& x,
typename state_map_type::value_type& y) {
......@@ -259,7 +264,8 @@ private:
}
x.second->emit_batches(this->self(), st.buf, force_underfull);
};
detail::zip_foreach(g, this->paths_.container(), state_map_.container());
detail::zip_foreach_if(g, not_closing, this->paths_.container(),
state_map_.container());
}
}
......
......@@ -25,7 +25,8 @@
namespace caf {
namespace detail {
/// Like `std::for_each`, but for multiple containers.
/// Like `std::for_each`, but for multiple containers and filters elements by
/// predicate.
/// @pre `x.size() <= y.size()` for each `y` in `xs`
template <class F, class Container, class... Containers>
void zip_foreach(F f, Container&& x, Containers&&... xs) {
......@@ -33,6 +34,15 @@ void zip_foreach(F f, Container&& x, Containers&&... xs) {
f(x[i], xs[i]...);
}
/// Like `std::for_each`, but for multiple containers.
/// @pre `x.size() <= y.size()` for each `y` in `xs`
template <class F, class Predicate, class Container, class... Containers>
void zip_foreach_if(F f, Predicate p, Container&& x, Containers&&... xs) {
for (size_t i = 0; i < x.size(); ++i)
if (p(x[i], xs[i]...))
f(x[i], xs[i]...);
}
/// Like `std::accumulate`, but for multiple containers.
/// @pre `x.size() <= y.size()` for each `y` in `xs`
template <class F, class T, class Container, class... Containers>
......@@ -42,6 +52,18 @@ T zip_fold(F f, T init, Container&& x, Containers&&... xs) {
return init;
}
/// Like `std::accumulate`, but for multiple containers and filters elements by
/// predicate.
/// @pre `x.size() <= y.size()` for each `y` in `xs`
template <class F, class Predicate, class T, class Container,
class... Containers>
T zip_fold_if(F f, Predicate p, T init, Container&& x, Containers&&... xs) {
for (size_t i = 0; i < x.size(); ++i)
if (p(x[i], xs[i]...))
init = f(init, x[i], xs[i]...);
return init;
}
/// Decorates a container of type `T` to appear as container of type `U`.
template <class F, class Container>
struct container_view {
......
......@@ -39,6 +39,9 @@ public:
/// Pointer to an outbound path.
using path_ptr = path_type*;
/// Pointer to an immutable outbound path.
using const_path_ptr = const path_type*;
/// Unique pointer to an outbound path.
using unique_path_ptr = std::unique_ptr<path_type>;
......@@ -95,6 +98,9 @@ public:
for_each_path_impl(g);
}
/// Returns all used slots.
std::vector<stream_slot> path_slots();
/// Checks whether `predicate` holds true for all paths.
template <class Predicate>
bool all_paths(Predicate predicate) const noexcept {
......@@ -120,20 +126,32 @@ public:
/// @returns The added path on success, `nullptr` otherwise.
path_ptr add_path(stream_slot slot, strong_actor_ptr target);
/// Removes a path from the manager and returns it.
/// Removes a path from the manager.
virtual bool remove_path(stream_slot slot, error reason,
bool silent) noexcept;
/// Returns the path associated to `slot` or `nullptr`.
virtual path_ptr path(stream_slot slot) noexcept;
/// Returns `true` if there is no data pending and no unacknowledged batch on
/// any path.
/// Returns the path associated to `slot` or `nullptr`.
const_path_ptr path(stream_slot slot) const noexcept;
/// Returns `true` if there is no data pending and all batches are
/// acknowledged batch on all paths.
bool clean() const noexcept;
/// Returns `true` if `slot` is unknown or if there is no data pending and
/// all batches are acknowledged on `slot`. The default implementation
/// returns `false` for all paths, even if `clean()` return `true`.
bool clean(stream_slot slot) const noexcept;
/// Removes all paths gracefully.
virtual void close();
/// Removes path `slot` gracefully by sending pending batches before removing
/// it. Effectively calls `path(slot)->closing = true`.
virtual void close(stream_slot slot);
/// Removes all paths with an error message.
virtual void abort(error reason);
......@@ -165,6 +183,9 @@ public:
/// Queries the size of the output buffer.
virtual size_t buffered() const noexcept;
/// Queries an estimate of the size of the output buffer for `slot`.
virtual size_t buffered(stream_slot slot) const noexcept;
/// Queries whether the manager cannot make any progress, because its buffer
/// is full and no more credit is available.
bool stalled() const noexcept;
......
......@@ -140,6 +140,11 @@ public:
return slots.receiver == invalid_stream_slot;
}
/// Returns whether no pending ACKs exist.
inline bool clean() const noexcept {
return next_batch_id == next_ack_id;
}
// -- member variables -------------------------------------------------------
/// Slot IDs for sender (self) and receiver (hdl).
......@@ -160,6 +165,13 @@ public:
/// ID of the first unacknowledged batch. Note that CAF uses accumulative
/// ACKs, i.e., receiving an ACK with a higher ID is not an error.
int64_t next_ack_id;
/// Stores whether an outbound path is marked for removal. The
/// `downstream_manger` no longer sends new batches to a closing path, but
/// buffered batches are still shipped. The owning `stream_manager` removes
/// the path when receiving an `upstream_msg::ack_batch` and no pending
/// batches for this path exist.
bool closing;
};
/// @relates outbound_path
......
......@@ -64,6 +64,15 @@ bool downstream_manager::terminal() const noexcept {
// -- path management ----------------------------------------------------------
std::vector<stream_slot> downstream_manager::path_slots() {
std::vector<stream_slot> xs;
xs.reserve(num_paths());
for_each_path([&](outbound_path& x) {
xs.emplace_back(x.slots.sender);
});
return xs;
}
size_t downstream_manager::num_paths() const noexcept {
return 0;
}
......@@ -80,21 +89,47 @@ bool downstream_manager::remove_path(stream_slot, error, bool) noexcept {
return false;
}
downstream_manager::const_path_ptr
downstream_manager::path(stream_slot slot) const noexcept {
// The `const` is restored when returning the pointer.
return const_cast<downstream_manager&>(*this).path(slot);
}
auto downstream_manager::path(stream_slot) noexcept -> path_ptr {
return nullptr;
}
bool downstream_manager::clean() const noexcept {
auto pred = [](const outbound_path& x) {
return x.next_batch_id == x.next_ack_id;
return x.clean();
};
return buffered() == 0 && all_paths(pred);
}
bool downstream_manager::clean(stream_slot slot) const noexcept {
auto ptr = path(slot);
return ptr != nullptr ? buffered(slot) == 0 && ptr->clean() : false;
}
void downstream_manager::close() {
CAF_LOG_TRACE("");
for_each_path([&](outbound_path& x) { about_to_erase(&x, false, nullptr); });
clear_paths();
if (clean()) {
for_each_path([&](outbound_path& x) { about_to_erase(&x, false, nullptr); });
clear_paths();
} else {
for_each_path([&](outbound_path& x) { x.closing = true; });
}
}
void downstream_manager::close(stream_slot slot) {
CAF_LOG_TRACE(CAF_ARG(slot));
if (clean(slot))
remove_path(slot, none, false);
else {
auto ptr = path(slot);
if (ptr != nullptr)
ptr->closing = true;
}
}
void downstream_manager::abort(error reason) {
......@@ -148,6 +183,10 @@ size_t downstream_manager::buffered() const noexcept {
return 0;
}
size_t downstream_manager::buffered(stream_slot) const noexcept {
return 0;
}
bool downstream_manager::stalled() const noexcept {
auto no_credit = [](const outbound_path& x) {
return x.open_credit == 0;
......
......@@ -32,7 +32,8 @@ outbound_path::outbound_path(stream_slot sender_slot,
next_batch_id(1),
open_credit(0),
desired_batch_size(50),
next_ack_id(1) {
next_ack_id(1),
closing(false) {
// nop
}
......
......@@ -92,6 +92,9 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
path->open_credit += x.new_capacity;
path->desired_batch_size = x.desired_batch_size;
path->next_ack_id = x.acknowledged_id + 1;
// Gravefully remove path after receiving its final ACK.
if (path->closing && out().clean(slots.receiver))
out().remove_path(slots.receiver, none, false);
push();
}
}
......
......@@ -34,10 +34,18 @@ using namespace caf;
namespace {
/// Returns the sum of natural numbers up until `n`, i.e., 1 + 2 + ... + n.
int sum(int n) {
return (n * (n + 1)) / 2;
}
TESTEE_SETUP();
TESTEE_STATE(file_reader) {
std::vector<int> buf;
};
VARARGS_TESTEE(file_reader, size_t buf_size) {
using buf = std::deque<int>;
return {
[=](string& fname) -> output_stream<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
......@@ -46,12 +54,14 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](buf& xs) {
[=](unit_t&) {
auto& xs = self->state.buf;
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[](buf& xs, downstream<int>& out, size_t num) {
[=](unit_t&, downstream<int>& out, size_t num) {
auto& xs = self->state.buf;
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
......@@ -59,8 +69,8 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[=](const buf& xs) {
if (xs.empty()) {
[=](const unit_t&) {
if (self->state.buf.empty()) {
CAF_MESSAGE(self->name() << " is done");
return true;
}
......@@ -132,6 +142,10 @@ TESTEE(stream_multiplexer) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->state.stage->add_inbound_path(in);
},
[=](close_atom, int sink_index) {
auto& out = self->state.stage->out();
out.close(out.path_slots().at(static_cast<size_t>(sink_index)));
},
};
}
......@@ -190,4 +204,44 @@ CAF_TEST(depth_3_pipeline_with_join) {
self->send_exit(stg, exit_reason::kill);
}
CAF_TEST(closing_downstreams_before_end_of_stream) {
auto src = sys.spawn(file_reader, 10000u);
auto stg = sys.spawn(stream_multiplexer);
auto snk1 = sys.spawn(sum_up);
auto snk2 = sys.spawn(sum_up);
auto& st = deref<stream_multiplexer_actor>(stg).state;
CAF_MESSAGE("connect sinks to the stage (fork)");
self->send(snk1, join_atom::value, stg);
self->send(snk2, join_atom::value, stg);
sched.run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_MESSAGE("connect source to the stage (fork)");
self->send(stg * src, "numbers.txt");
sched.run();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
CAF_MESSAGE("do a single round of credit");
sched.clock().current_time += streaming_cycle;
sched.dispatch();
sched.run();
CAF_MESSAGE("make sure the stream isn't done yet");
CAF_REQUIRE(!deref<file_reader_actor>(src).state.buf.empty());
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
CAF_MESSAGE("get the next not-yet-buffered integer");
auto next_pending = deref<file_reader_actor>(src).state.buf.front();
CAF_REQUIRE_GREATER(next_pending, 0);
auto sink1_result = sum(next_pending - 1);
CAF_MESSAGE("gracefully close sink 1, next pending: " << next_pending);
self->send(stg, close_atom::value, 0);
expect((atom_value, int), from(self).to(stg));
CAF_MESSAGE("ship remaining elements");
run_exhaustively();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_LESS(deref<sum_up_actor>(snk1).state.x, sink1_result);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, sum(10000));
self->send_exit(stg, exit_reason::kill);
}
CAF_TEST_FIXTURE_SCOPE_END()
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