[Sumover-dev] [svn commit] r4831 - in vic/branches/cc: codec rtp
sumover-dev at cs.ucl.ac.uk
sumover-dev at cs.ucl.ac.uk
Wed May 12 19:19:07 BST 2010
Author: soohyunc
Date: Wed May 12 19:19:07 2010
New Revision: 4831
Modified:
vic/branches/cc/cc/tfrc_sndr.cpp
vic/branches/cc/cc/tfrc_sndr.h
vic/branches/cc/cc/tfwc_sndr.cpp
vic/branches/cc/cc/tfwc_sndr.h
vic/branches/cc/codec/encoder-h261.cpp
vic/branches/cc/rtp/session.cpp
vic/branches/cc/rtp/session.h
vic/branches/cc/rtp/transmitter.cpp
vic/branches/cc/rtp/transmitter.h
Log:
It is critically important for TFWC (or any kind of ack-clock based CC
mechanisms) to receive Ack as soon as it arrives to the data sender, so that the
data sender can update CC information (cwnd or send rate) in a timely fashion.
We have discovered that Ack can be arrived to the sender in a bulky manner -
i.e., multimple Acks arrived in a very short time scale.
Although TFWC had a mechanism to pull out the arrived-ack before sending every
data packet, it wasn't sufficient when MANY Acks are arrived already.
Therefore, we have created a while loop to pull out the arrived XR as soon as
possible as long as it is there. So, these Acks are not the normal Ack clock,
hence we have changed some variable names accordingly (recv_by_ch => ack_clock).
Also, there was a typo in tfwc_sndr.h when returning the number of bytes that
this Ack is giving.
Modified: vic/branches/cc/cc/tfrc_sndr.cpp
==============================================================================
--- vic/branches/cc/cc/tfrc_sndr.cpp (original)
+++ vic/branches/cc/cc/tfrc_sndr.cpp Wed May 12 19:19:07 2010
@@ -161,9 +161,9 @@
* main TFRC reception path
*/
void TfrcSndr::recv(u_int16_t type, u_int16_t begin, u_int16_t end,
- u_int16_t *chunk, double so_rtime, bool recv_by_ch, pktbuf* pb)
+ u_int16_t *chunk, double so_rtime, bool ack_clock, pktbuf* pb)
{
- UNUSED(recv_by_ch);
+ UNUSED(ack_clock);
UNUSED(pb);
switch(type) {
Modified: vic/branches/cc/cc/tfrc_sndr.h
==============================================================================
--- vic/branches/cc/cc/tfrc_sndr.h (original)
+++ vic/branches/cc/cc/tfrc_sndr.h Wed May 12 19:19:07 2010
@@ -49,7 +49,7 @@
virtual ~TfrcSndr() {};
// virtual functions
- virtual void tfrc_output(bool recv_by_ch=0) {UNUSED(recv_by_ch);};
+ virtual void tfrc_output(bool ack_clock=0) {UNUSED(ack_clock);};
virtual void tfrc_output(pktbuf*) {};
virtual double tx_ts_offset() {};
virtual int tx_buf_size() {};
Modified: vic/branches/cc/cc/tfwc_sndr.cpp
==============================================================================
--- vic/branches/cc/cc/tfwc_sndr.cpp (original)
+++ vic/branches/cc/cc/tfwc_sndr.cpp Wed May 12 19:19:07 2010
@@ -199,9 +199,9 @@
* main TFWC reception path
*/
void TfwcSndr::recv(u_int16_t type, u_int16_t begin, u_int16_t end,
- u_int16_t *chunk, double so_rtime, bool recv_by_ch, pktbuf* pb)
+ u_int16_t *chunk, double so_rtime, bool ack_clock, pktbuf* pb)
{
- UNUSED(recv_by_ch);
+ UNUSED(ack_clock);
UNUSED(pb);
switch (type) {
@@ -255,7 +255,7 @@
if(jacked_ < aoa_) {
debug_msg("warning: this ack(%d) is older than AoA(%d)!\n", jacked_,aoa_);
// trigger a packet out to keep Jacob's packet conservation rule
- //packet_clocking(pb, recv_by_ch);
+ //packet_clocking(pb, ack_clock);
// this ack is already too old,
// so revert to the eariler history
Modified: vic/branches/cc/cc/tfwc_sndr.h
==============================================================================
--- vic/branches/cc/cc/tfwc_sndr.h (original)
+++ vic/branches/cc/cc/tfwc_sndr.h Wed May 12 19:19:07 2010
@@ -63,8 +63,8 @@
virtual ~TfwcSndr() {};
// virtual functions
- virtual void tfwc_output(bool recv_by_ch=0) {UNUSED(recv_by_ch);};
- virtual void tfwc_output(pktbuf*) {};
+ virtual void tfwc_output(bool ack_clock=0) {UNUSED(ack_clock);};
+ virtual void tfwc_output(pktbuf*, bool) {};
virtual void tfwc_trigger(pktbuf* pb=0) {UNUSED(pb);};
virtual double tx_ts_offset() {};
virtual int tx_buf_size() {};
@@ -74,14 +74,14 @@
// main reception path (XR packet)
void recv(u_int16_t type, u_int16_t begin, u_int16_t end,
- u_int16_t *chunk, double so_rtime, bool recv_by_ch, pktbuf* pb);
+ u_int16_t *chunk, double so_rtime, bool ack_clock, pktbuf* pb);
// return ackofack
inline u_int16_t get_aoa() { return aoa_; }
// just acked seqno in packets and in bytes
inline u_int16_t jacked() { return jacked_; }
- inline int b_jacked() { return record_[bjacked_%PSR]; }
+ inline int b_jacked() { return record_[jacked_%PSR]; }
// tfwc controlled cwnd value in packets and in bytes
inline u_int32_t magic() { return cwnd_; }
Modified: vic/branches/cc/codec/encoder-h261.cpp
==============================================================================
--- vic/branches/cc/codec/encoder-h261.cpp (original)
+++ vic/branches/cc/codec/encoder-h261.cpp Wed May 12 19:19:07 2010
@@ -847,7 +847,7 @@
}
// increment the number of packets for this frame
ppframe_[vfno_%FHSIZE]++;
- print_ppframe();
+ //print_ppframe();
tx_->send(pb);
Modified: vic/branches/cc/rtp/session.cpp
==============================================================================
--- vic/branches/cc/rtp/session.cpp (original)
+++ vic/branches/cc/rtp/session.cpp Wed May 12 19:19:07 2010
@@ -132,8 +132,7 @@
{
int i = 0; int cc = 0;
while((cc = sm_->recv(this)) > 0) {
- //fprintf(stderr, "\tackvec reception: now: %f cc[%d]: %d\n",
- //sm_->tx_get_now(), ++i, cc);
+ i++; //dispatch_info(sm_->tx_get_now(), cc, i);
}
}
@@ -226,8 +225,7 @@
confid_(-1),
seqno_(0), // RTP data packet seqno (from RTP header)
lastseq_(0), // last received packet's seqno
-ackvec_(0), // bit vector (AckVec)
-new_ack_(0)
+ackvec_(0) // bit vector (AckVec)
{
/*XXX For adios() to send bye*/
manager = this;
@@ -453,19 +451,14 @@
return (TCL_ERROR);
}
-void SessionManager::transmit(pktbuf* pb, bool recv_by_ch)
+void SessionManager::transmit(pktbuf* pb, bool ack_clock)
{
// mh_.msg_iov = pb->iov;
// dh_[.net()->send(mh_);
// debug_msg("L %d,",pb->layer);
- int i = 0; int cc = 0;
-
// receive XR before sending
- while ((cc = recv_xreport(ch_, pb, recv_by_ch)) > 0) {
- //fprintf(stderr, "\tackvec reception: now: %f cc[%d]: %d\n",
- //tx_get_now(), ++i, cc);
- }
+ tot_num_acked_ = check_xr_arrival(pb, 1);
// print RTP seqno
print_rtp_seqno(pb);
@@ -497,15 +490,10 @@
}
}
-void SessionManager::tx_data_only(pktbuf* pb, bool recv_by_ch)
+void SessionManager::tx_data_only(pktbuf* pb, bool ack_clock)
{
- int i = 0; int cc = 0;
-
// receive XR before sending
- while ((cc = recv_xreport(ch_, pb, recv_by_ch)) > 0) {
- //fprintf(stderr, "\tackvec reception: now: %f cc[%d]: %d\n",
- //tx_get_now(), ++i, cc);
- }
+ tot_num_acked_ = check_xr_arrival(pb, 1);
// print RTP seqno
print_rtp_seqno(pb);
@@ -1341,7 +1329,7 @@
}
void SessionManager::parse_xr(rtcphdr* rh, int flags, u_char* ep,
- Source* ps, Address & addr, int layer, bool recv_by_ch, pktbuf* pb)
+ Source* ps, Address & addr, int layer, bool ack_clock, pktbuf* pb)
{
Source* s;
u_int32_t ssrc = rh->rh_ssrc; // RTCP's ssrc
@@ -1352,11 +1340,11 @@
s->layer(layer).lts_ctrl(unixtime());
int cnt = flags >> 8 & 0x1f; // reception report count
- parse_xr_records(ssrc, (rtcp_xr*)(rh + 1), cnt, ep, addr, recv_by_ch, pb);
+ parse_xr_records(ssrc, (rtcp_xr*)(rh + 1), cnt, ep, addr, ack_clock, pb);
}
void SessionManager::parse_xr_records(u_int32_t ssrc, rtcp_xr* xr, int cnt,
- const u_char* ep, Address & addr, bool recv_by_ch, pktbuf* pb)
+ const u_char* ep, Address & addr, bool ack_clock, pktbuf* pb)
{
parse_xr_banner_top();
UNUSED(ssrc);
@@ -1397,27 +1385,21 @@
//so_rtime = ch_[0].net()->recv_so_time();
//sender_xr_ts_info(so_rtime);
- // if this XR reception is called not because of ack clock,
- // then this XR reception is called just before sending packet,
- // hence, mark it that we have retrieved a new ack
- if(!recv_by_ch)
- set_new_ack();
-
switch (cc_type_) {
case WBCC:
// TFWC sender (getting AckVec)
- tfwc_sndr_.recv(xr->BT,begin,end,chunk,recv_ts_,recv_by_ch,pb);
+ tfwc_sndr_.recv(xr->BT,begin,end,chunk,recv_ts_,ack_clock,pb);
// we need to call Transmitter::output(pb) to make Ack driven
- if(recv_by_ch)
- tfwc_output(recv_by_ch);
+ if(ack_clock)
+ tfwc_output(ack_clock);
break;
case RBCC:
// TFRC sender (getting AckVec)
- tfrc_sndr_.recv(xr->BT,begin,end,chunk,recv_ts_,recv_by_ch,pb);
+ tfrc_sndr_.recv(xr->BT,begin,end,chunk,recv_ts_,ack_clock,pb);
// XXX
- if(recv_by_ch)
- tfrc_output(recv_by_ch);
+ if(ack_clock)
+ tfrc_output(ack_clock);
break;
} // switch (cc_type_)
}
@@ -1446,8 +1428,26 @@
parse_xr_banner_bottom();
}
+/*
+ * check if XR has arrived
+ * @pb: RTP data packet
+ * @ack_clock: this SHOULD be always set to 1,
+ * although it is not an ack-clocking actually
+ * (i.e., check XR reception ONLY)
+ * return: total number of bytes that the retrieved Acks have given
+ */
+int SessionManager::check_xr_arrival(pktbuf* pb, bool ack_clock){
+ int i = 0; int cc = 0;
+ // receive XR before sending
+ while (recv_xreport(ch_, pb, ack_clock) > 0) {
+ cc += tfwc_sndr_.b_jacked(); i++;
+ // xr_arrival_info(cc, i);
+ }
+ return (cc);
+}
+
// receive XR (AckVec)
-int SessionManager::recv_xreport(CtrlHandler *ch, pktbuf* pb, bool recv_by_ch) {
+int SessionManager::recv_xreport(CtrlHandler *ch, pktbuf* pb, bool ack_clock) {
// timestamp for XR reception
recv_ts_ = tx_get_now();
@@ -1522,7 +1522,7 @@
switch (flags & 0xff) {
case RTCP_PT_XR:
rtcp_pkt_id = RTCP_PT_XR;
- parse_xr(rh, flags, ep, ps, addr, layer, !(recv_by_ch), pb);
+ parse_xr(rh, flags, ep, ps, addr, layer, !(ack_clock), pb);
break;
case RTCP_PT_SR:
rtcp_pkt_id = RTCP_PT_SR;
Modified: vic/branches/cc/rtp/session.h
==============================================================================
--- vic/branches/cc/rtp/session.h (original)
+++ vic/branches/cc/rtp/session.h Wed May 12 19:19:07 2010
@@ -117,6 +117,11 @@
double ctrl_inv_bw_;
double ctrl_avg_size_; /* (estimated) average size of ctrl packets */
double rint_; /* current session report rate (in ms) */
+
+ // control channel dispatch debug info
+ inline void dispatch_info(double now, int nbytes, int i) {
+ fprintf(stderr, " \tnow: %f\tdispatched[%d]: %d\n",now,i,nbytes);
+ }
};
class ReportTimer : public Timer {
@@ -157,17 +162,15 @@
// am i a data sender?
inline bool am_i_sender() { return is_sender_; }
- // was there a new ack?
- inline virtual bool new_ack() { return new_ack_; }
- inline virtual void set_new_ack() { new_ack_ = true; }
- inline virtual void reset_new_ack() { new_ack_ = false; }
+ // see if any XR has arrived
+ virtual int check_xr_arrival(pktbuf*, bool);
protected:
// void demux(rtphdr* rh, u_char* bp, int cc, Address & addr, int layer);
void demux(pktbuf* pb, Address & addr);
virtual int check_format(int fmt) const = 0;
- virtual void transmit(pktbuf* pb, bool recv_by_ch=0);
- virtual void tx_data_only(pktbuf* pb, bool recv_by_ch);
+ virtual void transmit(pktbuf* pb, bool ack_clock=0);
+ virtual void tx_data_only(pktbuf* pb, bool ack_clock);
void send_report(int bye);
void send_ECNXreport(CtrlHandler* ch, u_int8_t tos, u_int16_t begin_seq);
int build_bye(rtcphdr* rh, Source& local);
@@ -181,11 +184,11 @@
void parse_rr(rtcphdr* rh, int flags, u_char* ep,
Source* ps, Address & addr, int layer);
void parse_xr(rtcphdr* rh, int flags, u_char* ep,
- Source* ps, Address & addr, int layer, bool recv_by_ch, pktbuf* pb=0);
+ Source* ps, Address & addr, int layer, bool ack_clock, pktbuf* pb=0);
void parse_rr_records(u_int32_t ssrc, rtcp_rr* r, int cnt,
const u_char* ep, Address & addr);
void parse_xr_records(u_int32_t ssrc, rtcp_xr* xr, int cnt,
- const u_char* ep, Address & addr, bool recv_by_ch, pktbuf* pb);
+ const u_char* ep, Address & addr, bool ack_clock, pktbuf* pb);
int sdesbody(u_int32_t* p, u_char* ep, Source* ps,
Address & addr, u_int32_t ssrc, int layer);
void parse_sdes(rtcphdr* rh, int flags, u_char* ep, Source* ps,
@@ -240,8 +243,6 @@
u_int16_t ackvec_; // this is a bit vector
// timestamp
double recv_ts_; // receive timestamp
- // was there a new ack?
- bool new_ack_;
private:
// print RTP data packet's seqno
@@ -250,13 +251,12 @@
}
inline void print_rtp_seqno(pktbuf* pb) {
rtphdr* rh = (rtphdr *) pb->data;
- fprintf(stderr, "\n\tnow: %f\tseqno: %d\n\n",tx_get_now(),ntohs(rh->rh_seqno));
+ fprintf(stderr, "\n\tnow: %f\tseqno: %d\n\n",
+ tx_get_now(),ntohs(rh->rh_seqno));
}
-
// print sender's XR info
inline void sender_xr_info(const char* str, const int i,
- u_int16_t b, u_int16_t e, rtcp_xr_BT_1_hdr* xrh, u_int16_t l)
- {
+ u_int16_t b, u_int16_t e, rtcp_xr_BT_1_hdr* xrh, u_int16_t l) {
fprintf(stderr, " [%s +%d] beg:%d, end:%d, xr1len:%d (xrlen:%d)\n",
str, i, b, e, ntohs(xrh->xr_len),l);
}
@@ -278,6 +278,9 @@
fprintf(stderr,
"-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-\n");
}
+ inline void xr_arrival_info(int nbytes, int i) {
+ fprintf(stderr, " \tnow: %f\tnbytes[%d]: %d\n",tx_get_now(),i,nbytes);
+ }
};
class AudioSessionManager : public SessionManager {
Modified: vic/branches/cc/rtp/transmitter.cpp
==============================================================================
--- vic/branches/cc/rtp/transmitter.cpp (original)
+++ vic/branches/cc/rtp/transmitter.cpp Wed May 12 19:19:07 2010
@@ -252,7 +252,7 @@
} else
tail_ = head_ = pb;
pb->next = 0;
- tfwc_output();
+ tfwc_output(0);
is_buf_empty_ = false;
}
// if not, check if cwnd allows send this packet
@@ -263,7 +263,7 @@
} else
tail_ = head_ = pb;
pb->next = 0;
- tfwc_output(pb);
+ tfwc_output(pb, 0);
}
break;
@@ -321,18 +321,26 @@
} // switch (cc_type)
}
-void Transmitter::tfwc_output(pktbuf* pb)
+void Transmitter::tfwc_output(pktbuf* pb, bool ack_clock)
{
//cc_output_banner_top("tfwc");
// byte mode? or packet mode?
switch (cwnd_mode_) {
case BYM:
{
- if(pb->len < tfwc_sndr_.b_magic()) {
+ // see if any XR has arrived to pick up
+ int cc = check_xr_arrival(pb, 1);
+
+ while(pb->len <= tfwc_sndr_.b_magic() + cc) {
// move head pointer
head_ = pb->next;
// call Transmitter::output_data_only w/ XR reception
- output_data_only(pb, XR_RECV);
+ output_data_only(pb, ack_clock);
+
+ if (head_ != 0)
+ pb = head_;
+ else
+ break;
}
}
break;
@@ -340,15 +348,22 @@
{
// pb is not null, hence parse it.
rtphdr* rh = (rtphdr *) pb->data;
+ // see if any XR has arrived to pick up
+ check_xr_arrival(pb, 1);
- if (ntohs(rh->rh_seqno) <= tfwc_sndr_.magic() + tfwc_sndr_.jacked()) {
+ while (ntohs(rh->rh_seqno) <= tfwc_sndr_.magic() + tfwc_sndr_.jacked()) {
//debug_msg("cwnd: %d\n", tfwc_sndr_.magic());
//debug_msg("jack: %d\n", tfwc_sndr_.jacked());
// move head pointer
head_ = pb->next;
// call Transmitter::output_data_only w/ XR reception
- output_data_only(pb, XR_RECV);
+ output_data_only(pb, ack_clock);
+
+ if (head_ != 0)
+ pb = head_;
+ else
+ break;
}
}
break;
@@ -359,7 +374,7 @@
/*
* main TFWC CC output routines
*/
-void Transmitter::tfwc_output(bool recv_by_ch)
+void Transmitter::tfwc_output(bool ack_clock)
{
//cc_output_banner_top("tfwc");
// head of the RTP data packet buffer (pb)
@@ -379,26 +394,19 @@
case BYM:
{
int len = 0;
- int num_acked = 0;
-
// cwnd (in bytes)
int b_magic = tfwc_sndr_.b_magic();
+ // see if any XR has arrived to pick up
+ int cc = check_xr_arrival(pb, 1);
- // this output called upon ack clock
- if(recv_by_ch)
- num_acked = tfwc_sndr_.b_jacked();
-
- while(pb->len < b_magic + num_acked - len) {
+ while(pb->len <= b_magic + cc - len) {
len += pb->len;
// move head pointer
head_ = pb->next;
// call Transmitter::output(pb)
- output(pb, recv_by_ch);
-
- // number of acked bytes
- if (new_ack())
- num_acked += tfwc_sndr_.b_jacked();
- reset_new_ack();
+ output(pb, ack_clock);
+ // inflate nbytes as we receive ack
+ cc += tot_num_acked_;
if (head_ != 0)
pb = head_;
@@ -411,6 +419,8 @@
{
// pb is not null, hence parse it.
rtphdr* rh = (rtphdr *) pb->data;
+ // see if any XR has arrived to pick up
+ check_xr_arrival(pb, 1);
// while packet seqno is within "cwnd + jack", send that packet
while (ntohs(rh->rh_seqno) <= tfwc_sndr_.magic() + tfwc_sndr_.jacked()) {
@@ -420,7 +430,7 @@
// move head pointer
head_ = pb->next;
// call Transmitter::output(pb)
- output(pb, recv_by_ch);
+ output(pb, ack_clock);
// if the moved head pointer is not null, parse packet buffer.
// otherwise, break while statement.
@@ -453,6 +463,9 @@
return;
}
+ // see if any XR has arrived in the mean time
+ check_xr_arrival(pb, 1);
+
// parse pb data
//rtphdr* rh = (rtphdr *) pb->data;
// move head pointer
@@ -473,7 +486,7 @@
cc_output_banner_bottom();
}
-void Transmitter::tfrc_output(bool recv_by_ch) {
+void Transmitter::tfrc_output(bool ack_clock) {
cc_output_banner_top("tfrc");
// head of the RTP data packet buffer
pktbuf* pb = head_;
@@ -489,7 +502,7 @@
// move head pointer
head_ = pb->next;
// call Transmitter::output(pb)
- output(pb, recv_by_ch);
+ output(pb, ack_clock);
if (head_ != 0) {
pb = head_;
@@ -540,13 +553,13 @@
}
}
-void Transmitter::output(pktbuf* pb, bool recv_by_ch)
+void Transmitter::output(pktbuf* pb, bool ack_clock)
{
//fprintf(stderr, "\n\tTransmitter::output()\n");
//if (dumpfd_ >= 0)
// dump(dumpfd_, pb->iov, mh_.msg_iovlen);
//dprintf("layer: %d \n",pb->layer);
- transmit(pb, recv_by_ch);
+ transmit(pb, ack_clock);
loopback(pb);
// pb->release() is called by decoder in loopback;
}
Modified: vic/branches/cc/rtp/transmitter.h
==============================================================================
--- vic/branches/cc/rtp/transmitter.h (original)
+++ vic/branches/cc/rtp/transmitter.h Wed May 12 19:19:07 2010
@@ -101,11 +101,11 @@
void send(pktbuf*);
inline bool is_cc_on() { return is_cc_active_; }
// TFWC output
- virtual void tfwc_output(bool recv_by_ch=0);
- virtual void tfwc_output(pktbuf*);
+ virtual void tfwc_output(bool ack_clock=0);
+ virtual void tfwc_output(pktbuf*, bool ack_clock);
virtual void tfwc_trigger(pktbuf*);
// TFRC output
- virtual void tfrc_output(bool recv_by_ch=0);
+ virtual void tfrc_output(bool ack_clock=0);
virtual void tfrc_output(pktbuf*);
/*
@@ -136,17 +136,16 @@
void update(int nbytes);
void dump(int fd, iovec*, int iovel) const;
void loopback(pktbuf*);
- void output(pktbuf* pb, bool recv_by_ch=0);
+ void output(pktbuf* pb, bool ack_clock=0);
void output_data_only(pktbuf* pb, bool flag);
- virtual void transmit(pktbuf* pb, bool recv_by_ch=0) = 0;
+ virtual void transmit(pktbuf* pb, bool ack_clock=0) = 0;
virtual void tx_data_only(pktbuf* pb, bool flag) = 0;
double gettimeofday_secs() const;
double txtime(pktbuf* pb);
- // new ack arrived?
- virtual bool new_ack() = 0;
- virtual void set_new_ack() = 0;
- virtual void reset_new_ack() = 0;
+ // check if new XR(s) have arrived
+ virtual int check_xr_arrival(pktbuf*, bool) = 0;
+ int tot_num_acked_;
int mtu_; /* mtu of wire (as seen by application) */
msghdr mh_;
More information about the Sumover-dev
mailing list