Commit ba47f3d1 authored by Dominik Charousset's avatar Dominik Charousset

WIP

parent 1d632b76
......@@ -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>;
......@@ -120,20 +123,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 +180,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
......
......@@ -80,21 +80,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("");
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 +174,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();
}
}
......
......@@ -132,6 +132,9 @@ TESTEE(stream_multiplexer) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->state.stage->add_inbound_path(in);
},
[=](close_atom, stream_slot slot) {
self->state.stage.out().close(slot);
},
};
}
......@@ -167,6 +170,32 @@ CAF_TEST(depth_3_pipeline_with_fork) {
self->send_exit(stg, exit_reason::kill);
}
CAF_TEST(depth_3_pipeline_with_fork) {
auto src = sys.spawn(file_reader, 100000u);
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("run until sink 1 received at least 50 items");
auto pred = [&] { return deref<sum_up_actor>(snk1).state.x < 1275; };
sched.run_dispatch_loop(pred, streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk2).state.x, 1275);
self->send_exit(stg, exit_reason::kill);
}
CAF_TEST(depth_3_pipeline_with_join) {
auto src1 = sys.spawn(file_reader, 50u);
auto src2 = sys.spawn(file_reader, 50u);
......
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