mctxrx.h

Go to the documentation of this file.

/*
** Copyright (C) 2004 Russell Leighton
**
** This program 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.
**
** This program 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 this program; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*/


#ifndef __AWHANDLER_MCTXRX__
#define __AWHANDLER_MCTXRX__

#include "bitvector.h"

/* shared data structures between tx and rx */

/* default max tx rate in Mbps */
#define AW_MCTXRX_DEFAULT_MAX_TX_RATE 80

/* default ip and port  */
#define AW_MCTXRX_DEFAULT_MCAST_GROUP "224.0.0.1"
#define AW_MCTXRX_DEFAULT_MCAST_PORT 1800
#define AW_MCTXRX_DEFAULT_RETX_PORT 1801

/* default re-transmit timeout */
#define AW_MCTX_DEFAULT_RETX_TIMEOUT (3*AW_TIMESTAMP_PER_SEC)
#define AW_MCRX_DEFAULT_RETX_TIMEOUT (1*AW_TIMESTAMP_PER_SEC)

/* NOTE: version MUST change if you change msg structures */
#define AW_MCTXRX_MSG_VERSION 1

#define AW_MCTXRX_MAX_PATHLEN 256

/* NOTE: all integers sent in "network" byte order */

/* basic message header */
typedef struct {
    
  /* version of message format */
  u_int16_t
   version;

  /* unique (per tx) transaction id, this plus tx ip address should be unique */
  u_int32_t
   xtnid;

  /* block of file xfer, if 0 is info block */
  int32_t
   blockid;

  /* size in bytes of the payload data that follows */
  u_int16_t
   payload_size;

  /* ptr to start of payload */
  void
   *payload;

} aw_mctxrx_msghdr_t;

/* meta info for file xfer */
typedef struct {

  /* string holding path of file */
  byte_t
   filename[AW_MCTXRX_MAX_PATHLEN];

  /* file size */
  u_int64_t
   filesize;

  /* number of total blocks in xfer */
  u_int32_t
   nblocks;

  /* blocksize in xfer EXCEPT last block */
  u_int16_t
   blocksize;

  /* port sender is listening for retx packets */
  u_int16_t
   retxport;

} aw_mctxrx_info_t;

/* safe max payload, slightly smaller than it needs to be...
   account for header plus some safety padding
 */
#define AW_MCTXRX_MAX_PAYLOAD_SIZE (  65000 - sizeof(aw_mctxrx_msghdr_t) - 8 )
#define AW_MCTXRX_DEFAULT_PAYLOAD_SIZE 27500

/* write info into message data, return message size */
inline static u_int32_t mctxrx_create_infomsg(u_int32_t xtnid, aw_mctxrx_info_t *info, byte_t *msg)
{

  /* version first 2 bytes */
  {
    *((u_int16_t *)msg) = htons(AW_MCTXRX_MSG_VERSION);
  }

  /* transaction id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 2)) = htonl(xtnid);
  }

  /* block id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 6)) = htonl(0);
  }

  /* payload size follows for 2 bytes */
  {
    /* sum the elements in the info struct, CAN'T use sizeof() due to potential padding */
    *((u_int16_t *)(msg + 10)) = htons( AW_MCTXRX_MAX_PATHLEN + 8 + 4 + 2);
  }

  /* filename follows for AW_MCTXRX_MAX_PATHLEN bytes */
  {
    strncpy(msg + 12, info->filename, AW_MCTXRX_MAX_PATHLEN);
  }

  /* filesize follows for 8 bytes */
  {
    *( (u_int64_t *)(msg + 12 + AW_MCTXRX_MAX_PATHLEN) ) = htonll(info->filesize);
  }

  /* number of blocks to be sent follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 12 + AW_MCTXRX_MAX_PATHLEN + 8)) = htonl(info->nblocks);
  }

  /* blocksize follows for 2 bytes */
  {
    /* sum the elements in the info struct, CAN'T use sizeof() due to potential padding */
    *((u_int16_t *)(msg + 12 + AW_MCTXRX_MAX_PATHLEN + 8 + 4)) = htons( info->blocksize );
  }

  /* retxport follows for 2 bytes */
  {
    /* sum the elements in the info struct, CAN'T use sizeof() due to potential padding */
    *((u_int16_t *)(msg + 12 + AW_MCTXRX_MAX_PATHLEN + 8 + 4 + 2)) = htons( info->retxport );
  }

  return 12 + AW_MCTXRX_MAX_PATHLEN + 8 + 4 + 2 + 2;
}

/* write data block into message data, return message size on success, else there was a read error */
inline static int32_t mctxrx_create_datamsg(u_int32_t xtnid, u_int32_t blockid, 
                                            u_int16_t payload_size, int32_t fd,
                                            byte_t *msg)
{
  /* version first 2 bytes */
  {
    *((u_int16_t *)msg) = htons(AW_MCTXRX_MSG_VERSION);
  }

  /* transaction id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 2)) = htonl(xtnid);
  }

  /* block id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 6)) = htonl(blockid);
  }

  /* payload size follows for 2 bytes */
  {
    *((u_int16_t *)(msg + 10)) = htons( payload_size );
  }

  if ( payload_size && read(fd, msg + 12, payload_size) != payload_size )
    return -1;
  else
    return 12 + payload_size ;

}

