monitor.cc

rtpmon-1.0a7.tar.gz for UNIX like

/*
 * Copyright (c) 1996 The Regents of the University of California.
 * All rights reserved.
 *
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without written agreement
 * is hereby granted, provided that the above copyright notice and the
 * following two paragraphs appear in all copies of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING
 * OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF THE
 * UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE UNIVERSITY OF CALIFORNIA HAS NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 *
 * $Id: monitor.cc,v 1.24 1997/01/13 23:31:40 aswan Exp $
 */

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <errno.h>
#include <string.h>

#include "Tcl.h"
//#include "crypt.h"
#include "config.h"
#include "member.h"
#include "rtp.h"
#include "monitor.h"

#ifndef HAVE_SNPRINTF
extern "C" {
    int snprintf(char* buf, int s, const char* fmt, ...);
}
#endif

// forget about source if no control message for 6 report intervals
#define CTRL_IDLE 6

#define HASH(a) ((int)((((a) >> 20) ^ ((a) >> 10) ^ (a)) & (HASH_SIZE-1)))

static class MonitorMatcher : public Matcher {
  public:
    MonitorMatcher() : Matcher("monitor") {}
    TclObject* match(const char*) {
	return (new Monitor());
    }
} session_matcher;

UpdateTimer::UpdateTimer(Monitor& m) : m_(m), inactive_(1)
{
    Tcl& tcl = Tcl::instance();
    const char* bwstr = tcl.attr("sessbw");
    bw_ = (bwstr == 0) ? 64. : atof(bwstr);

    // Convert session bandwidth from kb/s to bytes/sec (*128)
    // and then to control b/w (* 5%)
    bw_ *= 128. * .05;
}

#define TIMER_INTERVAL 2000
void
UpdateTimer::timeout() {
    // update median
    m_.median();

    if (--inactive_ == 0) {
	int newinterval = m_.CheckActive(bw_);
	inactive_ = newinterval / TIMER_INTERVAL;
    }

    msched(TIMER_INTERVAL);
}

Monitor::Monitor() : net_(0), nrunt_(0), badversion_(0), badpt_(0),
    nmembers_(0),members_(0), senders_(0), last_(0), median_(0),
    avgpktsize_(0), bits_(0), sorttype_(SORT_MAX), stat_(0),
    thresh_(0), ut_(*this)
{
    pktbuf_ = new u_char[2 * RTP_MTU];
    memset((char*)hashtab_, 0, sizeof(hashtab_));

    // start the timer firing every 2 seconds
    ut_.msched(TIMER_INTERVAL);
}

Monitor::~Monitor()
{
    delete pktbuf_;
}

