Slot provides Control endpoints ts

baranovmv · baranovmv · commit 5e8528929835 · 2024-12-14T13:47:29.000+01:00
So that control endpoints on receiver and sender
could mark unmarked packets with a timestamp. By
doing so we avoid necessity of mocking core::timestamp.

This note relates to the pipeline test mostly.
diff --git a/src/internal_modules/roc_pipeline/receiver_endpoint.cpp b/src/internal_modules/roc_pipeline/receiver_endpoint.cpp
@@ -199,7 +199,7 @@ packet::IWriter& ReceiverEndpoint::inbound_writer() {
     return *this;
 }
 
-status::StatusCode ReceiverEndpoint::pull_packets() {
+status::StatusCode ReceiverEndpoint::pull_packets(core::nanoseconds_t current_time) {
     roc_panic_if(init_status_ != status::StatusOK);
 
     roc_panic_if(!parser_);
@@ -209,7 +209,7 @@ status::StatusCode ReceiverEndpoint::pull_packets() {
     // queue were added in a very short time or are being added currently. It's
     // acceptable to consider such packets late and pull them next time.
     while (packet::PacketPtr packet = inbound_queue_.try_pop_front_exclusive()) {
-        const status::StatusCode code = handle_packet_(packet);
+        const status::StatusCode code = handle_packet_(packet, current_time);
         state_tracker_.unregister_packet();
 
         if (code != status::StatusOK) {
@@ -220,7 +220,13 @@ status::StatusCode ReceiverEndpoint::pull_packets() {
     return status::StatusOK;
 }
 
-status::StatusCode ReceiverEndpoint::handle_packet_(const packet::PacketPtr& packet) {
+status::StatusCode ReceiverEndpoint::handle_packet_(const packet::PacketPtr& packet,
+                                                    core::nanoseconds_t current_time) {
+    // Apparently the packet is not from network, set its TS manually.
+    if (packet->udp() && packet->udp()->receive_timestamp == 0 && current_time != 0) {
+        packet->udp()->receive_timestamp = current_time;
+    }
+
     if (!parser_->parse(*packet, packet->buffer())) {
         roc_log(LogDebug, "receiver endpoint: dropping bad packet: can't parse");
         return status::StatusOK;
diff --git a/src/internal_modules/roc_pipeline/receiver_endpoint.h b/src/internal_modules/roc_pipeline/receiver_endpoint.h
@@ -93,12 +93,13 @@ class ReceiverEndpoint : public core::RefCounted<ReceiverEndpoint, core::ArenaAl
     //!  Packets are written to inbound_writer() from network thread.
     //!  They don't appear in pipeline immediately. Instead, pipeline thread
     //!  should periodically call pull_packets() to make them available.
-    ROC_ATTR_NODISCARD status::StatusCode pull_packets();
+    ROC_ATTR_NODISCARD status::StatusCode pull_packets(core::nanoseconds_t current_time);
 
 private:
     virtual ROC_ATTR_NODISCARD status::StatusCode write(const packet::PacketPtr& packet);
 
-    status::StatusCode handle_packet_(const packet::PacketPtr& packet);
+    status::StatusCode handle_packet_(const packet::PacketPtr& packet,
+                                      core::nanoseconds_t current_time);
 
     const address::Protocol proto_;
 
diff --git a/src/internal_modules/roc_pipeline/receiver_slot.cpp b/src/internal_modules/roc_pipeline/receiver_slot.cpp
@@ -84,19 +84,19 @@ status::StatusCode ReceiverSlot::refresh(core::nanoseconds_t current_time,
     status::StatusCode code = status::NoStatus;
 
     if (source_endpoint_) {
-        if ((code = source_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = source_endpoint_->pull_packets(0)) != status::StatusOK) {
             return code;
         }
     }
 
     if (repair_endpoint_) {
-        if ((code = repair_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = repair_endpoint_->pull_packets(0)) != status::StatusOK) {
             return code;
         }
     }
 
     if (control_endpoint_) {
-        if ((code = control_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = control_endpoint_->pull_packets(current_time)) != status::StatusOK) {
             return code;
         }
     }
diff --git a/src/internal_modules/roc_pipeline/sender_endpoint.cpp b/src/internal_modules/roc_pipeline/sender_endpoint.cpp
@@ -184,7 +184,7 @@ packet::IWriter* SenderEndpoint::inbound_writer() {
     return this;
 }
 
-status::StatusCode SenderEndpoint::pull_packets() {
+status::StatusCode SenderEndpoint::pull_packets(core::nanoseconds_t current_time) {
     roc_panic_if(init_status_ != status::StatusOK);
 
     if (!parser_) {
@@ -197,7 +197,7 @@ status::StatusCode SenderEndpoint::pull_packets() {
     // queue were added in a very short time or are being added currently. It's
     // acceptable to consider such packets late and pull them next time.
     while (packet::PacketPtr packet = inbound_queue_.try_pop_front_exclusive()) {
-        const status::StatusCode code = handle_packet_(packet);
+        const status::StatusCode code = handle_packet_(packet, current_time);
         state_tracker_.unregister_packet();
 
         if (code != status::StatusOK) {
@@ -208,7 +208,13 @@ status::StatusCode SenderEndpoint::pull_packets() {
     return status::StatusOK;
 }
 
-status::StatusCode SenderEndpoint::handle_packet_(const packet::PacketPtr& packet) {
+status::StatusCode SenderEndpoint::handle_packet_(const packet::PacketPtr& packet,
+                                                  core::nanoseconds_t current_time) {
+    // Apparently the packet is not from network, set it's TS manually.
+    if (packet->udp() && packet->udp()->receive_timestamp == 0 && current_time != 0) {
+        packet->udp()->receive_timestamp = current_time;
+    }
+
     if (!parser_->parse(*packet, packet->buffer())) {
         roc_log(LogDebug, "sender endpoint: dropping bad packet: can't parse");
         return status::StatusOK;
diff --git a/src/internal_modules/roc_pipeline/sender_endpoint.h b/src/internal_modules/roc_pipeline/sender_endpoint.h
@@ -91,12 +91,13 @@ class SenderEndpoint : public core::NonCopyable<>, private packet::IWriter {
     //!  Packets are written to inbound_writer() from network thread.
     //!  They don't appear in pipeline immediately. Instead, pipeline thread
     //!  should periodically call pull_packets() to make them available.
-    ROC_ATTR_NODISCARD status::StatusCode pull_packets();
+    ROC_ATTR_NODISCARD status::StatusCode pull_packets(core::nanoseconds_t current_time);
 
 private:
     virtual ROC_ATTR_NODISCARD status::StatusCode write(const packet::PacketPtr& packet);
 
-    status::StatusCode handle_packet_(const packet::PacketPtr& packet);
+    status::StatusCode handle_packet_(const packet::PacketPtr& packet,
+                                      core::nanoseconds_t current_time);
 
     const address::Protocol proto_;
 
diff --git a/src/internal_modules/roc_pipeline/sender_slot.cpp b/src/internal_modules/roc_pipeline/sender_slot.cpp
@@ -142,19 +142,19 @@ status::StatusCode SenderSlot::refresh(core::nanoseconds_t current_time,
     status::StatusCode code = status::NoStatus;
 
     if (source_endpoint_) {
-        if ((code = source_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = source_endpoint_->pull_packets(0)) != status::StatusOK) {
             return code;
         }
     }
 
     if (repair_endpoint_) {
-        if ((code = repair_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = repair_endpoint_->pull_packets(0)) != status::StatusOK) {
             return code;
         }
     }
 
     if (control_endpoint_) {
-        if ((code = control_endpoint_->pull_packets()) != status::StatusOK) {
+        if ((code = control_endpoint_->pull_packets(current_time)) != status::StatusOK) {
             return code;
         }
     }
diff --git a/src/internal_modules/roc_rtcp/communicator.cpp b/src/internal_modules/roc_rtcp/communicator.cpp
@@ -52,8 +52,7 @@ Communicator::Communicator(const Config& config,
     , processed_packet_count_(0)
     , generated_packet_count_(0)
     , log_limiter_(LogInterval)
-    , init_status_(status::NoStatus)
-    , dumper_(dumper) {
+    , init_status_(status::NoStatus) {
     if ((init_status_ = reporter_.init_status()) != status::StatusOK) {
         return;
     }
diff --git a/src/internal_modules/roc_rtcp/communicator.h b/src/internal_modules/roc_rtcp/communicator.h
@@ -170,8 +170,6 @@ class Communicator : public core::NonCopyable<> {
     core::RateLimiter log_limiter_;
 
     status::StatusCode init_status_;
-
-    dbgio::CsvDumper* dumper_;
 };
 
 } // namespace rtcp
diff --git a/src/tests/roc_pipeline/test_helpers/control_writer.h b/src/tests/roc_pipeline/test_helpers/control_writer.h
@@ -180,7 +180,6 @@ class ControlWriter : public core::NonCopyable<> {
 
         pp->udp()->src_addr = src_addr_;
         pp->udp()->dst_addr = dst_addr_;
-        pp->udp()->receive_timestamp = core::timestamp(core::ClockUnix);
 
         pp->set_buffer(buffer);
 
diff --git a/src/tests/roc_pipeline/test_loopback_sink_2_source.cpp b/src/tests/roc_pipeline/test_loopback_sink_2_source.cpp
@@ -179,7 +179,6 @@ class PacketProxy : core::NonCopyable<> {
                     continue;
                 }
                 print_packet_(pp);
-                set_ts_(pp);
                 CHECK(source_writer_);
                 LONGS_EQUAL(status::StatusOK, source_writer_->write(copy_packet_(pp)));
                 n_source_++;
@@ -188,13 +187,11 @@ class PacketProxy : core::NonCopyable<> {
                     continue;
                 }
                 print_packet_(pp);
-                set_ts_(pp);
                 CHECK(repair_writer_);
                 LONGS_EQUAL(status::StatusOK, repair_writer_->write(copy_packet_(pp)));
                 n_repair_++;
             } else if (pp->flags() & packet::Packet::FlagControl) {
                 print_packet_(pp);
-                set_ts_(pp);
                 CHECK(control_writer_);
                 LONGS_EQUAL(status::StatusOK, control_writer_->write(copy_packet_(pp)));
                 n_control_++;
@@ -230,12 +227,6 @@ class PacketProxy : core::NonCopyable<> {
         }
     }
 
-    void set_ts_(packet::PacketPtr& pp) {
-        if (pp->udp()) {
-            pp->udp()->receive_timestamp = core::timestamp(core::ClockUnix);
-        }
-    }
-
     packet::PacketFactory& packet_factory_;
 
     address::SocketAddr proxy_addr_;

Original file line number	Diff line number	Diff line change
`@@ -84,19 +84,19 @@ status::StatusCode ReceiverSlot::refresh(core::nanoseconds_t current_time,`
`84`	`84`	`status::StatusCode code = status::NoStatus;`
`85`	`85`
`86`	`86`	`if (source_endpoint_) {`
`87`		`- if ((code = source_endpoint_->pull_packets()) != status::StatusOK) {`
	`87`	`+ if ((code = source_endpoint_->pull_packets(0)) != status::StatusOK) {`
`88`	`88`	`return code;`
`89`	`89`	`}`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`if (repair_endpoint_) {`
`93`		`- if ((code = repair_endpoint_->pull_packets()) != status::StatusOK) {`
	`93`	`+ if ((code = repair_endpoint_->pull_packets(0)) != status::StatusOK) {`
`94`	`94`	`return code;`
`95`	`95`	`}`
`96`	`96`	`}`
`97`	`97`
`98`	`98`	`if (control_endpoint_) {`
`99`		`- if ((code = control_endpoint_->pull_packets()) != status::StatusOK) {`
	`99`	`+ if ((code = control_endpoint_->pull_packets(current_time)) != status::StatusOK) {`
`100`	`100`	`return code;`
`101`	`101`	`}`
`102`	`102`	`}`
Original file line number	Diff line number	Diff line change
`@@ -142,19 +142,19 @@ status::StatusCode SenderSlot::refresh(core::nanoseconds_t current_time,`
`142`	`142`	`status::StatusCode code = status::NoStatus;`
`143`	`143`
`144`	`144`	`if (source_endpoint_) {`
`145`		`- if ((code = source_endpoint_->pull_packets()) != status::StatusOK) {`
	`145`	`+ if ((code = source_endpoint_->pull_packets(0)) != status::StatusOK) {`
`146`	`146`	`return code;`
`147`	`147`	`}`
`148`	`148`	`}`
`149`	`149`
`150`	`150`	`if (repair_endpoint_) {`
`151`		`- if ((code = repair_endpoint_->pull_packets()) != status::StatusOK) {`
	`151`	`+ if ((code = repair_endpoint_->pull_packets(0)) != status::StatusOK) {`
`152`	`152`	`return code;`
`153`	`153`	`}`
`154`	`154`	`}`
`155`	`155`
`156`	`156`	`if (control_endpoint_) {`
`157`		`- if ((code = control_endpoint_->pull_packets()) != status::StatusOK) {`
	`157`	`+ if ((code = control_endpoint_->pull_packets(current_time)) != status::StatusOK) {`
`158`	`158`	`return code;`
`159`	`159`	`}`
`160`	`160`	`}`