receiver.cpp

voudp - VoIP for NetBSD

//
// Copyright 2004 Alan Post
//
// This file is part of voudp.
//
// voudp is free software; you can redistribute it and/or modify it under the
// terms of the GNU General Public License as published by the Free Software
// Foundation; either version 2 of the License, or (at your option) any later
// version.
//
// voudp is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
// A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along with
// voudp; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
// Suite 330, Boston, MA  02111-1307  USA
//

//
// XXX:  on receipt of explicit shutdown packet, don't nuke channel immediately,
// but instead wait for all the queued frames to drain.
//

#include "receiver.h"
#include "constants.h"
#include "die.h"

#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <map>
#include <set>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/audioio.h>
#include <sys/ioctl.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <err.h>


#include "speex.h"

class frame_t {
public:
    //
    // We copy the data on the way in.
    //
    frame_t( const u_int8_t *data_in, ssize_t len_in )
    : data( new u_int8_t[ len_in ]), len( len_in )
    {
        memcpy( data, data_in, len );
    }
    ~frame_t () { delete[] data; }
    u_int8_t *data;
    ssize_t len;
};

class decoder_t {
public:
    decoder_t()
    {
        st = speex_decoder_init( &speex_nb_mode );
        int tmp = 1;
        speex_decoder_ctl( st, SPEEX_SET_ENH, &tmp );
        speex_bits_init( &bits );
    }

    ~decoder_t()
    {
        speex_bits_destroy( &bits );
        speex_decoder_destroy( st ); 
    }

    void output( float *samples, frame_t *f )
    {
        speex_bits_read_from( &bits, f?(char*)f->data:NULL, f?f->len:0 );
        speex_decode( st, &bits, samples ); 
    }

private:

    void *st;
    SpeexBits bits;
public:
};

class frame_queue_t {
private:
    typedef std::map<frame_id_t, frame_t*> frame_map_t;
    frame_map_t fm;
    frame_id_t fid_floor;
    decoder_t decoder;

    size_t len_check_interval;
    size_t min_queue_len;

public:
    size_t max_f_lag;

    size_t n_p_received;
    size_t n_f_received;
    size_t n_f_played;
    size_t n_f_guessed;
    size_t n_f_late;
    size_t n_f_cut;

    size_t n_f_cut_mark;
    size_t n_f_guessed_mark;

    client_id_t cid;
    timeval last_used;

    ~frame_queue_t()
    {
        for ( frame_map_t::iterator i = fm.begin(); i != fm.end(); ++i )
            delete (*i).second;
    }

    frame_queue_t( const size_t      &max_frame_lag,
                   const client_id_t &cid_in,
                   const timeval     &now,
                   const size_t      &lci )
    : fm(), fid_floor( 0 ), decoder(), len_check_interval( lci ),
      min_queue_len( SIZE_T_MAX ),
      max_f_lag( max_frame_lag ), 
      n_p_received( 0 ),
      n_f_received( 0 ),
      n_f_played( 0 ),
      n_f_guessed( 0 ),
      n_f_late( 0 ),
      n_f_cut( 0 ),
      n_f_cut_mark( 0 ),
      n_f_guessed_mark( 0 ),
      cid( cid_in ), last_used( now )
    {}

    bool empty() const { return fm.empty(); }
    size_t size() const { return fm.size(); }

    //
    // Caller takes responsibility for deleting.
    //
    // May return NULL to indicate padding frame.
    //
    frame_t *pop_playable()
    {
        n_f_played++;
        frame_t *f = NULL;

        if ( fm.empty())
        {
            n_f_guessed++;
        }
        else if ( ! fm.count( fid_floor ))
        {
            n_f_guessed++;
            fid_floor++;
        }
        else
        {
            f = fm[ fid_floor ];
            fm.erase( fid_floor );
            fid_floor++;
        }

        min_queue_len = min( min_queue_len, fm.size());
        if ( n_f_played % len_check_interval == 0 )
        {
            if ( min_queue_len > 0 ) max_f_lag--;
            else
            {
                size_t cut = n_f_cut - n_f_cut_mark;
                size_t guessed = n_f_guessed - n_f_guessed_mark;

                if (    cut > 0 && guessed > 0
                     && len_check_interval / cut < 100
                     && len_check_interval / guessed < 100 )
                {
                    max_f_lag++;
                }
            }
            min_queue_len = SIZE_T_MAX;
            n_f_cut_mark = n_f_cut;
            n_f_guessed_mark = n_f_guessed;
        }

        return f;
    }

