monitor.cc

rtpmon-1.0a7.tar.gz for UNIX like

	  case RTCP_PT_BYE:
	    parse_bye(rh, flags, ep, m);
	    break;

	  default:
	    m->badopt(1);
	    break;
	}
	rh = (rtcphdr*)ep;
    }
}

void
Monitor::parse_sr(rtcphdr* rh, int flags, u_char* ep,
		  Member* m, u_int32_t addr)
{
    rtcp_sr* sr = (rtcp_sr*)(rh + 1);
    u_int32_t ssrc = rh->rh_ssrc;
    if (m->srcid() != ssrc) {
	m = lookup(ssrc, addr);
    }
    time_t tm = time(0);
    m->ts(tm);

#ifdef RTCP_DEBUG
    Tcl& tcl = Tcl::instance();
    tcl.evalf("rtcp_debug \"%u\" \"sr: npkts: %d nbytes: %d\"", ssrc,
	      ntohl(sr->sr_np), ntohl(sr->sr_nb));
#endif
    int cnt = flags >> 8 & 0x1f;
    parse_rr_records(ssrc, (rtcp_rr*)(sr + 1), cnt, ep, m, addr);
}

void
Monitor::parse_rr(rtcphdr* rh, int flags, u_char* ep,
		  Member* m, u_int32_t addr)
{
    u_int32_t ssrc = rh->rh_ssrc;
    if (m->srcid() != ssrc) {
	m = lookup(ssrc, addr);
    }
    time_t tm = time(0);
    m->ts(tm);

    int cnt = flags >> 8 & 0x1f;
    parse_rr_records(ssrc, (rtcp_rr*)(rh + 1), cnt, ep, m, addr);
}

void
Monitor::parse_rr_records(u_int32_t ssrc, rtcp_rr* rr, int cnt,
			     u_char* ep, Member *m, u_int32_t addr)
{
    while ((u_char*)rr < ep) {
	u_int32_t srcid = rr->rr_srcid;
	Member* s = lookup(srcid, 0);
#ifdef RTCP_DEBUG
	{
	    Tcl& tcl = Tcl::instance();
	    const char *p = NULL;

	    if (s != 0)
		p = s->sdes(RTCP_SDES_CNAME);
	    tcl.evalf("rtcp_debug \"%u\" \"rr: srcid: %u (%s) fracloss: %u%% cumloss: %u ehsr: %u dv: %u lsr: %u dlsr: %u\"",
		      m->srcid(), ntohl(rr->rr_srcid), p ? p : "?",
		      ((ntohl(rr->rr_loss) >> 24) & 0xff) * 100 / 256,
		      (ntohl(rr->rr_loss) & 0xffffff), ntohl(rr->rr_ehsr),
		      ntohl(rr->rr_dv), ntohl(rr->rr_lsr), ntohl(rr->rr_dlsr));
	}
#endif
	if (s == 0) {
	    rr++;
	    continue;
	}
	if (! s->sender()) {
	    /*
	     * senders list must be kept in order -- add to the
	     * end of the list
	     */
	    s->snext_ = 0;
	    if (senders_ == 0)
		senders_ = s;
	    else {
		Member* sn = senders_;
		while (sn->snext_ != 0)
		    sn = sn->snext_;
		sn->snext_ = s;
	    }
		
	    s->pos(getpos());
	    s->sender(true);
	    /*
	     * don't update stats yet; wait until we hear at least
	     * twice that this srcid is a sender
	     */
	    rr++;
	    continue;
	}

	int n = s->pos();
	m->loss(n, ntohl(rr->rr_loss)&0x0ffffff, ntohl(rr->rr_ehsr));
	m->filtloss(n, ntohl(rr->rr_loss) >> 24);

	u_int32_t jit = ntohl(rr->rr_dv);
	if ((jit > 0x1000) && ((jit & 0x00f0ffff) == 0)) {
	    jit = (jit & 0xff000000) >> 24 | (jit & 0x000f0000) >> 16;
	}
	m->jitter(n, jit);

	char idx[64];
	sprintf(idx, "%s:%s", m->name(), s->name());
	char statbuf[32];
	bool isnew = false;

	//XXX this is ugly; fix it later
	u_char l;
	switch (stat_) {
	  case 0:  // packet loss
	    l = m->loss(n);
	    if (!m->locked() && l >= thresh_)
		isnew = true;
	    sprintf(statbuf, "%d %%", (int)l );
	    break;

	  case 1:  // filtered packet loss
	    l = m->filtloss(n);
	    if (!m->locked() && l >= thresh_)
		isnew = true;
	    sprintf(statbuf, "%d %%", (int)m->filtloss(n) );
	    break;

	  case 2:  // jitter
	    u_int32_t j = m->jitter(n);
	    if (!m->locked() && j >= thresh_)
		isnew = true;
	    j = (j&0x0ffff) * 1000 >> 16;
	    sprintf(statbuf, "%u ms", j);
	    break;
	}

	Tcl& tcl = Tcl::instance();

	if (isnew) {
	    m->locked(1);
	    tcl.evalf("new-listener %s", m->name());
	}
	Tcl_Interp* interp = tcl.interp();
	Tcl_SetVar2(interp, "stats", idx, statbuf, TCL_GLOBAL_ONLY);
	rr++;
    }
}

