Skip to content

Commit

Permalink
Fix bug of broadcast stop on receiving the last FEC piece (#1040)
Browse files Browse the repository at this point in the history
  • Loading branch information
@EmelyanenkoK
EmelyanenkoK authored Jun 27, 2024
1 parent 2634e63 commit 11f39ac
Show file tree
Hide file tree
Showing 2 changed files with 34 additions and 21 deletions.
24 changes: 15 additions & 9 deletions overlay/overlay-fec-broadcast.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,6 @@ td::Status OverlayFecBroadcastPart::check_signature() {
}

td::Status OverlayFecBroadcastPart::run_checks() {

TRY_STATUS(check_time());
TRY_STATUS(check_duplicate());
TRY_STATUS(check_source());
Expand All @@ -94,14 +93,17 @@ void BroadcastFec::broadcast_checked(td::Result<td::Unit> R) {
overlay_->deliver_broadcast(get_source().compute_short_id(), data_.clone());
auto manager = overlay_->overlay_manager();
while (!parts_.empty()) {
distribute_part(parts_.begin()->first);
distribute_part(parts_.begin()->first);
}

is_checked_ = true;
}

// Do we need status here??
td::Status BroadcastFec::distribute_part(td::uint32 seqno) {
td::Status BroadcastFec::distribute_part(td::uint32 seqno) {
auto i = parts_.find(seqno);
if (i == parts_.end()) {
VLOG(OVERLAY_WARNING) << "not distibuting empty part " << seqno;
// should not get here
return td::Status::OK();
}
Expand Down Expand Up @@ -132,7 +134,6 @@ td::Status BroadcastFec::distribute_part(td::uint32 seqno) {
}

td::Status OverlayFecBroadcastPart::apply() {

if (!bcast_) {
bcast_ = overlay_->get_fec_broadcast(broadcast_hash_);
}
Expand Down Expand Up @@ -165,16 +166,20 @@ td::Status OverlayFecBroadcastPart::apply() {
return S;
}
} else {
if(untrusted_) {
if (untrusted_) {
auto P = td::PromiseCreator::lambda(
[id = broadcast_hash_, overlay_id = actor_id(overlay_)](td::Result<td::Unit> RR) mutable {
td::actor::send_closure(std::move(overlay_id), &OverlayImpl::broadcast_checked, id, std::move(RR));
});
[id = broadcast_hash_, overlay_id = actor_id(overlay_)](td::Result<td::Unit> RR) mutable {
td::actor::send_closure(std::move(overlay_id), &OverlayImpl::broadcast_checked, id, std::move(RR));
});
overlay_->check_broadcast(bcast_->get_source().compute_short_id(), R.move_as_ok(), std::move(P));
} else {
overlay_->deliver_broadcast(bcast_->get_source().compute_short_id(), R.move_as_ok());
}
}
} else {
bcast_->set_overlay(overlay_);
bcast_->set_src_peer_id(src_peer_id_);
TRY_STATUS(bcast_->add_part(seqno_, data_.clone(), export_serialized_short(), export_serialized()));
}
return td::Status::OK();
}
Expand Down Expand Up @@ -304,7 +309,8 @@ td::Status OverlayFecBroadcastPart::create_new(OverlayImpl *overlay, td::actor::

auto B = std::make_unique<OverlayFecBroadcastPart>(
broadcast_hash, part_hash, PublicKey{}, overlay->get_certificate(local_id), data_hash, size, flags,
part_data_hash, std::move(part), seqno, std::move(fec_type), date, td::BufferSlice{}, false, nullptr, overlay, adnl::AdnlNodeIdShort::zero());
part_data_hash, std::move(part), seqno, std::move(fec_type), date, td::BufferSlice{}, false, nullptr, overlay,
adnl::AdnlNodeIdShort::zero());
auto to_sign = B->to_sign();

auto P = td::PromiseCreator::lambda(
Expand Down
31 changes: 19 additions & 12 deletions overlay/overlay-fec-broadcast.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,15 +82,15 @@ class BroadcastFec : public td::ListNode {
}
}

td::Status add_part(td::uint32 seqno, td::BufferSlice data,
td::BufferSlice serialized_fec_part_short,
td::Status add_part(td::uint32 seqno, td::BufferSlice data, td::BufferSlice serialized_fec_part_short,
td::BufferSlice serialized_fec_part) {
CHECK(decoder_);
td::fec::Symbol s;
s.id = seqno;
s.data = std::move(data);
if (decoder_) {
td::fec::Symbol s;
s.id = seqno;
s.data = std::move(data);

decoder_->add_symbol(std::move(s));
decoder_->add_symbol(std::move(s));
}
parts_[seqno] = std::pair<td::BufferSlice, td::BufferSlice>(std::move(serialized_fec_part_short),
std::move(serialized_fec_part));

Expand Down Expand Up @@ -200,8 +200,13 @@ class BroadcastFec : public td::ListNode {

td::Status distribute_part(td::uint32 seqno);

bool is_checked() const {
return is_checked_;
}

private:
bool ready_ = false;
bool is_checked_ = false;

Overlay::BroadcastHash hash_;
Overlay::BroadcastDataHash data_hash_;
Expand Down Expand Up @@ -281,7 +286,7 @@ class OverlayFecBroadcastPart : public td::ListNode {
, signature_(std::move(signature))
, is_short_(is_short)
, bcast_(bcast)
, overlay_(overlay)
, overlay_(overlay)
, src_peer_id_(src_peer_id) {
}

Expand All @@ -300,7 +305,7 @@ class OverlayFecBroadcastPart : public td::ListNode {
signature_ = std::move(signature);
}
void update_overlay(OverlayImpl *overlay);

tl_object_ptr<ton_api::overlay_broadcastFec> export_tl();
tl_object_ptr<ton_api::overlay_broadcastFecShort> export_tl_short();
td::BufferSlice export_serialized();
Expand All @@ -310,14 +315,16 @@ class OverlayFecBroadcastPart : public td::ListNode {
td::Status run() {
TRY_STATUS(run_checks());
TRY_STATUS(apply());
if(!untrusted_) {
if (!untrusted_ || bcast_->is_checked()) {
TRY_STATUS(distribute());
}
return td::Status::OK();
}

static td::Status create(OverlayImpl *overlay, adnl::AdnlNodeIdShort src_peer_id, tl_object_ptr<ton_api::overlay_broadcastFec> broadcast);
static td::Status create(OverlayImpl *overlay, adnl::AdnlNodeIdShort src_peer_id, tl_object_ptr<ton_api::overlay_broadcastFecShort> broadcast);
static td::Status create(OverlayImpl *overlay, adnl::AdnlNodeIdShort src_peer_id,
tl_object_ptr<ton_api::overlay_broadcastFec> broadcast);
static td::Status create(OverlayImpl *overlay, adnl::AdnlNodeIdShort src_peer_id,
tl_object_ptr<ton_api::overlay_broadcastFecShort> broadcast);
static td::Status create_new(OverlayImpl *overlay, td::actor::ActorId<OverlayImpl> overlay_actor_id,
PublicKeyHash local_id, Overlay::BroadcastDataHash data_hash, td::uint32 size,
td::uint32 flags, td::BufferSlice part, td::uint32 seqno, fec::FecType fec_type,
Expand Down

0 comments on commit 11f39ac

Please sign in to comment.