Commit a5befd70 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/781', close #781

parents fb08549d 098ae275
...@@ -107,8 +107,9 @@ public: ...@@ -107,8 +107,9 @@ public:
<< CAF_ARG(force_underfull)); << CAF_ARG(force_underfull));
if (pending()) if (pending())
return; return;
CAF_ASSERT(open_credit >= 0);
CAF_ASSERT(desired_batch_size > 0); CAF_ASSERT(desired_batch_size > 0);
CAF_ASSERT(cache.size() < std::numeric_limits<int32_t>::max()); CAF_ASSERT(cache.size() <= std::numeric_limits<int32_t>::max());
auto first = cache.begin(); auto first = cache.begin();
auto last = first + std::min(open_credit, auto last = first + std::min(open_credit,
static_cast<int32_t>(cache.size())); static_cast<int32_t>(cache.size()));
...@@ -137,15 +138,17 @@ public: ...@@ -137,15 +138,17 @@ public:
/// Returns whether this path is pending, i.e., didn't receive an `ack_open` /// Returns whether this path is pending, i.e., didn't receive an `ack_open`
/// yet. /// yet.
inline bool pending() const noexcept { bool pending() const noexcept {
return slots.receiver == invalid_stream_slot; return slots.receiver == invalid_stream_slot;
} }
/// Returns whether no pending ACKs exist. /// Returns whether no pending ACKs exist.
inline bool clean() const noexcept { bool clean() const noexcept {
return next_batch_id == next_ack_id; return next_batch_id == next_ack_id;
} }
void set_desired_batch_size(int32_t value) noexcept;
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Slot IDs for sender (self) and receiver (hdl). /// Slot IDs for sender (self) and receiver (hdl).
......
...@@ -120,14 +120,15 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items, ...@@ -120,14 +120,15 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
auto credit = std::max((x.max_throughput * 2) auto credit = std::max((x.max_throughput * 2)
- (assigned_credit + queued_items), - (assigned_credit + queued_items),
0); 0);
// Protect against overflow on `assigned_credit`.
auto max_new_credit = std::numeric_limits<int32_t>::max() - assigned_credit;
// The manager can restrict or adjust the amount of credit. // The manager can restrict or adjust the amount of credit.
credit = mgr->acquire_credit(this, credit); credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
if (credit == 0 && up_to_date()) { if (credit == 0 && up_to_date())
return; return;
}
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
if (credit > 0) if (credit > 0)
assigned_credit += credit; assigned_credit += credit;
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
CAF_LOG_DEBUG(CAF_ARG(credit) << CAF_ARG(desired_batch_size)); CAF_LOG_DEBUG(CAF_ARG(credit) << CAF_ARG(desired_batch_size));
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), make<upstream_msg::ack_batch>(slots.invert(),
......
...@@ -25,6 +25,13 @@ ...@@ -25,6 +25,13 @@
namespace caf { namespace caf {
namespace {
// TODO: consider making this parameter configurable
constexpr int32_t max_batch_size = 128 * 1024;
} // namespace <anonymous>
outbound_path::outbound_path(stream_slot sender_slot, outbound_path::outbound_path(stream_slot sender_slot,
strong_actor_ptr receiver_hdl) strong_actor_ptr receiver_hdl)
: slots(sender_slot, invalid_stream_slot), : slots(sender_slot, invalid_stream_slot),
...@@ -63,6 +70,7 @@ void outbound_path::emit_batch(local_actor* self, int32_t xs_size, message xs) { ...@@ -63,6 +70,7 @@ void outbound_path::emit_batch(local_actor* self, int32_t xs_size, message xs) {
CAF_ASSERT(xs_size <= std::numeric_limits<int32_t>::max()); CAF_ASSERT(xs_size <= std::numeric_limits<int32_t>::max());
CAF_ASSERT(open_credit >= xs_size); CAF_ASSERT(open_credit >= xs_size);
open_credit -= xs_size; open_credit -= xs_size;
CAF_ASSERT(open_credit >= 0);
auto bid = next_batch_id++; auto bid = next_batch_id++;
downstream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs), downstream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs),
bid}; bid};
...@@ -99,4 +107,11 @@ void outbound_path::emit_irregular_shutdown(local_actor* self, ...@@ -99,4 +107,11 @@ void outbound_path::emit_irregular_shutdown(local_actor* self,
std::move(reason))); std::move(reason)));
} }
void outbound_path::set_desired_batch_size(int32_t value) noexcept {
if (value == desired_batch_size)
return;
desired_batch_size = value < 0 || value > max_batch_size ? max_batch_size
: value;
}
} // namespace caf } // namespace caf
...@@ -82,7 +82,8 @@ bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { ...@@ -82,7 +82,8 @@ bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
} }
ptr->slots.receiver = slots.sender; ptr->slots.receiver = slots.sender;
ptr->open_credit = x.initial_demand; ptr->open_credit = x.initial_demand;
ptr->desired_batch_size = x.desired_batch_size; CAF_ASSERT(ptr->open_credit >= 0);
ptr->set_desired_batch_size(x.desired_batch_size);
--pending_handshakes_; --pending_handshakes_;
push(); push();
return true; return true;
...@@ -94,7 +95,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -94,7 +95,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
auto path = out().path(slots.receiver); auto path = out().path(slots.receiver);
if (path != nullptr) { if (path != nullptr) {
path->open_credit += x.new_capacity; path->open_credit += x.new_capacity;
path->desired_batch_size = x.desired_batch_size; CAF_ASSERT(path->open_credit >= 0);
path->set_desired_batch_size(x.desired_batch_size);
path->next_ack_id = x.acknowledged_id + 1; path->next_ack_id = x.acknowledged_id + 1;
// Gravefully remove path after receiving its final ACK. // Gravefully remove path after receiving its final ACK.
if (path->closing && out().clean(slots.receiver)) if (path->closing && out().clean(slots.receiver))
......
...@@ -116,7 +116,7 @@ public: ...@@ -116,7 +116,7 @@ public:
void add_path_to(entity& x, int32_t desired_batch_size) { void add_path_to(entity& x, int32_t desired_batch_size) {
auto ptr = mgr.out().add_path(next_slot++, x.ctrl()); auto ptr = mgr.out().add_path(next_slot++, x.ctrl());
CAF_REQUIRE(ptr != nullptr); CAF_REQUIRE(ptr != nullptr);
ptr->desired_batch_size = desired_batch_size; ptr->set_desired_batch_size(desired_batch_size);
ptr->slots.receiver = x.next_slot++; ptr->slots.receiver = x.next_slot++;
paths.emplace_back(ptr); paths.emplace_back(ptr);
} }
......
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