void
Monitor::parse_sdes(rtcphdr* rh, int flags, u_char* ep, Member* m,
		       u_int32_t ssrc, u_int32_t addr)
{
    int cnt = flags >> 8 & 0x1f;
    u_int32_t* p = (u_int32_t*)&rh->rh_ssrc;
    while (--cnt >= 0) {
	int n = sdesbody(p, ep, m, addr);
	if (n == 0)
	    break;
	p += n;
    }
    if (cnt >= 0)
	m->badsdes(1);
}

int
Monitor::sdesbody(u_int32_t* p, u_char* ep, Member* m, u_int32_t addr)
{
    u_int32_t srcid = *p;
    if (m->srcid() != srcid)
	m = lookup(srcid, addr);
    if (m == 0)
	return (0);

    u_char* cp = (u_char*)(p + 1);
    while (cp < ep) {
	u_int type = cp[0];
	if (type == 0) {
	    /* end of chunk */
	    return (((cp - (u_char*)p) >> 2) + 1);
	}
	u_int len = cp[1];
	u_char* eopt = cp + len + 2;
	if (eopt > ep)
	    return (0);

	if (type >= RTCP_SDES_MIN && type <= RTCP_SDES_MAX) {
	    char buf[256];
	    memcpy(buf, (char*)&cp[2], len);
	    buf[len] = 0;

	    const char* oldval= m->sdes(RTCP_SDES_NAME);
	    char newval[256];
	    newval[0] = '\0';
	    if (type == RTCP_SDES_NAME && buf != 0
		&& (oldval == 0 || strcmp(oldval, buf)))
		strcpy(newval, buf);

	    oldval = m->sdes(RTCP_SDES_CNAME);
	    if (type == RTCP_SDES_CNAME && buf != 0
		&& (oldval == 0 || strcmp(oldval, buf)))
		strcpy(newval, buf);

	    m->sdes(type, buf);

	    if (newval[0] != 0) {
		char objname[32];
		strcpy(objname, m->name());
		Tcl_Interp* interp = Tcl::instance().interp();
		Tcl_SetVar2(interp, "names", objname, newval, TCL_GLOBAL_ONLY);
	    }

	}
	cp = eopt;
    }
    return (0);
}

void
Monitor::parse_bye(rtcphdr* rh, int flags, u_char* ep, Member* m)
{
    int cnt = flags >> 8 & 0x1f;
    u_int32_t* p = (u_int32_t*)&rh->rh_ssrc;

    while (--cnt >= 0) {
	if (p >= (u_int32_t*)ep) {
	    //m->badbye(1);
	    return;
	}
	if (m->srcid() != rh->rh_ssrc)
	    m = lookup(*p, 0);

	if (m != 0)
	    m->active(0);

	++p;
    }
}

void
Monitor::sort()
{
    Member* m;
    int i, b;
    u_int32_t v;

    for (m=members_; m!=NULL; m=m->next_) {
	v = 0;
	
	switch(sorttype_) {
	  case SORT_MAX:
	    for (i=0, b=bits_; b!=0; i++, b>>=1) {
		if (! b&1) continue;
		if (stat_ == 0 && m->loss(i) > v)
		    v = m->loss(i);
		if (stat_ == 1 && m->filtloss(i) > v)
		    v = m->filtloss(i);
		if (stat_ == 2 && m->jitter(i) > v)
		    v = m->jitter(i);
	    }
	    m->val(v);
	    break;

	  case SORT_AVG:
	    for (i=0, b=bits_; b!=0; i++, b>>=1) {
		if (! b&1) continue;
		switch (stat_) {
		  case 0:
		    v += m->loss(i);
		    break;
		  case 1:
		    v += m->filtloss(i);
		    break;
		  case 2:
		    v += m->jitter(i);
		    break;
		}
	    }
	    m->val(v);
	    break;
	    
	  case SORT_ADDR:
	    m->val(~m->addr());
	    break;
	    
	  default:
	    if (! (bits_>>sorttype_)&1)
		return;  // XXX error
	    switch (stat_) {
	      case 0:
		m->val(m->loss(sorttype_));
		break;
	      case 1:
		m->val(m->filtloss(sorttype_));
		break;
	      case 2:
		m->val(m->jitter(sorttype_));
		break;
	    }
	}
    }
    
    members_ = dosort(members_);

    // reset the end pointer
    for (Member* t = members_; t != 0; t = t->next_)
	if (t->next_ == 0)
	    last_ = t;
}

