/*------------------------------------------------------------------------
 *
 *  File:	o1k_network.cc
 *  Revision:	1.23
 *  Last Mod:	14:52:30, 03/06/97
 *
 *  Description:
 *	
 *      Implementation of conference network class. Contains
 *      routines for starting and stopping network threads.
 *      The code for the various network threads are also
 *      within this file. 
 *    
 *----------------------------------------------------------------------*/
#pragma ident	"@(#)o1k_network.cc	1.23\t97/03/06  "

#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <thread.h>
#include <synch.h>
#include <sys/types.h>
#include <sys/time.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <malloc.h>
#include <memory.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>

#include "o1k_conf.h"
#include "o1k_misc.h"
#include "packet.h"
#include "transmit.h"
#include "receive.h"

/*
 * The current implementation of this conference example limits
 * a conference to a point to point connection. This following
 * define is not meant to imply that without it, a multipoint
 * conference is possible.
 */
#define LIMIT_ONE_HOST

/* Prototype Network Worker Threads */
extern void* tNetThrdA(void* vp);
extern void* tNetThrdV(void* vp);
extern void* tNetThrdC(void* vp);
extern void* rNetThrdA(void* vp);
extern void* rNetThrdV(void* vp);
extern void* rNetThrdC(void* vp);

/* Prototype receive packet routine */
extern BigPacketType* ReceivePackets(O1kNetwork* network, O1kThread* thread, int type, int init);

/*
 * Debugging flags which enable writing to /tmp a record of
 * all incoming/outgoing network data.
 */
int AUDIO_WRITE_SHADOW = 0;
int CONTROL_WRITE_SHADOW = 0;
int VIDEO_WRITE_SHADOW = 0;

/*
 * Other entities communicate with the network threads via
 * a ring buffer of data objects (TNetData and TNetData
 * below). The reader/writer relationship between these
 * entities and the network work threads is handled by
 * semaphores. These data objects are declared here.
 */

#define  SEMA_EMPTY 0
#define  SEMA_FULL  1

sema_t   tbuf_sem[2][NET_MAX_TYPES];
sema_t   rbuf_sem[2][NET_MAX_TYPES];

unsigned int tring[2][NET_MAX_TYPES];
unsigned int rring[2][NET_MAX_TYPES];

typedef struct _TNetData {
  unsigned char  data[MAX_FRAME_SIZE];
  unsigned int   size;
} TNetData;

typedef struct _RNetData {
  unsigned char  data[MAX_FRAME_SIZE];
  unsigned int   size;
} RNetData;

#define NET_NUM_DATA_BUFS 2

TNetData* tnetdata[NET_NUM_DATA_BUFS][NET_MAX_TYPES];
RNetData* rnetdata[NET_NUM_DATA_BUFS][NET_MAX_TYPES];

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::O1kNetwork()
 *
 *  Description:
 *	
 *     Constructor for O1kNetwork object. The global semaphores
 *     are initialized here as well as the allocation of the
 *     ring buffer data. O1kThread objects for each of the various 
 *     network workers are created.
 *	
 *  Notes:
 *	
 *     Failure of O1kNetwork() is not currently being detected.
 *	
 *----------------------------------------------------------------------*/