/* bitvector into message data, return message size on success */
inline static int32_t mctxrx_create_retxmsg(u_int32_t xtnid, 
                                            aw_bitvector_t *bv,
                                            byte_t *msg)
{
  int32_t
    payload_size = AW_BITVECTOR_GET_VECTOR_SIZE(bv) + sizeof(*bv);

  /* important to keep this in sync with bitvector.h */
  assert( sizeof(*bv) == sizeof(u_int32_t) );

  /* version first 2 bytes */
  {
    *((u_int16_t *)msg) = htons(AW_MCTXRX_MSG_VERSION);
  }

  /* transaction id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 2)) = htonl(xtnid);
  }

  /* block id follows for 4 bytes */
  {
    *((u_int32_t *)(msg + 6)) = htonl(~0); /* all 1's indicates not a valid block */
  }

  /* payload size follows for 2 bytes */
  {
    *((u_int16_t *)(msg + 10)) = htons( payload_size );
  }

  /* take each u_int32_t to network byte order */
  {
    u_int32_t
      i,
      nwords = payload_size / sizeof(u_int32_t),
      *words = (u_int32_t *)(msg + 12);

    /* copy data */
    memcpy(words, bv, payload_size );

    /* convert words to network order */
    for(i=0; i<nwords; i++) words[i] = htonl( words[i] );
  }

  return 12 + payload_size ;

}


/* read msg header, return 0 on success */
inline static int32_t mctxrx_extract_msghdr(aw_mctxrx_msghdr_t *hdr, byte_t *msg)
{

  /* version first 2 bytes */
  {
    hdr->version = ntohs( *((u_int16_t *)msg) );
  }

  /* validate */
  if (hdr->version != AW_MCTXRX_MSG_VERSION ) return -1;

  /* transaction id follows for 4 bytes */
  {
    hdr->xtnid = ntohl( *((u_int32_t *)(msg + 2)) );
  }

  /* block id follows for 4 bytes */
  {
    hdr->blockid = ntohl( *((u_int32_t *)(msg + 6)) );
  }

  /* payload size follows for 2 bytes */
  {
    hdr->payload_size = ntohs( *((u_int16_t *)(msg + 10)) );
  }

  /* set to rest of message */
  hdr->payload = msg + 12;

  return 0 ;
}

/* read info, return 0 on success */
inline static int32_t mctxrx_extract_info(aw_mctxrx_info_t *info, aw_mctxrx_msghdr_t *hdr)
{
  /* only 0 has info! */
  if ( hdr->blockid != 0 ) return -1;

  /* filename follows for AW_MCTXRX_MAX_PATHLEN bytes */
  {
    strncpy(info->filename, hdr->payload, AW_MCTXRX_MAX_PATHLEN);
    info->filename[AW_MCTXRX_MAX_PATHLEN -1] = 0; /* string safety! */
  }

  /* filesize follows for 8 bytes */
  {
    info->filesize = ntohll ( *( (u_int64_t *)(hdr->payload + AW_MCTXRX_MAX_PATHLEN) ) );
  }

  /* number of blocks to be sent follows for 4 bytes */
  {
    info->nblocks =  ntohl( *((u_int32_t *)(hdr->payload + AW_MCTXRX_MAX_PATHLEN + 8)) );
  }

  /* blocksize follows for 2 bytes */
  {
    info->blocksize =  ntohs( *((u_int16_t *)(hdr->payload + AW_MCTXRX_MAX_PATHLEN + 8 + 4)) );
  }

  /* retxport follows for 2 bytes */
  {
    info->retxport =  ntohs( *((u_int16_t *)(hdr->payload + AW_MCTXRX_MAX_PATHLEN + 8 + 4 + 2)) );
  }


  return 0 ;
}

/* extract a bitvector, dynamically allocated. Return NULL on failure */
inline static aw_bitvector_t *mctxrx_extract_retxmsg(aw_mctxrx_msghdr_t *hdr)
{
  aw_bitvector_t
    *bv = NULL;

  /* important to keep this in sync with bitvector.h */
  assert( sizeof(*bv) == sizeof(u_int32_t) );

  /* all 1's is a re-tx */
  if ( hdr->blockid != ~0 ) goto done;

  bv = aw_bitvector_create( 8*(hdr->payload_size - sizeof(*bv)) );
  if (bv == NULL) goto done;

  /* convert msg to host order and copy */
  {
    u_int32_t
      i,
      nwords = hdr->payload_size / sizeof(u_int32_t),
      *words = (u_int32_t *)(hdr->payload);

    /* convert words to network order */
    for(i=0; i<nwords; i++) words[i] = ntohl( words[i] );

    /* copy */
    *bv = *( (aw_bitvector_t *)(hdr->payload) );
    memcpy(AW_BITVECTOR_GET_VECTOR(bv), hdr->payload + sizeof(*bv), (nwords*sizeof(u_int32_t)) - sizeof(*bv));

  }

 done:;

  return bv;
}

#endif

Aware 0.11.1 Copyright (C) 1998-2005 Russell Leighton (russ@elegant-software.com)