Member*
Monitor::dosort(Member* l)
{
    if (l == NULL || l->next_ == NULL)
	return (l);
    Member* a = l;
    Member* b = l->next_;
    while (b != NULL && b->next_ != NULL) {
	a = a->next_;
	b = b->next_->next_;
    }
    b = a->next_;
    a->next_ = NULL;
    b = dosort(b);
    a = dosort(l);

    Member* first;
    if (a->val() >= b->val()) {
	first = a;
	a = a->next_;
    } else {
	first = b;
	b = b->next_;
    }
    Member* m = first;
    while (a != NULL && b != NULL) {
	if (a->val() >= b->val()) {
	    m->next_ = a;
	    m = a;
	    a = a->next_;
	} else {
	    m->next_ = b;
	    m = b;
	    b = b->next_;
	}
    }
    if (a != NULL)
	m->next_ = a;
    if (b != NULL)
	m->next_ = b;

    return (first);
}

int
Monitor::CheckActive(double maxbw)
{
    time_t now = time(0);
    double idletime = (double(nmembers_) * avgpktsize_) / maxbw;
    if (idletime < 5.0) idletime = 5.0;

    u_int maxidle = u_int(idletime) * CTRL_IDLE;
    
    Member *m, *next;
    for (m = members_; m != 0; m = next) {
	next = m->next_;
	u_int32_t t = (u_int32_t) m->ts();
	if (u_int(now - t) > maxidle)
	    m->active(0);
    }

    // Return interval until next re-sort
    return (int(idletime * 1000.));
}

void
Monitor::median()
{
    int n, b;
    Member *less, *more;
    int nless, nmore;

    if (nmembers_ == 0) return;

    Tcl_Interp* interp = Tcl::instance().interp();

    for (Member* s = senders_ ; s!= 0; s = s->snext_) {
	n = s->pos();

	less = more = 0;
	nless = nmore = 0;

	if (nmembers_ > 2) {
	    u_int32_t pivot;
	    if (median_ == 0)
		pivot = 5;	// XXX
	    else
		pivot = median_->getstat(stat_, n);

	    u_int32_t max = 0;

	    // do the first separation before calling mselect
	    // since the member list is linked by next, not mnext
	    for (Member* m = members_; m != 0; m = m->next_) {
		u_int32_t stat = m->getstat(stat_, n);

		if (stat > max) max = stat;

		if (stat < pivot) {
		    m->mnext_ = less;
		    less = m;
		    nless++;
		} else {
		    m->mnext_ = more;
		    more = m;
		    nmore++;
		}
	    }

	    int i = nmembers_ / 2;
	    if (nless > i)
		median_ = mselect(less, i, 0, pivot, n);
	    else
		median_ = mselect(more, i-nless, pivot, max, n);
	} else {
	    median_ = members_;
	}

	char val[32];
	switch (stat_) {
	  case 0:  case 1:
	    sprintf(val, "%d %%", (int)median_->getstat(stat_, n));
	    break;

	  case 2:
	    u_int32_t j = median_->jitter(n);
	    j = (j&0x0ffff) * 1000 > 16;
	    sprintf(val, "%u ms", j);
	    break;
	}
	Tcl_SetVar2(interp, "median", s->name(), val, TCL_GLOBAL_ONLY);
    }
}

Member*
Monitor::mselect(Member* m, int i, int low, int high, int n)
{
    Member *less = 0, *more = 0;
    int nless = 0, nmore = 0, done = 1;

    if (m->mnext_ == 0 || high == low) return m;
    if (high == low+1) return m;	// XXX

    u_int32_t v = m->getstat(stat_, n);
    u_int32_t pivot = (low + high) / 2;

    Member *t, *next;
    for (t = m; t != 0; t = next) {
	next = t->mnext_;
	u_int32_t s = t->getstat(stat_, n);
	if (s != v) done = 0;
	if (s < pivot) {
	    t->mnext_ = less;
	    less = t;
	    nless++;
	} else {
	    t->mnext_ = more;
	    more = t;
	    nmore++;
	}
    }

    if (done) return m;
    return (nless>i) ? mselect(less, i, low, pivot, n)
	: mselect(more, i-nless, pivot, high, n);
}