O1kNetwork::O1kNetwork()
{
  int i,j;

  for (i=0; i<NET_MAX_TYPES; i++){
    tstate[i] = NULL;
    rstate[i] = NULL;

    if (sema_init(&tbuf_sem[SEMA_FULL][i],  0, USYNC_THREAD, NULL) < 0){
      fprintf(stderr,"sema_init t[%d] error\n",i);
    }
    if (sema_init(&tbuf_sem[SEMA_EMPTY][i], 2, USYNC_THREAD, NULL) < 0){
      fprintf(stderr,"sema_init t[%d] error\n",i);
    } 
    
    if (sema_init(&rbuf_sem[SEMA_FULL][i],  0, USYNC_THREAD, NULL) < 0){
      fprintf(stderr,"sema_init t[%d] error\n",i);
    }
    if (sema_init(&rbuf_sem[SEMA_EMPTY][i], 2, USYNC_THREAD, NULL) < 0){
      fprintf(stderr,"sema_init t[%d] error\n",i);
    }
    
    for (j=0; j<NET_NUM_DATA_BUFS; j++){
      if ((tnetdata[j][i] = (TNetData*)malloc(sizeof(TNetData))) == NULL){
        fprintf(stderr,"alloc tnetdata[%d] failed\n",i);
      }
      if ((rnetdata[j][i] = (RNetData*)malloc(sizeof(RNetData))) == NULL){
        fprintf(stderr,"alloc rnetdata[%d] failed\n",i);
      }
    }
    tring[0][i] = tring[1][i] = 0;
    rring[0][i] = rring[1][i] = 0;    
  }

  hostnameknt = 0;
  rchannel = 0;
  tchannel = 0;
  polling = 0;
  type = SOCK_DGRAM;

  status.packets_t_v   = 0;
  status.packets_r_v   = 0;
  status.drops_v       = 0;
  
  status.packets_t_a   = 0;
  status.packets_r_a   = 0;  
  status.drops_a       = 0;
  
  strcpy(destination,"");
  
  for (i = 0; i < MAXSENDTO; i++)
    hostname[i][0] = '\0';

  vid_thread_t = new O1kThread("Network Video Transmit");
  vid_thread_r = new O1kThread("Network Video Receive");
  aud_thread_t = new O1kThread("Network Audio Transmit");
  aud_thread_r = new O1kThread("Network Audio Receive");
  ctl_thread_t = new O1kThread("Network Control Transmit");
  ctl_thread_r = new O1kThread("Network Control Receive");
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::~O1kNetwork()
 *
 *  Description:
 *	
 *     Destructor for O1kNetwork object.
 *	
 *----------------------------------------------------------------------*/

O1kNetwork::~O1kNetwork()
{
  int i,j;
  
  /* delete thread objects */
  if (vid_thread_t)
    delete vid_thread_t;
  if (vid_thread_r)
    delete vid_thread_r;
  if (aud_thread_t)
    delete aud_thread_t;
  if (aud_thread_r)
    delete aud_thread_r;
  if (ctl_thread_t)
    delete ctl_thread_t;
  if (ctl_thread_r)
    delete ctl_thread_r;

  /* free ring buffer data */
  for (i=0; i<NET_MAX_TYPES; i++){
    for (j=0; j<NET_NUM_DATA_BUFS; j++){
      if (tnetdata[j][i])
        free(tnetdata[j][i]);
      if (rnetdata[j][i])
        free(rnetdata[j][i]);
    }
  }
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::settings()
 *
 *  Description:
 *	
 *     Debug print out of O1kNetwork settings.
 *	
 *----------------------------------------------------------------------*/

void O1kNetwork::settings()
{
  int i;

  printf("Network Settings\n");
  printf("----------------\n");
  
  if (hostnameknt == 0)
    printf("Broadcasting Mode\n");
  else
    for (i=0; i<hostnameknt; i++){
      printf("Sending to host: %s\n",hostname[i]);
    }
  
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::addHost()
 *
 *  Description:
 *	
 *     Add a host to the list of hosts to send to.
 *	
 *  Parameters:
 *     char* name    : target host name
 *	
 *  Notes:
 *	
 *     This conference example has only been tested under point-
 *     to-point conditions and for the most part is only currently
 *     designed as a point-to-point app. The underlying network
 *     transmit/receive mechanism, however, does support multicast.
 *     Thus at least that portion is ready for multi-point. For this
 *     reason, the code which allows multiple host was left intact
 *     below.
 *	
 *----------------------------------------------------------------------*/

void O1kNetwork::addHost(char* name)
{

  if (verbose) printf("Adding host %s\n",name);
#ifdef LIMIT_ONE_HOST
  strcpy(&hostname[0][0], name);
  hostnameknt = 1;
#else
  if (hostnameknt < MAXSENDTO){
    strcpy(&hostname[hostnameknt][0], name);
    hostnameknt++;
  } else {
    fprintf(stderr,"No more hosts allowed\n");
  }
#endif
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::addHost()
 *
 *  Description:
 *	
 *     Set the name of the target user and his/her hostname.
 *	
 *  Parameters:
 *     char* value   :  user@hostname string
 *	
 *----------------------------------------------------------------------*/

void O1kNetwork::setUserHost(char* value)
{
  char  hname[MAX_NAME_SIZE];
  char* p;

  /* this is in form user@hostname */  
  if (verbose) printf("o1k_conf_set_address(%s)\n",value);
  strcpy(destination, value);

  /* set hostname */
  p = (char*)strchr(value,'@');
  if (p){
    p++;
    strcpy(hname, p);
    addHost(hname);
  } else {
    fprintf(stderr,"Invalid user@host name %s\n",value);
  }

}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::stopTransmit()
 *
 *  Description:
 *     Stops the worker threads of the selected network types and
 *     then closes the network transmit state for these types.
 *	
 *  Parameters:
 *     int netids    : mask ids for selecting network types
 *	
 *  Returns:
 *     Currently always returns 0.
 *	
 *----------------------------------------------------------------------*/

int O1kNetwork::stopTransmit(int netids)
{
  if (verbose) printf("o1k_conf_network_stop_transmit %x\n",netids);
  
  if ((netids & NET_ID_VIDEO) && tstate[NET_TYPE_VIDEO]){
    vid_thread_t->stop();
    close_transmit(tstate[NET_TYPE_VIDEO]);
    tstate[NET_TYPE_VIDEO] = NULL;    
  }
  if ((netids & NET_ID_AUDIO) && tstate[NET_TYPE_AUDIO]){
    aud_thread_t->stop();
    close_transmit(tstate[NET_TYPE_AUDIO]);
    tstate[NET_TYPE_AUDIO] = NULL;    
  }
  if ((netids & NET_ID_CNTRL) && tstate[NET_TYPE_CNTRL]){
    ctl_thread_t->stop();
    close_transmit(tstate[NET_TYPE_CNTRL]);
    tstate[NET_TYPE_CNTRL] = NULL;    
  }
  
  return 0;  
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::startTransmit()
 *
 *  Description:
 *     Initializes the transmit state for the selected network types
 *     and starts the worker threads for these types.
 *	
 *  Parameters:
 *     int netids    : mask ids for selecting network types
 *	
 *  Returns:
 *      0  on success.
 *     -1  on error.
 *	
 *----------------------------------------------------------------------*/

int O1kNetwork::startTransmit(int netids)
{
  if (verbose) printf("o1k_conf_network_start_transmit %x\n",netids);
  
  if (netids & NET_ID_VIDEO){
    if (tstate[NET_TYPE_VIDEO]){
      fprintf(stderr,"Start Video Transmit Failing: already channel open\n");
      return -1;
    }
    tstate[NET_TYPE_VIDEO] = initialize_transmit(this, NET_TYPE_VIDEO);
    vid_thread_t->start(tNetThrdV, this);
  }
  if (netids & NET_ID_AUDIO){
    if (tstate[NET_TYPE_AUDIO]){
      fprintf(stderr,"Start Audio Transmit Failing: already channel open\n");
      return -1;
    }
    tstate[NET_TYPE_AUDIO] = initialize_transmit(this, NET_TYPE_AUDIO);
    aud_thread_t->start(tNetThrdA, this);
  }
  if (netids & NET_ID_CNTRL){
    if (tstate[NET_TYPE_CNTRL]){
      fprintf(stderr,"Start Cntrl Transmit Failing: already channel open\n");
      return -1;
    }
    tstate[NET_TYPE_CNTRL] = initialize_transmit(this, NET_TYPE_CNTRL);
    ctl_thread_t->start(tNetThrdC, this);
  }
  
  return 0;
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::stopReceive()
 *
 *  Description:
 *     Stops the worker threads of the selected network types and
 *     then closes the network receive state for these types.
 *	
 *  Parameters:
 *     int netids    : mask ids for selecting network types
 *	
 *  Returns:
 *     Currently always returns 0.
 *	
 *----------------------------------------------------------------------*/

int O1kNetwork::stopReceive(int netids)
{
  if (verbose) printf("o1k_conf_network_stop_receive %x\n",netids);
  
  if ((netids & NET_ID_VIDEO) && rstate[NET_TYPE_VIDEO]){
    vid_thread_r->stop();
    close_receive(rstate[NET_TYPE_VIDEO]);
    rstate[NET_TYPE_VIDEO] = NULL;    
  }
  if ((netids & NET_ID_AUDIO) && rstate[NET_TYPE_AUDIO]) {
    aud_thread_r->stop();
    close_receive(rstate[NET_TYPE_AUDIO]);
    rstate[NET_TYPE_AUDIO] = NULL;    
  }
  if ((netids & NET_ID_CNTRL) && rstate[NET_TYPE_CNTRL]) {
    ctl_thread_r->stop();
    close_receive(rstate[NET_TYPE_CNTRL]);
    rstate[NET_TYPE_CNTRL] = NULL;    
  }
  
  return 0;
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::startReceive()
 *
 *  Description:
 *     Initializes the receive state for the selected network types
 *     and starts the worker threads for these types.
 *	
 *  Parameters:
 *     int netids    : mask ids for selecting network types
 *	
 *  Returns:
 *      0  on success.
 *     -1  on error.
 *	
 *----------------------------------------------------------------------*/

int O1kNetwork::startReceive(int netids)
{
  if (verbose) printf("o1k_conf_network_start_receive %x\n",netids);
  
  if (netids & NET_ID_VIDEO){
    if (rstate[NET_TYPE_VIDEO]){
      fprintf(stderr,"Start Video Receive Failing: already channel open\n");
      return -1;
    }
    rstate[NET_TYPE_VIDEO] = initialize_receive(this, NET_TYPE_VIDEO);
    vid_thread_r->start(rNetThrdV, this);
  }
  if (netids & NET_ID_AUDIO){
    if (rstate[NET_TYPE_AUDIO]){
      fprintf(stderr,"Start Audio Receive Failing: already channel open\n");
      return -1;
    }
    rstate[NET_TYPE_AUDIO] = initialize_receive(this, NET_TYPE_AUDIO);
    aud_thread_r->start(rNetThrdA, this);
  }
  if (netids & NET_ID_CNTRL){
    if (rstate[NET_TYPE_CNTRL]){
      fprintf(stderr,"Start Cntrl Receive Failing: already channel open\n");
      return -1;
    }
    rstate[NET_TYPE_CNTRL] = initialize_receive(this, NET_TYPE_CNTRL);
    ctl_thread_r->start(rNetThrdC, this);
  }
  
  return 0;  
}


/*------------------------------------------------------------------------
 *
 *  Function:	o1k_conf_network_receive()
 *
 *  Description:
 *     This routine is used to receive data from the network. The caller
 *     supplies a buffer into which received data may be placed. The size
 *     of this received data is also returned via parameter reference.
 *     The caller must specify the network type it wants to receive from
 *     (i.e. NET_TYPE_VIDEO or NET_TYPE_AUDIO or NET_TYPE_CNTRL). A
 *     timeout value is also supplied to inform this routine how long it
 *     should wait for data to arrive from the network before returning.
 *     The timeout is in msecs. A timeout value of zero implies just to
 *     check to see if data is avail and it not, return immediately.
 *	
 *  Parameters:
 *     unsigned char* data  : buffer to place received data
 *     unsigned int* size   : size of data return in this variable
 *     unsigned int timeout : timeout value in msecs
 *     int type             : single network type 
 *	
 *  Returns:
 *     A return value of 0 implies data was recieved and placed into
 *     the passed in buffer. Otherwise, a value other than zero is
 *     returned meaning no data was avail by the end of the timeout
 *     interval.
 *
 *  Notes:
 *     This routine only supports a single caller for a given network
 *     type. One can not have more than one thread asking for data of
 *     a given type.
 *
 *     Currently the data received via the network is copied into the
 *     caller's data buffer. One might consider rewriting this code
 *     (specifically the sema post operation) such that this copy is
 *     no longer needed. A reference to the data could be returned to
 *     the caller who could then use it directly. The caller would then
 *     be responsible for posting the free buffer when done with the
 *     reference. Possibly a 'done_receive' routine could be added
 *     that would do the sema post.
 *
 *----------------------------------------------------------------------*/

int o1k_conf_network_receive(unsigned char* data, unsigned int* size, unsigned int timeout, int type)
{
  int tc = 1;
  int rc = 0;
  O1kThread* thread;

  /* set thread based on type */
  switch(type)
  {
    case NET_TYPE_VIDEO:
    {
      thread = conf->network->vid_thread_r;
      break;
    }
    case NET_TYPE_AUDIO:
    {
      thread = conf->network->aud_thread_r;
      break;
    }
    case NET_TYPE_CNTRL:
    {
      thread = conf->network->ctl_thread_r;
      break;
    }
    default:
    {
      fprintf(stderr,"o1k_conf_network_receive: invalid type %d\n",type);
      return -1;
    }
  }

  /* timeout is in ms */  
  if (timeout){
    tc = (timeout/5) + 1;
  }

  /*
   * Wait for full buffer of received data. Stop waiting if data
   * becomes avail, the timeout expires, or the network thread
   * responsible for supplying the data becomes disabled.
   */
  do {
    rc = sema_trywait(&rbuf_sem[SEMA_FULL][type]);
    tc--;
    if (rc != 0 && tc)
      poll(0,0,5);
  } while (rc != 0 && tc && thread->isEnabled());

  /* If rc == 0, data is available */
  if (rc == 0){
    /* copy data out of buffer */
    *size = rnetdata[rring[1][type]][type]->size;
    memcpy(data, rnetdata[rring[1][type]][type]->data, *size);

    /* post buffer as empty */
    sema_post(&rbuf_sem[SEMA_EMPTY][type]);
    rring[1][type]++;
    if (rring[1][type] == NET_NUM_DATA_BUFS)
      rring[1][type] = 0;
  } 

  return rc;
}

/*------------------------------------------------------------------------
 *
 *  Function:	o1k_conf_network_transmit()
 *
 *  Description:
 *     This routine is used to transmit data to the network. The caller
 *     supplies a buffer of data to be transmitted as well as the size of
 *     this data. The caller must specify the network type it wants to
 *     transmit to. A timeout value is also supplied to inform this routine
 *     how long it should wait for a free network transmit buffer data to
 *     become avail before returning. The timeout is in msecs. A timeout
 *     value of zero implies just to check to see if a network buffer is
 *     avail and it not, return immediately.
 *	
 *  Parameters:
 *     unsigned char* data  : buffer of data to send
 *     unsigned int* size   : size of data 
 *     unsigned int timeout : timeout value in msecs
 *     int type             : single network type 
 *	
 *  Returns:
 *     A return value of 0 implies data was copied into a network
 *     transmit buffer and thus is queued for transmittion by the
 *     responsible network worker thread. A non-zero value means
 *     a free network transmit buffer did not become avail in the
 *     timeout interval.
 *
 *  Notes:
 *     This routine only supports a single caller for a given network
 *     type. One can not have more than one thread transmitting data 
 *     of a given type.
 *
 *     Currently the data transmitted via the network is copied from 
 *     the caller's data buffer. One might consider rewriting this code
 *     (specifically the sema post operation) such that this copy is
 *     no longer needed. 
 *
 *----------------------------------------------------------------------*/

int o1k_conf_network_transmit(unsigned char* data, unsigned int size, unsigned int timeout, int type)
{
  int rc;
  int tc = 1;
  
  /* timeout is in ms */  
  if (timeout){
    tc = (timeout/5) + 1;
  }

  /* wait for empty buffer to place data */  
  do {
    rc = sema_trywait(&tbuf_sem[SEMA_EMPTY][type]);
    tc--;
    if (rc != 0 && tc)
      poll(0,0,5);
  } while (rc != 0 && tc);

  if (rc == 0){
    /* copy data to buffer */
    memcpy(tnetdata[tring[1][type]][type]->data, data, size);
    tnetdata[tring[1][type]][type]->size = size;
    /* post buffer as full */
    sema_post(&tbuf_sem[SEMA_FULL][type]);
    tring[1][type]++;
    if (tring[1][type] == NET_NUM_DATA_BUFS)
      tring[1][type] = 0;
  }

  return rc;
}

/*------------------------------------------------------------------------
 *
 *  Function:	O1kNetwork::getStatus
 *
 *  Description:
 *     Returns a string represent the interger value of the
 *     specified request status type.
 *	
 *  Parameters:
 *     NetworkStatus ns  : desired status info requested
 *	
 *  Returns:
 *     char* : pointer to status info
 *	
 *  Notes:
 *     The caller should copy the returned status info for it 
 *     is only valid until the next call to this routine.
 *	
 *----------------------------------------------------------------------*/

char* O1kNetwork::getStatus(NetworkStatus ns)
{
  static char status_buf[10];

  switch (ns)
  {
    case NS_PACKETS_VT:
    {
      sprintf(status_buf,"%d",status.packets_t_v);
      break;
    }
    case NS_PACKETS_VR:
    {
      sprintf(status_buf,"%d",status.packets_r_v);
      break;
    }
    case NS_DROPS_V:
    {
      sprintf(status_buf,"%d",status.drops_v);
      break;
    }
    case NS_PACKETS_AT:
    {
      sprintf(status_buf,"%d",status.packets_t_a);
      break;
    }
    case NS_PACKETS_AR:
    {
      sprintf(status_buf,"%d",status.packets_r_a);
      break;
    }
    case NS_DROPS_A:
    {
      sprintf(status_buf,"%d",status.drops_a);
      break;
    }
    default:
    {
      fprintf(stderr,"Unknown network status query %d\n",ns);
      strcpy(status_buf,"");
    }
  }
  return status_buf;
}

/*------------------------------------------------------------------------
 *
 *  Function:	xNetThrdX
 *
 *  Description:
 *	
 *     The following routines embody the various network worker
 *     threads. Each routine/thread is passed in as its sole
 *     parameter a pointer to the main O1kNetwork object. All
 *     of the receive routines are pretty much the same and with
 *     a little work could be made to be identical. The same is
 *     true of the transmit routines/threads.
 *     
 *     The 'transmit' threads all have pretty much the same format:
 *
 *      - a local O1kNetwork varible is set (from single parameter)
 *      - a local O1kThread variable is set
 *      - a local network 'type' variable is set
 *      - the ring buffer indices for the type are reset
 *      - the thread is marked as not being 'stopped'
 *      - the thread loops while the thread is enabled
 *         - wait for a network transmit buffer to be filled
 *         - break if the thread is no longer enabled
 *         - transmit the data
 *         - semaphore post the transmit buffer as being empty
 *      - when leaving loop, mark thread as being 'stopped'
 *     
 *	
 *     The 'receive' threads all have pretty much the same format. 
 *     Steps marked with a x are for audio/video threads only.
 *
 *      - a local O1kNetwork varible is set (from single parameter)
 *      - a local O1kThread variable is set
 *      - a local network 'type' variable is set
 *      - the ring buffer indices for the type are reset
 *      - initialize ReceivePackets routine
 *      - the thread is marked as not being 'stopped'
 *      x wait for the corresponding transmit thread to start
 *      x start the corresponding audio/video object
 *      - the thread loops while the thread is enabled
 *         - wait for an empty network receive buffer
 *         - break if the thread is no longer enabled
 *         - receive data from the network
 *         - semaphore post the receive buffer as being full
 *      x when leaving loop, stop the corresponding audio/video
 *         object
 *      - mark thread as being 'stopped'
 *     	
 *
 *  Notes:
 *     
 *     From the above descriptions you might have noticed the
 *     "lock-step" nature of starting the various components
 *     of a conference - at least from the perspetive of the 
 *     network threads. For both audio and video, the main
 *     portion of the receive threads wait for the transmit
 *     threads to start. Once it is known that the transmit
 *     threads have started, then the O1kAudio and O1kVideo
 *     objects of the conference are started, This illustrates
 *     the dependancies of these objects/threads on one another
 *     in this example application.
 *      
 *----------------------------------------------------------------------*/

/*****************************************************************/
/** N E T W O R K  :  T R A N S M I T   A U D I O   T H R E A D **/
/*****************************************************************/

void* tNetThrdA(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->aud_thread_t;
  
  int type = NET_TYPE_AUDIO; /* thread index type */
  int rc;
  
  /* vars for debugging, tracing only */
  FILE* tfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  tring[0][type] = tring[1][type] = 0; /* ring buffer index */
  
  /* debug shadow file open */
  if (AUDIO_WRITE_SHADOW)
    tfile = fopen("/tmp/atra","w");
  
  if (verbose) printf("***** Network Transmit tNetThrdA Entering\n");
  
  /* loop while enabled */
  thread->setStopped(FALSE);
  while (thread->isEnabled()){

    /* get buffer to send to network, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&tbuf_sem[SEMA_FULL][type]);
      
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;

    /* transmit control data */
    transmit_data(tnetdata[tring[0][type]][type]->data,
                  tnetdata[tring[0][type]][type]->size,
                  network->tstate[type], network,
                  type, 0);

    network->status.packets_t_a++;
    
    /* debug shadow file write */
    if (tfile){
      fwrite(tnetdata[tring[0][type]][type]->data,
             tnetdata[tring[0][type]][type]->size, 1, tfile);
      fflush(tfile);
    }

    /* post buffer free */
    sema_post(&tbuf_sem[SEMA_EMPTY][type]);
    tring[0][type]++;
    if (tring[0][type] == NET_NUM_DATA_BUFS)
      tring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Transmit tNetThrdA Exiting *****\n");
  thread->setStopped(TRUE);

  return 0;
}

/*****************************************************************/
/** N E T W O R K  :  R E C E I V E     A U D I O   T H R E A D **/
/*****************************************************************/

void* rNetThrdA(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->aud_thread_r;
  
  BigPacketType* bigpacket;  /* tmp packet for receiving data */
  int type = NET_TYPE_AUDIO; /* thread index type */
  int rc;

  /* vars for debugging, tracing only */
  FILE* rfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  rring[0][type] = rring[1][type] = 0; /* ring buffer index */

  /* debug shadow file open */
  if (AUDIO_WRITE_SHADOW)
    rfile = fopen("/tmp/arec","w"); 

  /* init ReceivePackets routine */
  ReceivePackets(network, thread, type,1);
  
  if (verbose) printf("***** Network Receive rNetThrdA Entering\n");
  thread->setStopped(FALSE);

  /* wait for audio net transmit to start */
  while (!network->aud_thread_t->isEnabled() && thread->isEnabled());
  if (!thread->isEnabled()){
    if (verbose) printf("***** Network Receive rNetThrdA Exiting Early *****\n");
    thread->setStopped(TRUE);
    return 0;
  }
  
  /* now start audio device */
  conf->audio->start();

  /* loop while enabled */  
  while (thread->isEnabled()){
    
    /* get buffer for network data, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&rbuf_sem[SEMA_EMPTY][type]);
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;
    
    /* get control data from network */
   if ((bigpacket = ReceivePackets(network, thread, type,0)) == NULL)
      break;

    network->status.packets_r_a++;
    
    /* copy packet to data buffer */
    memcpy(rnetdata[rring[0][type]][type]->data, bigpacket->data, bigpacket->header.size);
    rnetdata[rring[0][type]][type]->size = bigpacket->header.size;

    /* debug shadow file write */
    if (rfile){
      fwrite(rnetdata[rring[0][type]][type]->data,
             rnetdata[rring[0][type]][type]->size, 1, rfile);
      fflush(rfile);
    }

    /* post data avail */
    sema_post(&rbuf_sem[SEMA_FULL][type]);
    rring[0][type]++;
    if (rring[0][type] == NET_NUM_DATA_BUFS)
      rring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Receive rNetThrdA Exiting *****\n");
  conf->audio->stop();
  thread->setStopped(TRUE);
  
  return 0;
}

/*****************************************************************/
/* N E T W O R K  :  T R A N S M I T  C O N T R O L  T H R E A D */
/*****************************************************************/

void* tNetThrdC(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->ctl_thread_t;
  
  int type = NET_TYPE_CNTRL; /* thread index type */
  int rc;
  
  /* vars for debugging, tracing only */
  FILE* tfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  tring[0][type] = tring[1][type] = 0; /* ring buffer index */
  
  /* debug shadow file open */
  if (CONTROL_WRITE_SHADOW)
    tfile = fopen("/tmp/ctra","w");
  
  if (verbose) printf("***** Network Transmit tNetThrdC Entering\n");
  
  /* loop while enabled */
  thread->setStopped(FALSE);
  while (thread->isEnabled()){

    /* get buffer to send to network, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&tbuf_sem[SEMA_FULL][type]);
      
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;

    /* transmit control data */
    transmit_data(tnetdata[tring[0][type]][type]->data,
                  tnetdata[tring[0][type]][type]->size,
                  network->tstate[type], network,
                  type, 0);
        
    /* debug shadow file write */
    if (tfile){
      fwrite(tnetdata[tring[0][type]][type]->data,
             tnetdata[tring[0][type]][type]->size, 1, tfile);
      fflush(tfile);
    }

    /* post buffer free */
    sema_post(&tbuf_sem[SEMA_EMPTY][type]);
    tring[0][type]++;
    if (tring[0][type] == NET_NUM_DATA_BUFS)
      tring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Transmit tNetThrdC Exiting *****\n");
  thread->setStopped(TRUE);

  return 0;
}

/*****************************************************************/
/** N E T W O R K  :  R E C E I V E  C O N T R O L  T H R E A D **/
/*****************************************************************/

void* rNetThrdC(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->ctl_thread_r;
  
  BigPacketType* bigpacket;   /* tmp packet for receiving data */
  int type = NET_TYPE_CNTRL; /* thread index type */
  int rc;
  
  /* vars for debugging, tracing only */
  FILE* rfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  rring[0][type] = rring[1][type] = 0; /* ring buffer index */
  
  /* debug shadow file open */
  if (CONTROL_WRITE_SHADOW)
    rfile = fopen("/tmp/crec","w"); 

  /* init ReceivePackets routine */
  ReceivePackets(network, thread, type,1);

  if (verbose) printf("***** Network Receive rNetThrdC Entering\n");

  thread->setStopped(FALSE);
  
  /* loop while enabled */
  while (thread->isEnabled()){

    /* get buffer for network data, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&rbuf_sem[SEMA_EMPTY][type]);
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;
    
    /* get control data from network */
    bigpacket = ReceivePackets(network, thread, type,0);
    if (bigpacket == NULL)
      break;

    /* copy packet to data buffer */
    memcpy(rnetdata[rring[0][type]][type]->data, bigpacket->data, bigpacket->header.size);
    rnetdata[rring[0][type]][type]->size = bigpacket->header.size;

    if (verbose) printf("control received data %d\n", bigpacket->header.size);
            
    /* debug shadow file write */
    if (rfile){
      fwrite(rnetdata[rring[0][type]][type]->data,
             rnetdata[rring[0][type]][type]->size, 1, rfile);
      fflush(rfile);
    }

    /* post data avail */
    sema_post(&rbuf_sem[SEMA_FULL][type]);
    rring[0][type]++;
    if (rring[0][type] == NET_NUM_DATA_BUFS)
      rring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Receive rNetThrdC Exiting *****\n");
  thread->setStopped(TRUE);

  return 0;
}

/*****************************************************************/
/**  N E T W O R K  :  T R A N S M I T  V I D E O  T H R E A D  **/
/*****************************************************************/

void* tNetThrdV(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->vid_thread_t;
  
  int type = NET_TYPE_VIDEO; /* thread index type */
  int rc;
  
  /* vars for debugging, tracing only */
  FILE* tfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  tring[0][type] = tring[1][type] = 0; /* ring buffer index */
  
  /* debug shadow file open */
  if (VIDEO_WRITE_SHADOW)
    tfile = fopen("/tmp/vtra","w");

  if (verbose) printf("***** Network Transmit tNetThrdV Entering\n");
  
  /* loop while enabled */
  thread->setStopped(FALSE);
  while (thread->isEnabled()){

    /* get buffer to send to network, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&tbuf_sem[SEMA_FULL][type]);
      
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;
    
    /* transmit video data */
    
    transmit_data(tnetdata[tring[0][type]][type]->data,
                  tnetdata[tring[0][type]][type]->size,
                  network->tstate[type], network,
                  type, 0);

    network->status.packets_t_v++;
    
    /* debug shadow file write */
    if (tfile){
      fwrite(tnetdata[tring[0][type]][type]->data,
             tnetdata[tring[0][type]][type]->size, 1, tfile);
      fflush(tfile);
    }

    /* post buffer free */
    sema_post(&tbuf_sem[SEMA_EMPTY][type]);
    tring[0][type]++;
    if (tring[0][type] == NET_NUM_DATA_BUFS)
      tring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Transmit tNetThrdV Exiting *****\n");
  thread->setStopped(TRUE);
  return 0;
}


/*****************************************************************/
/**  N E T W O R K  :  R E C E I V E   V I D E O   T H R E A D  **/
/*****************************************************************/

void* rNetThrdV(void* vp)
{
  O1kNetwork* network = (O1kNetwork*)vp;
  O1kThread*  thread  = network->vid_thread_r;
  
  BigPacketType* bigpacket;  /* tmp packet for receiving data */
  int type = NET_TYPE_VIDEO; /* thread index type */
  int rc;

  /* vars for debugging, tracing only */
  FILE* rfile = NULL;

  /* see readme */
  if (register_trap)
    asm("ta 6");
  
  rring[0][type] = rring[1][type] = 0; /* ring buffer index */

  /* debug shadow file open */
  if (VIDEO_WRITE_SHADOW)
    rfile = fopen("/tmp/vrec","w"); 

  /* init ReceivePackets routine */
  ReceivePackets(network, thread, type,1);
  
  if (verbose) printf("***** Network Receive rNetThrdV Entering\n");
  thread->setStopped(FALSE);

  /* wait for video net transmit to start */
  while (!network->vid_thread_t->isEnabled() && thread->isEnabled());
  if (!thread->isEnabled()){
    if (verbose) printf("***** Network Receive rNetThrdV Exiting Early *****\n");
    thread->setStopped(TRUE);
    return 0;
  }
  
  /* now start video device */
  conf->video->start();

  /* loop while enabled */
  while (thread->isEnabled()){
    
    /* get buffer for network data, timeout if thread disabled */
    do {
      /* sleep for a little */
      poll(0,0,5);
      rc = sema_trywait(&rbuf_sem[SEMA_EMPTY][type]);
    } while (rc != 0 && thread->isEnabled());

    if (rc != 0)
      break;
    
    /* get control data from network */
    if ((bigpacket = ReceivePackets(network, thread, type,0)) == NULL)
      break;

    network->status.packets_r_v++;
    
    /* copy packet to data buffer */
    memcpy(rnetdata[rring[0][type]][type]->data, bigpacket->data, bigpacket->header.size);
    rnetdata[rring[0][type]][type]->size = bigpacket->header.size;

    /* debug shadow file write */
    if (rfile){
      fwrite(rnetdata[rring[0][type]][type]->data,
             rnetdata[rring[0][type]][type]->size, 1, rfile);
      fflush(rfile);
    }

    /* post data avail */
    sema_post(&rbuf_sem[SEMA_FULL][type]);
    rring[0][type]++;
    if (rring[0][type] == NET_NUM_DATA_BUFS)
      rring[0][type] = 0;
  }
  
  /* leaving terminates the thread */
  if (verbose) printf("***** Network Receive rNetThrdV Exiting *****\n");
  conf->video->stop();
  thread->setStopped(TRUE);
  return 0;
}

/*------------------------------------------------------------------------
 *
 *  Function:	packetnum()
 *
 *  Description:
 *     Returns packet number from packet flags data
 *	
 *  Parameters:
 *     unsigned short flag   : packet flag data
 *	
 *  Returns:
 *     packet number
 *	
 *----------------------------------------------------------------------*/

int packetnum(unsigned short flag)
{
  int count = 0;
  
  while ((flag & 0x1) == 1) {
    count++;
    flag >>= 1;
  }
  return count;
}

/*------------------------------------------------------------------------
 *
 *  Function:	create_big_packet()
 *
 *  Description:
 *     Takes a list of packets and returns a single packet 
 *	
 *  Parameters:
 *     BigPacketType * big      : pointer to packet to place data
 *     PacketType *packetlist[] : list of packets
 *
 *----------------------------------------------------------------------*/

void create_big_packet(BigPacketType * big, PacketType *packetlist[])
{
  int i;
  unsigned char *curloc;
  int totsize = 0;
  int size;
  
  curloc = big->data;
  big->header = packetlist[0]->header;
  for (i = 0; i < MAXPACKETS; i++) {
    if (packetlist[i]) {
      size = packetlist[i]->header.size;
      memcpy(curloc, packetlist[i]->data, size);
      totsize += size;
      curloc += size;
      big->header.tstamp = packetlist[i]->header.tstamp;
    }
  }
  big->header.size = totsize;
}

/*------------------------------------------------------------------------
 *
 *  Function:	ReceivePackets()
 *
 *  Description:
 *     This routine is called by the various network receive threads
 *     to receive data from the network. When a chuck of data (which
 *     we will call a 'frame' of data) is sent to be transmitted it
 *     might be broken up into smaller packets. This routine is 
 *     responsible for reading these packets from the network and
 *     created a single 'frame' of data from it (which is returned
 *     as a BigPacketType). 
 *     
 *     If a frame is broken up into multiple packets then from this
 *     routines point of view, they must arrive sequentially in 
 *     order to be recombined. If during the processing of a frame,
 *     a packet arrives indicating that it is from a different future
 *     frame, then the current frame is dropped.
 *     
 *     This routine maintains a bunch of static data for the various
 *     state of the various network types. A single call to this
 *     routine with the init parameter set to 1 will initialize
 *     the state of this data. 
 *     
 *  Parameters:
 *     O1kNetwork* network : main network object
 *     O1kThread* thread   : receiving thread
 *     int type            : type to receive
 *     int init            : init flag
 *     
 *  Returns:
 *     A BigPacketType* is returned with the received data. NULL is
 *     returned if nothing was received. The caller must use the
 *     data returned before calling this routine again.
 *     
 *----------------------------------------------------------------------*/

BigPacketType* ReceivePackets(O1kNetwork* network, O1kThread* thread, int type, int init)
{
  static PacketType  packets[NET_MAX_TYPES][MAXPACKETS];    /* circular buffer of rec'vd packets */
  static PacketType* packetlist[NET_MAX_TYPES][MAXPACKETS]; /* packets for this frame */
  static PacketType* curpacket[NET_MAX_TYPES];
  static int packetcnt[NET_MAX_TYPES];
  static unsigned int currentsession[NET_MAX_TYPES];
  static int currentframe[NET_MAX_TYPES];
  static unsigned short myflags[NET_MAX_TYPES];
  static BigPacketType bigpacket[NET_MAX_TYPES];
  
  int i;
  int frame_complete = 0;

  if (init){
    packetcnt[type] = 0;
    return NULL;
  }
  
  while (!frame_complete && thread->isEnabled()){
    curpacket[type] = &packets[type][packetcnt[type]++];
    if (packetcnt[type] >= MAXPACKETS)
      packetcnt[type] = 0;
    
    /* Get packet if available*/
    receive_packet(network->rstate[type], curpacket[type], thread);

    if (!thread->isEnabled())
      return NULL;

    if (currentsession[type] != curpacket[type]->header.session) {
      
      /*
       * if you join a conference in progress then the current frame
       * and session will be grabbed from the first received packet
       * received
       */      
      for (i = 0; i < MAXPACKETS; i++)
        packetlist[type][i] = NULL;
      myflags[type] = 0xffff;
      currentframe[type] = curpacket[type]->header.frame;
      currentsession[type] = curpacket[type]->header.session;

      if (verbose) {
        fprintf(stderr, "[%d] changing session from %d to %d\n",type,
                currentsession[type], curpacket[type]->header.session);
      }
    }

    if (curpacket[type]->header.frame < currentframe[type]) {
      /*
       * if this packet is from an earlier frame disregard it
       */
      if (verbose)
        fprintf(stderr, "##### [%d] received stale packet %d, current frame %d\n",type, curpacket[type]->header.frame, currentframe[type]);
      
      continue;
    }

    if (curpacket[type]->header.frame > currentframe[type]) {
      
      /*
       * If the frame number increased then we probably missed the last
       * packet of the previous frame. We drop the partial frame.
       */
      if (myflags[type] != 0) {
        int ndropped = curpacket[type]->header.frame - currentframe[type];

        if (verbose)
          fprintf(stderr, ">>>>> [%d] dropping %d due to getting packet %d (flags %x) %d frames dropped\n",type, currentframe[type], curpacket[type]->header.frame, myflags[type], ndropped);
        
        if (type == NET_TYPE_AUDIO)
          network->status.drops_a += ndropped;
        if (type == NET_TYPE_VIDEO)
          network->status.drops_v += ndropped;
      } else {
        /*
         * This is a start of a new frame
         */
        if (curpacket[type]->header.frame > currentframe[type]+1){
          /*
           * The new frame is more than one frame after the previous frame
           */
          int ndropped = curpacket[type]->header.frame - currentframe[type];

          if (verbose)
            fprintf(stderr, "***** [%d] packet %d did not arrive got packet %d last packet %d %d frames dropped\n",type, currentframe[type]+1, curpacket[type]->header.frame, currentframe[type], ndropped);
          
          if (type == NET_TYPE_AUDIO)
            network->status.drops_a += ndropped;
          if (type == NET_TYPE_VIDEO)
            network->status.drops_v += ndropped;
        }
      }
      
      for (i = 0; i < MAXPACKETS; i++)
        packetlist[type][i] = NULL;
      myflags[type] = 0xffff;
      currentframe[type] = curpacket[type]->header.frame;
    }

    /*
     * save the packet for decoding
     */
#if 0
    printf("[%d] save the packet[%d] f[%d] for decoding %d\n",type,
           packetnum(curpacket[type]->header.flags),
           curpacket[type]->header.frame,
           curpacket[type]->header.size); fflush(stdout);
#endif
    
    packetlist[type][packetnum(curpacket[type]->header.flags)]= curpacket[type];
    myflags[type] &= curpacket[type]->header.flags;

    if (myflags[type] == 0) {
      frame_complete = 1;
      create_big_packet(&bigpacket[type], &packetlist[type][0]);
    }
  }
  return &bigpacket[type];
}