    void output( float *samples )
    {
        frame_t *f = pop_playable();
        decoder.output( samples, f );
        delete f;
    }

    //
    // Makes a private copy of the data.
    //
    void insert_frame( const frame_id_t &fid, const u_int8_t *data,
                       const ssize_t &len )
    {
        n_f_received++;
        if ( fid < fid_floor )
        {
            n_f_late++;
            return;
        }

        if ( fm.count( fid ))
        {
            delete fm[ fid ];
            fm.erase( fid );
        }

        fm[ fid ] = new frame_t( data, len );
        if ( fid > max_f_lag )
            fid_floor = max( fid_floor, fid - max_f_lag + 1 );

        frame_map_t::iterator i;
        while ( i = fm.begin(),
                (*i).first < fid_floor )
        {
            fm.erase( i );
            n_f_cut++;
        }
    }
};

struct timeval operator-( const struct timeval &a,
                          const struct timeval &b )
{
    struct timeval result;
    timersub( &a, &b, &result );
    return result;
}

bool operator<( const struct timeval &a,
                const struct timeval &b )
{
    return timercmp( &a, &b, < );
}

bool
cmp_fqs_by_time_t::operator()( frame_queue_t *a, frame_queue_t *b )
{
    return a->last_used < b->last_used;
}


static int hookup( const int &port )
{
    struct sockaddr_in sa;
    bzero( &sa, sizeof( sa ));
#ifndef LINUX
    sa.sin_len = sizeof( sa );
#endif
    sa.sin_family = AF_INET;
    sa.sin_port = htons( port );
    sa.sin_addr.s_addr = htonl( INADDR_ANY );
    int sfd = socket( PF_INET, SOCK_DGRAM, 0 );
    if ( sfd == -1 ) die( "socket" );
    if ( -1 == bind( sfd, (struct sockaddr*) &sa, sizeof( sa ))) die( "bind" );
    return sfd;
}


//
// OK to call with nonexistent cid
//
static void nuke_fq( const client_id_t    &cid,
                           frame_queues_t &fqs,
                           fqs_by_time_t  &fqs_by_time )
{
    if ( ! fqs.count( cid )) return;

    frame_queue_t *fq = fqs[ cid ];
    fqs_by_time.erase( fq );
    fqs.erase( cid );
    delete fq;
}

Receiver::Receiver( const int    &port,
                    const size_t &max_ms_lag,
                    const size_t &max_ms_guess )
{
    p_fqs = new frame_queues_t;
    p_fqs_by_time = new fqs_by_time_t;

    max_f_lag_default =   max_ms_lag * ( samples_per_sec / 1000 )
                        / samples_per_frame;
    timeval foo = { max_ms_guess/1000, (max_ms_guess%1000)*1000 };
    max_guess = foo;
    recv_fd = hookup( port );

    n_f_silence = 0;
    n_f_sound = 0;
}

Receiver::~Receiver()
{
    delete p_fqs;
    delete p_fqs_by_time;
}

int
Receiver::get_sock_fd() const { return recv_fd; }

