gh-738: Use status codes in FEC block codecs

roc-streaming · Jul 26, 2024 · 5cf7355 · 5cf7355
1 parent 363a3d5
commit 5cf7355
Show file tree

Hide file tree

Showing 12 changed files with 138 additions and 58 deletions.
diff --git a/src/internal_modules/roc_fec/block_reader.cpp b/src/internal_modules/roc_fec/block_reader.cpp
@@ -80,6 +80,11 @@ status::StatusCode BlockReader::read(packet::PacketPtr& pp, packet::PacketReadMo
 
     const status::StatusCode code = read_(pp, mode);
 
+    if (code != status::StatusOK && code != status::StatusDrain) {
+        pp = NULL;
+        return code;
+    }
+
     if (!alive_) {
         pp = NULL;
         return status::StatusAbort;
@@ -88,10 +93,8 @@ status::StatusCode BlockReader::read(packet::PacketPtr& pp, packet::PacketReadMo
     if (code == status::StatusOK && mode == packet::ModeFetch) {
         n_packets_++;
     }
-
     return code;
 }
-
 status::StatusCode BlockReader::read_(packet::PacketPtr& pp,
                                       packet::PacketReadMode mode) {
     const status::StatusCode code = fetch_all_packets_();
@@ -159,8 +162,12 @@ status::StatusCode BlockReader::get_next_packet_(packet::PacketPtr& result_pkt,
         if (pkt) {
             next_index = head_index_ + 1;
         } else {
-            // Try repairing as much as possible and store in block.
-            try_repair_();
+            // try repairing as much as possible and store in block
+            const status::StatusCode status_code = try_repair_();
+            if (status_code != status::StatusOK) {
+                roc_panic_if(status_code == status::StatusDrain);
+                return status_code;
+            }
 
             // Find first present packet in block, starting from head.
             for (next_index = head_index_; next_index < source_block_.size();
@@ -229,24 +236,26 @@ void BlockReader::next_block_() {
     fill_block_();
 }
 
-void BlockReader::try_repair_() {
-    if (!can_repair_) {
-        return;
-    }
+bool BlockReader::is_block_resized_() const {
+    return source_block_resized_ && repair_block_resized_ && payload_resized_;
+}
 
-    if (!source_block_resized_ || !repair_block_resized_ || !payload_resized_) {
-        return;
+status::StatusCode BlockReader::try_repair_() {
+    if (!can_repair_ || !is_block_resized_()) {
+        return status::StatusOK;
     }
 
-    if (!block_decoder_.begin_block(source_block_.size(), repair_block_.size(),
-                                    payload_size_)) {
+    const status::StatusCode status_code = block_decoder_.begin_block(
+        source_block_.size(), repair_block_.size(), payload_size_);
+
+    if (status_code != status::StatusOK) {
         roc_log(LogDebug,
                 "fec block reader: can't begin decoder block, shutting down:"
                 " sbl=%lu rbl=%lu payload_size=%lu",
                 (unsigned long)source_block_.size(), (unsigned long)repair_block_.size(),
                 (unsigned long)payload_size_);
         alive_ = false;
-        return;
+        return status_code;
     }
 
     for (size_t n = 0; n < source_block_.size(); n++) {
@@ -284,6 +293,7 @@ void BlockReader::try_repair_() {
 
     block_decoder_.end_block();
     can_repair_ = false;
+    return status::StatusOK;
 }
 
 packet::PacketPtr

diff --git a/src/internal_modules/roc_fec/block_reader.h b/src/internal_modules/roc_fec/block_reader.h
@@ -96,7 +96,7 @@ class BlockReader : public packet::IReader, public core::NonCopyable<> {
                                         packet::PacketReadMode mode);
 
     void next_block_();
-    void try_repair_();
+    status::StatusCode try_repair_();
 
     packet::PacketPtr parse_repaired_packet_(const core::Slice<uint8_t>& buffer);
 
@@ -128,6 +128,8 @@ class BlockReader : public packet::IReader, public core::NonCopyable<> {
 
     void update_block_duration_(const packet::PacketPtr& curr_block_pkt);
 
+    bool is_block_resized_() const;
+
     IBlockDecoder& block_decoder_;
 
     packet::IReader& source_reader_;

diff --git a/src/internal_modules/roc_fec/block_writer.cpp b/src/internal_modules/roc_fec/block_writer.cpp
@@ -120,8 +120,7 @@ status::StatusCode BlockWriter::write(const packet::PacketPtr& pp) {
     roc_panic_if(!pp);
 
     if (!alive_) {
-        // TODO(gh-183): return StatusDead
-        return status::StatusOK;
+        return status::StatusAbort;
     }
 
     validate_fec_packet_(pp);
@@ -131,9 +130,9 @@ status::StatusCode BlockWriter::write(const packet::PacketPtr& pp) {
     }
 
     if (cur_packet_ == 0) {
-        if (!begin_block_(pp)) {
-            // TODO(gh-183): return status
-            return status::StatusOK;
+        const status::StatusCode status_code = begin_block_(pp);
+        if (status_code != status::StatusOK) {
+            return status_code;
         }
     }
 
@@ -156,27 +155,32 @@ status::StatusCode BlockWriter::write(const packet::PacketPtr& pp) {
     return status::StatusOK;
 }
 
-bool BlockWriter::begin_block_(const packet::PacketPtr& pp) {
+status::StatusCode BlockWriter::begin_block_(const packet::PacketPtr& pp) {
     update_block_duration_(pp);
 
     if (!apply_sizes_(next_sblen_, next_rblen_, pp->fec()->payload.size())) {
-        return false;
+        roc_log(LogError,
+                "fec block writer: apply_sizes in begin_block_ failed with StatusNoMem");
+        return status::StatusNoMem;
     }
 
     roc_log(LogTrace,
             "fec block writer: begin block: sbn=%lu sblen=%lu rblen=%lu payload_size=%lu",
             (unsigned long)cur_sbn_, (unsigned long)cur_sblen_, (unsigned long)cur_rblen_,
             (unsigned long)cur_payload_size_);
 
-    if (!block_encoder_.begin_block(cur_sblen_, cur_rblen_, cur_payload_size_)) {
+    const status::StatusCode status_code =
+        block_encoder_.begin_block(cur_sblen_, cur_rblen_, cur_payload_size_);
+
+    if (status_code != status::StatusOK) {
         roc_log(LogError,
                 "fec block writer: can't begin encoder block, shutting down:"
                 " sblen=%lu rblen=%lu",
                 (unsigned long)cur_sblen_, (unsigned long)cur_rblen_);
-        return (alive_ = false);
+        alive_ = false;
     }
 
-    return true;
+    return status_code;
 }
 
 void BlockWriter::end_block_() {

diff --git a/src/internal_modules/roc_fec/block_writer.h b/src/internal_modules/roc_fec/block_writer.h
@@ -89,7 +89,7 @@ class BlockWriter : public packet::IWriter, public core::NonCopyable<> {
     virtual ROC_ATTR_NODISCARD status::StatusCode write(const packet::PacketPtr&);
 
 private:
-    bool begin_block_(const packet::PacketPtr& pp);
+    status::StatusCode begin_block_(const packet::PacketPtr& pp);
     void end_block_();
     void next_block_();
 

diff --git a/src/internal_modules/roc_fec/iblock_decoder.h b/src/internal_modules/roc_fec/iblock_decoder.h
@@ -34,7 +34,10 @@ class IBlockDecoder {
     //! @remarks
     //!  Performs an initial setup for a block. Should be called before
     //!  any operations for the block.
-    virtual bool begin_block(size_t sblen, size_t rblen, size_t payload_size) = 0;
+    //! @returns status::StatusOK on success, or a specific error code on failure (e.g.,
+    //! status::StatusNoMem if memory allocation fails).
+    virtual ROC_ATTR_NODISCARD status::StatusCode
+    begin_block(size_t sblen, size_t rblen, size_t payload_size) = 0;
 
     //! Store source or repair packet buffer for current block.
     //! @pre

diff --git a/src/internal_modules/roc_fec/iblock_encoder.h b/src/internal_modules/roc_fec/iblock_encoder.h
@@ -35,9 +35,12 @@ class IBlockEncoder {
 
     //! Start block.
     //! @remarks
-    //!  Performs an initial setup for a block. Should be called before
-    //!  any operations for the block.
-    virtual bool begin_block(size_t sblen, size_t rblen, size_t payload_size) = 0;
+    //! Performs an initial setup for a block. Should be called before any operations for
+    //! the block.
+    //! @returns status::StatusOK on success, or a specific error code on failure (e.g.,
+    //! status::StatusNoMem if memory allocation fails).
+    virtual ROC_ATTR_NODISCARD status::StatusCode
+    begin_block(size_t sblen, size_t rblen, size_t payload_size) = 0;
 
     //! Store source or repair packet buffer for current block.
     //! @pre

diff --git a/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_decoder.cpp b/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_decoder.cpp
@@ -82,11 +82,16 @@ size_t OpenfecDecoder::max_block_length() const {
     return max_block_length_;
 }
 
-bool OpenfecDecoder::begin_block(size_t sblen, size_t rblen, size_t payload_size) {
+status::StatusCode
+OpenfecDecoder::begin_block(size_t sblen, size_t rblen, size_t payload_size) {
     roc_panic_if(init_status_ != status::StatusOK);
 
     if (!resize_tabs_(sblen + rblen)) {
-        return false;
+        roc_log(
+            LogError,
+            "openfec decoder: failed to resize tabs in begin_block, sblen=%lu, rblen=%lu",
+            (unsigned long)sblen, (unsigned long)rblen);
+        return status::StatusNoMem;
     }
 
     sblen_ = sblen;
@@ -97,7 +102,7 @@ bool OpenfecDecoder::begin_block(size_t sblen, size_t rblen, size_t payload_size
     update_session_params_(sblen, rblen, payload_size);
     reset_session_();
 
-    return true;
+    return status::StatusOK;
 }
 
 void OpenfecDecoder::set_buffer(size_t index, const core::Slice<uint8_t>& buffer) {

diff --git a/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_decoder.h b/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_decoder.h
@@ -53,7 +53,8 @@ class OpenfecDecoder : public IBlockDecoder, public core::NonCopyable<> {
     virtual size_t max_block_length() const;
 
     //! Start block.
-    virtual bool begin_block(size_t sblen, size_t rblen, size_t payload_size);
+    virtual ROC_ATTR_NODISCARD status::StatusCode
+    begin_block(size_t sblen, size_t rblen, size_t payload_size);
 
     //! Store source or repair packet buffer for current block.
     virtual void set_buffer(size_t index, const core::Slice<uint8_t>& buffer);

diff --git a/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_encoder.cpp b/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_encoder.cpp
@@ -75,15 +75,20 @@ size_t OpenfecEncoder::buffer_alignment() const {
     return Alignment;
 }
 
-bool OpenfecEncoder::begin_block(size_t sblen, size_t rblen, size_t payload_size) {
+status::StatusCode
+OpenfecEncoder::begin_block(size_t sblen, size_t rblen, size_t payload_size) {
     roc_panic_if(init_status_ != status::StatusOK);
 
     if (sblen_ == sblen && rblen_ == rblen && payload_size_ == payload_size) {
-        return true;
+        return status::StatusOK;
     }
 
     if (!resize_tabs_(sblen + rblen)) {
-        return false;
+        roc_log(
+            LogError,
+            "openfec encoder: failed to resize tabs in begin_block, sblen=%lu, rblen=%lu",
+            (unsigned long)sblen, (unsigned long)rblen);
+        return status::StatusNoMem;
     }
 
     sblen_ = sblen;
@@ -93,7 +98,7 @@ bool OpenfecEncoder::begin_block(size_t sblen, size_t rblen, size_t payload_size
     update_session_params_(sblen, rblen, payload_size);
     reset_session_();
 
-    return true;
+    return status::StatusOK;
 }
 
 void OpenfecEncoder::set_buffer(size_t index, const core::Slice<uint8_t>& buffer) {

diff --git a/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_encoder.h b/src/internal_modules/roc_fec/target_openfec/roc_fec/openfec_encoder.h
@@ -56,7 +56,8 @@ class OpenfecEncoder : public IBlockEncoder, public core::NonCopyable<> {
     virtual size_t buffer_alignment() const;
 
     //! Start block.
-    virtual bool begin_block(size_t sblen, size_t rblen, size_t payload_size);
+    virtual ROC_ATTR_NODISCARD status::StatusCode
+    begin_block(size_t sblen, size_t rblen, size_t payload_size);
 
     //! Store packet data for current block.
     virtual void set_buffer(size_t index, const core::Slice<uint8_t>& buffer);