int
Monitor::command(int argc, const char*const* argv)
{
    Tcl& tcl = Tcl::instance();
    static char *work;
    static int worksize = 0;

    if (worksize == 0) {
	worksize = 1024;
	work = new char[worksize];
    }
    int size = worksize;

    if (argc == 2) {
	if (strcmp(argv[1], "clean") == 0) {
	    for (Member* m = members_; m != NULL; m = m->next_) {
		if (! m->active()) {
		    /*
		     * remove this member
		     */

		    Member** p = &hashtab_[HASH(m->srcid())];
		    while (*p != m)
			p = &(*p)->hlink_;
		    *p = (*p)->hlink_;
	    
		    if (m->sender()) {
			p = &senders_;
			while (*p != 0 && *p != m)
			    p = &(*p)->snext_;
			if (*p != 0)
			    *p = (*p)->snext_;
		    }

		    p = &members_;
		    while (*p != m)
			p = &(*p)->next_;
		    if (*p == last_) {
			for (Member* t = members_; t != 0; t = t->next_)
			    if (t->next_ == 0)
				last_ = t;
		    }
		    *p = (*p)->next_;

		    if (median_ == m)
			median_ = 0;	/* XXX do something more intelligent */

		    delete m;
		    --nmembers_;
		}

		if (m->locked()) {
		    for (Member* s = senders_; s != 0; s = s->snext_) {
			int n = s->pos();
			if ((stat_ == 0 && m->loss(n) < thresh_)
			    || (stat_ == 1 && m->filtloss(n) < thresh_)
			    || (stat_ == 2 && m->jitter(n) < thresh_))
			    m->locked(false);
		    }
		}
	    }
	}
	if (strcmp(argv[1], "inactive") == 0) {
	    Member* m;
	    char *p = work;
	    *p = 0;
	    for (m=members_; m!=NULL; m=m->next_) {
		if (! m->active()) {
		    strncpy(p, m->name(), size);
		    int len = strlen(p);
		    size -= len+1;
		    p += len;
		    *p++ = ' ';
		    *p = 0;

		    if (size < 2) {
			int newsize = worksize << 1;
			char* newwork = new char[worksize];
			memcpy(newwork, work, worksize);
			worksize = newsize;
			p = newwork + (p - work);
			delete[] work;
			work = newwork;
		    }
		}
	    }
	    tcl.result(work);
	    return(TCL_OK);
	}
	if (strcmp(argv[1], "listeners") == 0) {
	    char *p = work;
	    *p = 0;

	    for (Member* m=members_; m!=NULL; m=m->next_) {
		if (!m->locked())
		    continue;

		strncpy(p, m->name(), size);
		int len = strlen(p);
		size -= len+1;
		p += len;
		*p++ = ' ';
		*p = 0;

		if (size < 2) {
		    int newsize = worksize << 1;
		    char* newwork = new char[worksize];
		    memcpy(newwork, work, worksize);
		    worksize = newsize;
		    p = newwork + (p - work);
		    delete[] work;
		    work = newwork;
		}
	    }
	    tcl.result(work);
	    return (TCL_OK);
	}
	if (strcmp(argv[1], "senders") == 0) {
	    char *p = work;
	    *p = 0;
	    for (Member* s = senders_; s != 0; s = s->snext_) {
		strncpy(p, s->name(), size);
		int len = strlen(p);

		size -= len+1;
		p += len;
		*p++ = ' ';
		*p = 0;

		if (size < 2) {
		    int newsize = worksize << 1;
		    char* newwork = new char[worksize];
		    memcpy(newwork, work, worksize);
		    worksize = newsize;
		    p = newwork + (p - work);
		    delete[] work;
		    work = newwork;
		}
	    }
	    tcl.result(work);
	    return (TCL_OK);
	}
	if (strcmp(argv[1], "sort") == 0) {
	    sort();
	    tcl.result("");
	    return (TCL_OK);
	}
	if (strcmp(argv[1], "statnames") == 0) {
	    tcl.result("{loss \"Packet Loss\"} {filtlost \"Filtered Packet Loss\"} {jitter \"Delay Jitter\"}");
	    return (TCL_OK);
	}
	if (strcmp(argv[1], "stats") == 0) {
	    char* p = work;

	    int len = snprintf(p, size, "{\"Bad Version\" %d} {\"Unexpected Payload Type\" %d} {\"Runt Packets\" %d} {\"Average Packet Size\" \"%d bytes\"}", badversion_, badpt_, nrunt_, avgpktsize_);
	    tcl.result(work);
	    return(TCL_OK);
	}
    } else if (argc == 3) {
	if (strcmp(argv[1], "ignore") == 0) {
	    Member* m = (Member*)TclObject::lookup(argv[2]);
	    if (!m->sender()) return(TCL_ERROR);
	    m->ignore(true);

	    Member** p = &senders_;
	    while (*p != 0 && *p != m)
		p = &(*p)->snext_;
	    if (*p != 0)
		*p = (*p)->snext_;

	    return(TCL_OK);
	}
	if (strcmp(argv[1], "net") == 0) {
	    net((Network*)TclObject::lookup(argv[2]));
	    return (TCL_OK);
	}
   	if (strcmp(argv[1], "sort") == 0) {
	    if (strcmp(argv[2], "max") == 0)
		sort(SORT_MAX);
	    else if (strcmp(argv[2], "avg") == 0)
		sort(SORT_AVG);
	    else if (strcmp(argv[2], "address") == 0)
		sort(SORT_ADDR);
	    else {
		TclObject* obj = TclObject::lookup(argv[2]);
		if (obj != 0) {
		    Member* m = (Member*)obj;
		    sort(m->pos());
		}
	    }

	    tcl.result("");
	    return (TCL_OK);
	}
	if (strcmp(argv[1], "stat") == 0) {
	    int v = atoi(argv[2]);
	    if (v < 0 || v > 2)
		return(TCL_ERROR);
	    stat_ = v;

	    Tcl_Interp* interp = Tcl::instance().interp();
	    char tmpname[32];

	    for (Member* s = members_; s != 0; s = s->next_) {
		int n = s->pos();
		for (Member* l = members_; l != 0; l = l->next_) {
		    sprintf(tmpname, "%s:%s", l->name(), s->name() );
		    char val[32];

		    switch (stat_) {
		      case 0:
			sprintf(val, "%d %%", (int)l->loss(n));
			break;
		      case 1:
			sprintf(val, "%d %%", (int)l->filtloss(n));
			break;
		      case 2:
			u_int32_t j = l->jitter(n);
			j = (j&0x0ffff) * 1000 > 16;
			sprintf(val, "%u ms", j);
			break;
		    }
		    Tcl_SetVar2(interp, "stats", tmpname, val, TCL_GLOBAL_ONLY);
		}
	    }
	    
	    tcl.result("");
	    return(TCL_OK);
	}
	if (strcmp(argv[1], "thresh") == 0) {
	    int v = atoi(argv[2]);

	    // XXX this is ugly; fix it later
	    if (stat_ == 2)
		thresh_ = (v << 16) / 1000;
	    else
		thresh_ = v;
	    Tcl& tcl = Tcl::instance();

	    for (Member* m = members_; m != 0; m = m->next_) {
		if (!m->locked()) {
		    for (Member* s = senders_; s != 0; s = s->snext_) {
			int n = s->pos();
			if ((stat_ == 0 && m->loss(n) >= thresh_)
			    || (stat_ == 1 && m->filtloss(n) >= thresh_)
			    || (stat_ == 2 && m->jitter(n) >= thresh_)) {
			    m->locked(true);
			    tcl.evalf("new-listener %s", m->name());
			}
		    }
		}
	    }

	    return (TCL_OK);
	}
    }
    return (TCL_OK);
}

