Commit 098ae275 authored by Dominik Charousset's avatar Dominik Charousset

Protect against credit overflow and huge batches

parent cc1ac1f4
......@@ -107,8 +107,9 @@ public:
<< CAF_ARG(force_underfull));
if (pending())
return;
CAF_ASSERT(open_credit >= 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 last = first + std::min(open_credit,
static_cast<int32_t>(cache.size()));
......@@ -137,15 +138,17 @@ public:
/// Returns whether this path is pending, i.e., didn't receive an `ack_open`
/// yet.
inline bool pending() const noexcept {
bool pending() const noexcept {
return slots.receiver == invalid_stream_slot;
}
/// Returns whether no pending ACKs exist.
inline bool clean() const noexcept {
bool clean() const noexcept {
return next_batch_id == next_ack_id;
}
void set_desired_batch_size(int32_t value) noexcept;
// -- member variables -------------------------------------------------------
/// 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,
auto credit = std::max((x.max_throughput * 2)
- (assigned_credit + queued_items),
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.
credit = mgr->acquire_credit(this, credit);
if (credit == 0 && up_to_date()) {
credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
if (credit == 0 && up_to_date())
return;
}
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
if (credit > 0)
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));
unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(),
......
......@@ -25,6 +25,13 @@
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,
strong_actor_ptr receiver_hdl)
: slots(sender_slot, invalid_stream_slot),
......@@ -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(open_credit >= xs_size);
open_credit -= xs_size;
CAF_ASSERT(open_credit >= 0);
auto bid = next_batch_id++;
downstream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs),
bid};
......@@ -99,4 +107,11 @@ void outbound_path::emit_irregular_shutdown(local_actor* self,
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
......@@ -82,7 +82,8 @@ bool stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
}
ptr->slots.receiver = slots.sender;
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_;
push();
return true;
......@@ -94,7 +95,8 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
auto path = out().path(slots.receiver);
if (path != nullptr) {
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;
// Gravefully remove path after receiving its final ACK.
if (path->closing && out().clean(slots.receiver))
......
......@@ -116,7 +116,7 @@ public:
void add_path_to(entity& x, int32_t desired_batch_size) {
auto ptr = mgr.out().add_path(next_slot++, x.ctrl());
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++;
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