1/*
2 *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
3 *
4 *  Use of this source code is governed by a BSD-style license
5 *  that can be found in the LICENSE file in the root of the source
6 *  tree. An additional intellectual property rights grant can be found
7 *  in the file PATENTS.  All contributing project authors may
8 *  be found in the AUTHORS file in the root of the source tree.
9 */
10
11#include "webrtc/modules/video_coding/main/test/rtp_player.h"
12
13#include <stdio.h>
14
15#include <map>
16
17#include "webrtc/modules/rtp_rtcp/interface/rtp_header_parser.h"
18#include "webrtc/modules/rtp_rtcp/interface/rtp_payload_registry.h"
19#include "webrtc/modules/rtp_rtcp/interface/rtp_receiver.h"
20#include "webrtc/modules/rtp_rtcp/interface/rtp_rtcp.h"
21#include "webrtc/modules/video_coding/main/source/internal_defines.h"
22#include "webrtc/modules/video_coding/main/test/test_util.h"
23#include "webrtc/system_wrappers/interface/clock.h"
24#include "webrtc/system_wrappers/interface/critical_section_wrapper.h"
25#include "webrtc/system_wrappers/interface/scoped_ptr.h"
26#include "webrtc/test/rtp_file_reader.h"
27
28#if 1
29# define DEBUG_LOG1(text, arg)
30#else
31# define DEBUG_LOG1(text, arg) (printf(text "\n", arg))
32#endif
33
34namespace webrtc {
35namespace rtpplayer {
36
37enum {
38  kMaxPacketBufferSize = 4096,
39  kDefaultTransmissionTimeOffsetExtensionId = 2
40};
41
42class RawRtpPacket {
43 public:
44  RawRtpPacket(const uint8_t* data, uint32_t length, uint32_t ssrc,
45               uint16_t seq_num)
46      : data_(new uint8_t[length]),
47        length_(length),
48        resend_time_ms_(-1),
49        ssrc_(ssrc),
50        seq_num_(seq_num) {
51    assert(data);
52    memcpy(data_.get(), data, length_);
53  }
54
55  const uint8_t* data() const { return data_.get(); }
56  uint32_t length() const { return length_; }
57  int64_t resend_time_ms() const { return resend_time_ms_; }
58  void set_resend_time_ms(int64_t timeMs) { resend_time_ms_ = timeMs; }
59  uint32_t ssrc() const { return ssrc_; }
60  uint16_t seq_num() const { return seq_num_; }
61
62 private:
63  scoped_ptr<uint8_t[]> data_;
64  uint32_t length_;
65  int64_t resend_time_ms_;
66  uint32_t ssrc_;
67  uint16_t seq_num_;
68
69  DISALLOW_IMPLICIT_CONSTRUCTORS(RawRtpPacket);
70};
71
72class LostPackets {
73 public:
74  LostPackets(Clock* clock, uint32_t rtt_ms)
75      : crit_sect_(CriticalSectionWrapper::CreateCriticalSection()),
76        debug_file_(fopen("PacketLossDebug.txt", "w")),
77        loss_count_(0),
78        packets_(),
79        clock_(clock),
80        rtt_ms_(rtt_ms) {
81    assert(clock);
82  }
83
84  ~LostPackets() {
85    if (debug_file_) {
86      fclose(debug_file_);
87      debug_file_ = NULL;
88    }
89    while (!packets_.empty()) {
90      delete packets_.back();
91      packets_.pop_back();
92    }
93  }
94
95  void AddPacket(RawRtpPacket* packet) {
96    assert(packet);
97    printf("Throw:  %08x:%u\n", packet->ssrc(), packet->seq_num());
98    CriticalSectionScoped cs(crit_sect_.get());
99    if (debug_file_) {
100      fprintf(debug_file_, "%u Lost packet: %u\n", loss_count_,
101              packet->seq_num());
102    }
103    packets_.push_back(packet);
104    loss_count_++;
105  }
106
107  void SetResendTime(uint32_t ssrc, int16_t resendSeqNum) {
108    int64_t resend_time_ms = clock_->TimeInMilliseconds() + rtt_ms_;
109    int64_t now_ms = clock_->TimeInMilliseconds();
110    CriticalSectionScoped cs(crit_sect_.get());
111    for (RtpPacketIterator it = packets_.begin(); it != packets_.end(); ++it) {
112      RawRtpPacket* packet = *it;
113      if (ssrc == packet->ssrc() && resendSeqNum == packet->seq_num() &&
114          packet->resend_time_ms() + 10 < now_ms) {
115        if (debug_file_) {
116          fprintf(debug_file_, "Resend %u at %u\n", packet->seq_num(),
117                  MaskWord64ToUWord32(resend_time_ms));
118        }
119        packet->set_resend_time_ms(resend_time_ms);
120        return;
121      }
122    }
123    // We may get here since the captured stream may itself be missing packets.
124  }
125
126  RawRtpPacket* NextPacketToResend(int64_t time_now) {
127    CriticalSectionScoped cs(crit_sect_.get());
128    for (RtpPacketIterator it = packets_.begin(); it != packets_.end(); ++it) {
129      RawRtpPacket* packet = *it;
130      if (time_now >= packet->resend_time_ms() &&
131          packet->resend_time_ms() != -1) {
132        packets_.erase(it);
133        return packet;
134      }
135    }
136    return NULL;
137  }
138
139  int NumberOfPacketsToResend() const {
140    CriticalSectionScoped cs(crit_sect_.get());
141    int count = 0;
142    for (ConstRtpPacketIterator it = packets_.begin(); it != packets_.end();
143        ++it) {
144      if ((*it)->resend_time_ms() >= 0) {
145        count++;
146      }
147    }
148    return count;
149  }
150
151  void LogPacketResent(RawRtpPacket* packet) {
152    int64_t now_ms = clock_->TimeInMilliseconds();
153    CriticalSectionScoped cs(crit_sect_.get());
154    if (debug_file_) {
155      fprintf(debug_file_, "Resent %u at %u\n", packet->seq_num(),
156              MaskWord64ToUWord32(now_ms));
157    }
158  }
159
160  void Print() const {
161    CriticalSectionScoped cs(crit_sect_.get());
162    printf("Lost packets: %u\n", loss_count_);
163    printf("Packets waiting to be resent: %d\n", NumberOfPacketsToResend());
164    printf("Packets still lost: %zd\n", packets_.size());
165    printf("Sequence numbers:\n");
166    for (ConstRtpPacketIterator it = packets_.begin(); it != packets_.end();
167        ++it) {
168      printf("%u, ", (*it)->seq_num());
169    }
170    printf("\n");
171  }
172
173 private:
174  typedef std::vector<RawRtpPacket*> RtpPacketList;
175  typedef RtpPacketList::iterator RtpPacketIterator;
176  typedef RtpPacketList::const_iterator ConstRtpPacketIterator;
177
178  scoped_ptr<CriticalSectionWrapper> crit_sect_;
179  FILE* debug_file_;
180  int loss_count_;
181  RtpPacketList packets_;
182  Clock* clock_;
183  uint32_t rtt_ms_;
184
185  DISALLOW_IMPLICIT_CONSTRUCTORS(LostPackets);
186};
187
188class SsrcHandlers {
189 public:
190  SsrcHandlers(PayloadSinkFactoryInterface* payload_sink_factory,
191               const PayloadTypes& payload_types)
192      : payload_sink_factory_(payload_sink_factory),
193        payload_types_(payload_types),
194        handlers_() {
195    assert(payload_sink_factory);
196  }
197
198  ~SsrcHandlers() {
199    while (!handlers_.empty()) {
200      delete handlers_.begin()->second;
201      handlers_.erase(handlers_.begin());
202    }
203  }
204
205  int RegisterSsrc(uint32_t ssrc, LostPackets* lost_packets, Clock* clock) {
206    if (handlers_.count(ssrc) > 0) {
207      return 0;
208    }
209    DEBUG_LOG1("Registering handler for ssrc=%08x", ssrc);
210
211    scoped_ptr<Handler> handler(
212        new Handler(ssrc, payload_types_, lost_packets));
213    handler->payload_sink_.reset(payload_sink_factory_->Create(handler.get()));
214    if (handler->payload_sink_.get() == NULL) {
215      return -1;
216    }
217
218    RtpRtcp::Configuration configuration;
219    configuration.clock = clock;
220    configuration.id = 1;
221    configuration.audio = false;
222    handler->rtp_module_.reset(RtpReceiver::CreateVideoReceiver(
223        configuration.id, configuration.clock, handler->payload_sink_.get(),
224        NULL, handler->rtp_payload_registry_.get()));
225    if (handler->rtp_module_.get() == NULL) {
226      return -1;
227    }
228
229    handler->rtp_module_->SetNACKStatus(kNackOff);
230    handler->rtp_header_parser_->RegisterRtpHeaderExtension(
231        kRtpExtensionTransmissionTimeOffset,
232        kDefaultTransmissionTimeOffsetExtensionId);
233
234    for (PayloadTypesIterator it = payload_types_.begin();
235        it != payload_types_.end(); ++it) {
236      VideoCodec codec;
237      memset(&codec, 0, sizeof(codec));
238      strncpy(codec.plName, it->name().c_str(), sizeof(codec.plName)-1);
239      codec.plType = it->payload_type();
240      codec.codecType = it->codec_type();
241      if (handler->rtp_module_->RegisterReceivePayload(codec.plName,
242                                                       codec.plType,
243                                                       90000,
244                                                       0,
245                                                       codec.maxBitrate) < 0) {
246        return -1;
247      }
248    }
249
250    handlers_[ssrc] = handler.release();
251    return 0;
252  }
253
254  void IncomingPacket(const uint8_t* data, uint32_t length) {
255    for (HandlerMapIt it = handlers_.begin(); it != handlers_.end(); ++it) {
256      if (!it->second->rtp_header_parser_->IsRtcp(data, length)) {
257        RTPHeader header;
258        it->second->rtp_header_parser_->Parse(data, length, &header);
259        PayloadUnion payload_specific;
260        it->second->rtp_payload_registry_->GetPayloadSpecifics(
261            header.payloadType, &payload_specific);
262        it->second->rtp_module_->IncomingRtpPacket(header, data, length,
263                                                   payload_specific, true);
264      }
265    }
266  }
267
268 private:
269  class Handler : public RtpStreamInterface {
270   public:
271    Handler(uint32_t ssrc, const PayloadTypes& payload_types,
272            LostPackets* lost_packets)
273        : rtp_header_parser_(RtpHeaderParser::Create()),
274          rtp_payload_registry_(new RTPPayloadRegistry(
275              RTPPayloadStrategy::CreateStrategy(false))),
276          rtp_module_(),
277          payload_sink_(),
278          ssrc_(ssrc),
279          payload_types_(payload_types),
280          lost_packets_(lost_packets) {
281      assert(lost_packets);
282    }
283    virtual ~Handler() {}
284
285    virtual void ResendPackets(const uint16_t* sequence_numbers,
286                               uint16_t length) {
287      assert(sequence_numbers);
288      for (uint16_t i = 0; i < length; i++) {
289        lost_packets_->SetResendTime(ssrc_, sequence_numbers[i]);
290      }
291    }
292
293    virtual uint32_t ssrc() const { return ssrc_; }
294    virtual const PayloadTypes& payload_types() const {
295      return payload_types_;
296    }
297
298    scoped_ptr<RtpHeaderParser> rtp_header_parser_;
299    scoped_ptr<RTPPayloadRegistry> rtp_payload_registry_;
300    scoped_ptr<RtpReceiver> rtp_module_;
301    scoped_ptr<PayloadSinkInterface> payload_sink_;
302
303   private:
304    uint32_t ssrc_;
305    const PayloadTypes& payload_types_;
306    LostPackets* lost_packets_;
307
308    DISALLOW_COPY_AND_ASSIGN(Handler);
309  };
310
311  typedef std::map<uint32_t, Handler*> HandlerMap;
312  typedef std::map<uint32_t, Handler*>::iterator HandlerMapIt;
313
314  PayloadSinkFactoryInterface* payload_sink_factory_;
315  PayloadTypes payload_types_;
316  HandlerMap handlers_;
317
318  DISALLOW_IMPLICIT_CONSTRUCTORS(SsrcHandlers);
319};
320
321class RtpPlayerImpl : public RtpPlayerInterface {
322 public:
323  RtpPlayerImpl(PayloadSinkFactoryInterface* payload_sink_factory,
324      const PayloadTypes& payload_types, Clock* clock,
325      scoped_ptr<test::RtpFileReader>* packet_source,
326      float loss_rate, uint32_t rtt_ms, bool reordering)
327    : ssrc_handlers_(payload_sink_factory, payload_types),
328      clock_(clock),
329      next_rtp_time_(0),
330      first_packet_(true),
331      first_packet_rtp_time_(0),
332      first_packet_time_ms_(0),
333      loss_rate_(loss_rate),
334      lost_packets_(clock, rtt_ms),
335      resend_packet_count_(0),
336      no_loss_startup_(100),
337      end_of_file_(false),
338      reordering_(false),
339      reorder_buffer_() {
340    assert(clock);
341    assert(packet_source);
342    assert(packet_source->get());
343    packet_source_.swap(*packet_source);
344    srand(321);
345  }
346
347  virtual ~RtpPlayerImpl() {}
348
349  virtual int NextPacket(int64_t time_now) {
350    // Send any packets ready to be resent.
351    for (RawRtpPacket* packet = lost_packets_.NextPacketToResend(time_now);
352         packet != NULL;
353         packet = lost_packets_.NextPacketToResend(time_now)) {
354      int ret = SendPacket(packet->data(), packet->length());
355      if (ret > 0) {
356        printf("Resend: %08x:%u\n", packet->ssrc(), packet->seq_num());
357        lost_packets_.LogPacketResent(packet);
358        resend_packet_count_++;
359      }
360      delete packet;
361      if (ret < 0) {
362        return ret;
363      }
364    }
365
366    // Send any packets from packet source.
367    if (!end_of_file_ && (TimeUntilNextPacket() == 0 || first_packet_)) {
368      if (first_packet_) {
369        if (!packet_source_->NextPacket(&next_packet_))
370          return 0;
371        first_packet_rtp_time_ = next_packet_.time_ms;
372        first_packet_time_ms_ = clock_->TimeInMilliseconds();
373        first_packet_ = false;
374      }
375
376      if (reordering_ && reorder_buffer_.get() == NULL) {
377        reorder_buffer_.reset(
378            new RawRtpPacket(next_packet_.data,
379                             static_cast<uint32_t>(next_packet_.length),
380                             0,
381                             0));
382        return 0;
383      }
384      int ret = SendPacket(next_packet_.data,
385                           static_cast<uint32_t>(next_packet_.length));
386      if (reorder_buffer_.get()) {
387        SendPacket(reorder_buffer_->data(), reorder_buffer_->length());
388        reorder_buffer_.reset(NULL);
389      }
390      if (ret < 0) {
391        return ret;
392      }
393
394      if (!packet_source_->NextPacket(&next_packet_)) {
395        end_of_file_ = true;
396        return 0;
397      }
398      else if (next_packet_.length == 0) {
399        return 0;
400      }
401    }
402
403    if (end_of_file_ && lost_packets_.NumberOfPacketsToResend() == 0) {
404      return 1;
405    }
406    return 0;
407  }
408
409  virtual uint32_t TimeUntilNextPacket() const {
410    int64_t time_left = (next_rtp_time_ - first_packet_rtp_time_) -
411        (clock_->TimeInMilliseconds() - first_packet_time_ms_);
412    if (time_left < 0) {
413      return 0;
414    }
415    return static_cast<uint32_t>(time_left);
416  }
417
418  virtual void Print() const {
419    printf("Resent packets: %u\n", resend_packet_count_);
420    lost_packets_.Print();
421  }
422
423 private:
424  int SendPacket(const uint8_t* data, uint32_t length) {
425    assert(data);
426    assert(length > 0);
427
428    scoped_ptr<RtpHeaderParser> rtp_header_parser(RtpHeaderParser::Create());
429    if (!rtp_header_parser->IsRtcp(data, length)) {
430      RTPHeader header;
431      if (!rtp_header_parser->Parse(data, length, &header)) {
432        return -1;
433      }
434      uint32_t ssrc = header.ssrc;
435      if (ssrc_handlers_.RegisterSsrc(ssrc, &lost_packets_, clock_) < 0) {
436        DEBUG_LOG1("Unable to register ssrc: %d", ssrc);
437        return -1;
438      }
439
440      if (no_loss_startup_ > 0) {
441        no_loss_startup_--;
442      } else if ((rand() + 1.0)/(RAND_MAX + 1.0) < loss_rate_) {
443        uint16_t seq_num = header.sequenceNumber;
444        lost_packets_.AddPacket(new RawRtpPacket(data, length, ssrc, seq_num));
445        DEBUG_LOG1("Dropped packet: %d!", header.header.sequenceNumber);
446        return 0;
447      }
448    }
449
450    ssrc_handlers_.IncomingPacket(data, length);
451    return 1;
452  }
453
454  SsrcHandlers ssrc_handlers_;
455  Clock* clock_;
456  scoped_ptr<test::RtpFileReader> packet_source_;
457  test::RtpFileReader::Packet next_packet_;
458  uint32_t next_rtp_time_;
459  bool first_packet_;
460  int64_t first_packet_rtp_time_;
461  int64_t first_packet_time_ms_;
462  float loss_rate_;
463  LostPackets lost_packets_;
464  uint32_t resend_packet_count_;
465  uint32_t no_loss_startup_;
466  bool end_of_file_;
467  bool reordering_;
468  scoped_ptr<RawRtpPacket> reorder_buffer_;
469
470  DISALLOW_IMPLICIT_CONSTRUCTORS(RtpPlayerImpl);
471};
472
473RtpPlayerInterface* Create(const std::string& input_filename,
474    PayloadSinkFactoryInterface* payload_sink_factory, Clock* clock,
475    const PayloadTypes& payload_types, float loss_rate, uint32_t rtt_ms,
476    bool reordering) {
477  scoped_ptr<test::RtpFileReader> packet_source(test::RtpFileReader::Create(
478      test::RtpFileReader::kRtpDump, input_filename));
479  if (packet_source.get() == NULL) {
480    packet_source.reset(test::RtpFileReader::Create(test::RtpFileReader::kPcap,
481                                                    input_filename));
482    if (packet_source.get() == NULL) {
483      return NULL;
484    }
485  }
486
487  scoped_ptr<RtpPlayerImpl> impl(new RtpPlayerImpl(payload_sink_factory,
488      payload_types, clock, &packet_source, loss_rate, rtt_ms, reordering));
489  return impl.release();
490}
491}  // namespace rtpplayer
492}  // namespace webrtc
493