Member*
Monitor::lookup(u_int32_t ssrc, u_int32_t addr)
{
    int h = HASH(ssrc);
    Member* m;
    for (m = hashtab_[h]; m != NULL; m = m->hlink_) {
	if (m->srcid() == ssrc) return m;
    }
    
    // if called from parse_rr_records, we don't want to
    // create a new entry so just ignore it for now
    if (addr == 0) return 0;

    m = new Member(ssrc, addr);
    m->active(1);
    if (members_ == 0) {
	m->next_ = 0;
	members_ = last_ = m;
    } else {
	m->next_ = 0;
	last_->next_ = m;
	last_ = m;
    }
    m->hlink_ = hashtab_[h];
    hashtab_[h] = m;
    ++nmembers_;

    Tcl_Interp* interp = Tcl::instance().interp();
    char tmpname[32], objname[32];
    sprintf(tmpname, "#%u", ssrc);
    strcpy(objname, m->name());
    Tcl_SetVar2(interp, "names", objname, tmpname, TCL_GLOBAL_ONLY);

    return m;
}

int
Monitor::getpos()
{
    for (int i=0; i<32; i++)
	if (!((bits_>>i) & 1)) {
	    bits_ |= 1<<i;
	    return i;
	}

    return -1;
}

void
Monitor::freepos(int pos)
{
    bits_ &= ~(1<<pos);
}

void
Monitor::dispatch(int mask)
{
    u_int32_t src;
    int cc = net_->recv(pktbuf_, 2 * RTP_MTU, src);
    if (cc <= 0)
	return;
    
    rtcphdr* rh = (rtcphdr*) pktbuf_;
    if (cc < (int)sizeof(*rh)) {
	++nrunt_;
	return;
    }
    /*
     * try to filter out junk: first thing in packet must be
     * sr, rr or bye & version number must be correct.
     */
    switch(ntohs(rh->rh_flags) & 0xc0ff) {
      case RTP_VERSION << 14 | RTCP_PT_SR:
      case RTP_VERSION << 14 | RTCP_PT_RR:
      case RTP_VERSION << 14 | RTCP_PT_BYE:
	break;
	
      default:
	u_short flags = ntohs(rh->rh_flags);
	if ((flags>>14)&3 != RTP_VERSION)
	    ++badversion_;
	else
	    ++badpt_;
	return;
    }

    // update average packet size
    avgpktsize_ += (cc - avgpktsize_) >> 4;

    u_int32_t ssrc = rh->rh_ssrc;
    Member *m = lookup(ssrc, src);
    m->active(1);

    /*
     * Outer loop parses multiple RTCP records of a "compound packet".
     * There is no framing between records.  Boundaries are implicit
     * and the overall length comes from UDP.
     */
    u_char* epack = (u_char*)rh + cc;
    while ((u_char*)rh < epack) {
	u_int len = (ntohs(rh->rh_len) << 2) + 4;
	u_char* ep = (u_char*)rh + len;
	if (ep > epack) {
	    m->badlen(1);
	    return;
	}
	u_int flags = ntohs(rh->rh_flags);
	if (flags >> 14 != RTP_VERSION) {
	    m->badver(1);
	    return;
	}
	switch (flags & 0xff) {

	  case RTCP_PT_SR:
	    parse_sr(rh, flags, ep, m, src);
	    break;

	  case RTCP_PT_RR:
	    parse_rr(rh, flags, ep, m, src);
	    break;

	  case RTCP_PT_SDES:
	    parse_sdes(rh, flags, ep, m, ssrc, src);
	    break;