void
Receiver::recv_packet()
{
    frame_queues_t &fqs = *p_fqs;
    fqs_by_time_t &fqs_by_time = *p_fqs_by_time;

    u_int8_t buf[ 65536 ];
    ssize_t nread = recv( recv_fd, buf, sizeof( buf ), 0 );
    if ( nread == -1 ) { warn( "recv" ); return; }
    if ( nread < (ssize_t)sizeof( client_id_t )) return;  // sanity check

    timeval now;
    if ( 0 != gettimeofday( &now, NULL )) die( "gettimeofday" );

    u_int32_t n_cid;
    memcpy( &n_cid, buf, sizeof( n_cid ));
    client_id_t cid = ntohl( n_cid );

    if ( nread == sizeof( cid ))
    {
        //
        // Explicit client shutdown packet
        //
        nuke_fq( cid, fqs, fqs_by_time );
        return;
    }

    if ( ! fqs.count( cid ))
        fqs[ cid ] = new frame_queue_t( max_f_lag_default, cid, now,
                                        4 * samples_per_sec / samples_per_frame );

    frame_queue_t *fq = fqs[ cid ];

    fq->n_p_received++;
    fqs_by_time.erase( fq );
    fq->last_used = now;
    fqs_by_time.insert( fq );

    ssize_t buf_pos = sizeof( cid );
    while ( buf_pos < nread )
    {
        if ( buf_pos + frame_header_size > nread ) return;

        u_int32_t n_fid;
        memcpy( &n_fid, buf + buf_pos, sizeof( n_fid ));
        buf_pos += sizeof( n_fid );
        frame_id_t fid = ntohl( n_fid );

        frame_len_t flen;
        memcpy( &flen, buf + buf_pos, sizeof( flen ));
        buf_pos += sizeof( flen );

        if ( buf_pos + flen > nread ) return;
        fq->insert_frame( fid, buf + buf_pos, flen );
        buf_pos += flen;
    }
}

void
Receiver::decode_frame( float *frame_acc )
{
    frame_queues_t &fqs = *p_fqs;
    fqs_by_time_t &fqs_by_time = *p_fqs_by_time;

    timeval now;
    if ( 0 != gettimeofday( &now, NULL )) die( "gettimeofday" );

    assert( fqs.size() == fqs_by_time.size());
    fqs_by_time_t::iterator i;
    while ( i = fqs_by_time.begin(),
               i != fqs_by_time.end()
            && (*i)->last_used < now - max_guess )
    {
        nuke_fq( (*i)->cid, fqs, fqs_by_time );
    }

    float frame[ samples_per_frame ];
    if ( fqs.empty())
    {
        n_f_silence++;
        bzero( frame_acc, sizeof( frame ));
    }
    else
    {
        n_f_sound++;
        if ( fqs.size() == 1 ) { (*(fqs.begin())).second->output( frame_acc ); }
        else
        {
            bzero( frame_acc, sizeof( frame ));
            for ( frame_queues_t::const_iterator i = fqs.begin();
                  i != fqs.end(); i++ )
            {
                (*i).second->output( frame );
                for ( size_t j = 0; j < samples_per_frame; j++ )
                    frame_acc[ j ] += frame[ j ];
            }
        }
    }
}

void
Receiver::print_diagnostics() const
{
    fprintf( stderr, "receiver\n"
                     "  frames of silence: %8lu\n"
                     "    frames of sound: %8lu\n",
             (unsigned long int) n_f_silence,
             (unsigned long int) n_f_sound );

    for ( frame_queues_t::const_iterator i = p_fqs->begin();
          i != p_fqs->end(); ++i )
    {
        fprintf( stderr, "frame queue id %10x length %8lu max %8lu\n"
                         "  packets received: %8lu\n"
                         "   frames received: %8lu\n"
                         "     frames played: %8lu\n"
                         "    frames guessed: %8lu\n"
                         "       frames late: %8lu\n"
                         "        frames cut: %8lu\n",
                                     (*i).second->cid,
                 (unsigned long int) (*i).second->size(),
                 (unsigned long int) (*i).second->max_f_lag,
                 (unsigned long int) (*i).second->n_p_received,
                 (unsigned long int) (*i).second->n_f_received,
                 (unsigned long int) (*i).second->n_f_played,
                 (unsigned long int) (*i).second->n_f_guessed,
                 (unsigned long int) (*i).second->n_f_late,
                 (unsigned long int) (*i).second->n_f_cut );
    }
}