mbus.c

网络MPEG4IP流媒体开发源代码

/*
 * FILE:     mbus.c
 * AUTHOR:   Colin Perkins
 * MODIFIED: Orion Hodson
 *           Markus Germeier
 * 
 * Copyright (c) 1997-2000 University College London
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, is permitted provided that the following conditions 
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *      This product includes software developed by the Computer Science
 *      Department at University College London
 * 4. Neither the name of the University nor of the Department may be used
 *    to endorse or promote products derived from this software without
 *    specific prior written permission.
 * THIS SOFTWARE IS PROVIDED BY THE AUTHORS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */

#include "config_unix.h"
#include "config_win32.h"
#include "debug.h"
#include "memory.h"
#include "net_udp.h"
#include "hmac.h"
#include "qfDES.h"
#include "base64.h"
#include "gettimeofday.h"
#include "vsnprintf.h"
#include "mbus.h"
#include "mbus_config.h"
#include "mbus_parser.h"
#include "mbus_addr.h"

#define MBUS_BUF_SIZE	  1500
#define MBUS_ACK_BUF_SIZE 1500
#define MBUS_MAX_ADDR	    10
#define MBUS_MAX_QLEN	    50 /* Number of messages we can queue with mbus_qmsg() */

#define MBUS_MAGIC	0x87654321
#define MBUS_MSG_MAGIC	0x12345678

struct mbus_msg {
	struct mbus_msg	*next;
	struct timeval	 send_time;	/* Time the message was sent, to trigger a retransmit */
	struct timeval	 comp_time;	/* Time the message was composed, the timestamp in the packet header */
	char		*dest;
	int		 reliable;
	int		 complete;	/* Indicates that we've finished adding cmds to this message */
	int		 seqnum;
	int		 retransmit_count;
	int		 message_size;
	int		 num_cmds;
	char		*cmd_list[MBUS_MAX_QLEN];
	char		*arg_list[MBUS_MAX_QLEN];
	uint32_t	 idx_list[MBUS_MAX_QLEN];
	uint32_t	 magic;		/* For debugging... */
};

struct mbus {
	socket_udp	 	 *s;
	char		 	 *addr;				/* Addresses we respond to. 					*/
	int		 	  max_other_addr;
	int		 	  num_other_addr;
	char			**other_addr;			/* Addresses of other entities on the mbus. 			*/
        struct timeval          **other_hello;                  /* Time of last mbus.hello we received from other entities      */
	int		 	  seqnum;
	struct mbus_msg	 	 *cmd_queue;			/* Queue of messages waiting to be sent */
	struct mbus_msg	 	 *waiting_ack;			/* The last reliable message sent, if we have not yet got the ACK */
	char		 	 *hashkey;
	int		 	  hashkeylen;
	char		 	 *encrkey;
	int		 	  encrkeylen;
	struct timeval	 	  last_heartbeat;		/* Last time we sent a heartbeat message */
	struct mbus_config	 *cfg;
	void (*cmd_handler)(char *src, char *cmd, char *arg, void *dat);
	void (*err_handler)(int seqnum, int reason);
	uint32_t		  magic;			/* For debugging...                                             */
	uint32_t		  index;
	uint32_t		  index_sent;
};

static void mbus_validate(struct mbus *m)
{
#ifdef DEBUG
	int	i;

	ASSERT(m->num_other_addr <= m->max_other_addr);
	ASSERT(m->num_other_addr >= 0);
	for (i = 0; i < m->num_other_addr; i++) {
		ASSERT(m->other_addr[i]  != NULL);
		ASSERT(m->other_hello[i] != NULL);
	}
	for (i = m->num_other_addr + 1; i < m->max_other_addr; i++) {
		ASSERT(m->other_addr[i]  == NULL);
		ASSERT(m->other_hello[i] == NULL);
	}
#endif
	ASSERT(m->magic == MBUS_MAGIC);
	xmemchk();
}

static void mbus_msg_validate(struct mbus_msg *m)
{
#ifdef DEBUG
	int	i;

	ASSERT((m->num_cmds < MBUS_MAX_QLEN) && (m->num_cmds >= 0));
	for (i = 0; i < m->num_cmds; i++) {
		ASSERT(m->cmd_list[i] != NULL);
		ASSERT(m->arg_list[i] != NULL);
		if (i > 0) {
			ASSERT(m->idx_list[i] > m->idx_list[i-1]);
		}
	}
	for (i = m->num_cmds + 1; i < MBUS_MAX_QLEN; i++) {
		ASSERT(m->cmd_list[i] == NULL);
		ASSERT(m->arg_list[i] == NULL);
	}	
	ASSERT(m->dest != NULL);
#endif
	ASSERT(m->magic == MBUS_MSG_MAGIC);
}

static void store_other_addr(struct mbus *m, char *a)
{
	/* This takes the address a and ensures it is stored in the   */
	/* m->other_addr field of the mbus structure. The other_addr  */
	/* field should probably be a hash table, but for now we hope */
	/* that there are not too many entities on the mbus, so the   */
	/* list is small.                                             */
	int	i;

	mbus_validate(m);

	for (i = 0; i < m->num_other_addr; i++) {
		if (mbus_addr_match(m->other_addr[i], a)) {
			/* Already in the list... */
			gettimeofday(m->other_hello[i],NULL);
			return;
		}
	}

	if (m->num_other_addr == m->max_other_addr) {
		/* Expand the list... */
		m->max_other_addr *= 2;
		m->other_addr = (char **) xrealloc(m->other_addr, m->max_other_addr * sizeof(char *));
		m->other_hello = (struct timeval **) xrealloc(m->other_hello, m->max_other_addr * sizeof(struct timeval *));
	}
	m->other_hello[m->num_other_addr]=(struct timeval *)xmalloc(sizeof(struct timeval));
	gettimeofday(m->other_hello[m->num_other_addr],NULL);
	m->other_addr[m->num_other_addr++] = xstrdup(a);
}

static void remove_other_addr(struct mbus *m, char *a)
{
	/* Removes the address a from the m->other_addr field of the */
	/* mbus structure.                                           */
	int	i, j;

	mbus_validate(m);

	for (i = 0; i < m->num_other_addr; i++) {
		if (mbus_addr_match(m->other_addr[i], a)) {
			xfree(m->other_addr[i]);
			xfree(m->other_hello[i]);
			for (j = i+1; j < m->num_other_addr; j++) {
				m->other_addr[j-1] = m->other_addr[j];
				m->other_hello[j-1] = m->other_hello[j];
			}
			m->other_addr[m->num_other_addr  - 1] = NULL;
			m->other_hello[m->num_other_addr - 1] = NULL;
			m->num_other_addr--;
		}
	}
}

static void remove_inactiv_other_addr(struct mbus *m, struct timeval t, int interval){
	/* Remove addresses we haven't heard from for about 5 * interval */
	/* Count backwards so it is safe to remove entries               */
	int i;
    
	mbus_validate(m);

	for (i=m->num_other_addr-1; i>=0; i--){
		if ((t.tv_sec-(m->other_hello[i]->tv_sec)) > 5 * interval) {
			debug_msg("remove dead entity (%s)\n", m->other_addr[i]);
			remove_other_addr(m, m->other_addr[i]);
		}
	}
}

int mbus_addr_valid(struct mbus *m, char *addr)
{
	int	i;

	mbus_validate(m);

	for (i = 0; i < m->num_other_addr; i++) {
		if (mbus_addr_match(m->other_addr[i], addr)) {
			return TRUE;
		}
	}
	return FALSE;
}

static int mbus_addr_unique(struct mbus *m, char *addr)
{
	int     i, n = 0;

	mbus_validate(m);

	for (i = 0; i < m->num_other_addr; i++) {
		if (mbus_addr_match(m->other_addr[i], addr)) {
			n++;
		}
	}
	return n==1;
}

/* The mb_* functions are used to build an mbus message up in the */
/* mb_buffer, and to add authentication and encryption before the */
/* message is sent.                                               */
char	 mb_cryptbuf[MBUS_BUF_SIZE];
char	*mb_buffer;
char	*mb_bufpos;

#define MBUS_AUTH_LEN 16

static void mb_header(int seqnum, int ts, char reliable, const char *src, const char *dst, int ackseq)
{
	xmemchk();
	mb_buffer   = (char *) xmalloc(MBUS_BUF_SIZE + 1);
	memset(mb_buffer,   0, MBUS_BUF_SIZE);
	memset(mb_buffer, ' ', MBUS_AUTH_LEN);
	mb_bufpos = mb_buffer + MBUS_AUTH_LEN;
	sprintf(mb_bufpos, "\nmbus/1.0 %6d %9d %c (%s) %s ", seqnum, ts, reliable, src, dst);
	mb_bufpos += 33 + strlen(src) + strlen(dst);
	if (ackseq == -1) {
		sprintf(mb_bufpos, "()\n");
		mb_bufpos += 3;
	} else {
		sprintf(mb_bufpos, "(%6d)\n", ackseq);
		mb_bufpos += 9;
	}
}

static void mb_add_command(const char *cmnd, const char *args)
{
	int offset = strlen(cmnd) + strlen(args) + 5;

	ASSERT((mb_bufpos + offset - mb_buffer) < MBUS_BUF_SIZE);

	sprintf(mb_bufpos, "%s (%s)\n", cmnd, args);
	mb_bufpos += offset - 1; /* The -1 in offset means we're not NUL terminated - fix in mb_send */
}

static void mb_send(struct mbus *m)
{
	char		digest[16];
	int		len;
	unsigned char	initVec[8] = {0,0,0,0,0,0,0,0};
 
	mbus_validate(m);

	*(mb_bufpos++) = '\0';
	ASSERT((mb_bufpos - mb_buffer) < MBUS_BUF_SIZE);
	ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);

	/* Pad to a multiple of 8 bytes, so the encryption can work... */
	while (((mb_bufpos - mb_buffer) % 8) != 0) {
		*(mb_bufpos++) = '\0';
	}
	len = mb_bufpos - mb_buffer;
	ASSERT(len < MBUS_BUF_SIZE);
	ASSERT(strlen(mb_buffer) < MBUS_BUF_SIZE);

	xmemchk();
	if (m->hashkey != NULL) {
		/* Authenticate... */
		hmac_md5(mb_buffer + MBUS_AUTH_LEN+1, strlen(mb_buffer) - (MBUS_AUTH_LEN+1), m->hashkey, m->hashkeylen, digest);
		base64encode(digest, 12, mb_buffer, MBUS_AUTH_LEN);
	}
	xmemchk();
	if (m->encrkey != NULL) {
		/* Encrypt... */
		memset(mb_cryptbuf, 0, MBUS_BUF_SIZE);
		memcpy(mb_cryptbuf, mb_buffer, len);
		ASSERT((len % 8) == 0);
		ASSERT(len < MBUS_BUF_SIZE);
		ASSERT(m->encrkeylen == 8);
		xmemchk();
		qfDES_CBC_e(m->encrkey, mb_cryptbuf, len, initVec);
		xmemchk();
		memcpy(mb_buffer, mb_cryptbuf, len);
	}
	xmemchk();
	udp_send(m->s, mb_buffer, len);
	xfree(mb_buffer);
}

static void resend(struct mbus *m, struct mbus_msg *curr) 
{
	/* Don't need to check for buffer overflows: this was done in mbus_send() when */
	/* this message was first transmitted. If it was okay then, it's okay now.     */
	int	 i;

	mbus_validate(m);

	mb_header(curr->seqnum, curr->comp_time.tv_sec, (char)(curr->reliable?'R':'U'), m->addr, curr->dest, -1);
	for (i = 0; i < curr->num_cmds; i++) {
		mb_add_command(curr->cmd_list[i], curr->arg_list[i]);
	}
	mb_send(m);
	curr->retransmit_count++;
}

void mbus_retransmit(struct mbus *m)
{
	struct mbus_msg	*curr = m->waiting_ack;
	struct timeval	time;
	long		diff;

	mbus_validate(m);

	if (!mbus_waiting_ack(m)) {
		return;
	}

	mbus_msg_validate(curr);

	gettimeofday(&time, NULL);

	/* diff is time in milliseconds that the message has been awaiting an ACK */
	diff = ((time.tv_sec * 1000) + (time.tv_usec / 1000)) - ((curr->send_time.tv_sec * 1000) + (curr->send_time.tv_usec / 1000));
	if (diff > 10000) {
		debug_msg("Reliable mbus message failed!\n");
		if (m->err_handler == NULL) {
			abort();
		}
		m->err_handler(curr->seqnum, MBUS_MESSAGE_LOST);
		/* if we don't delete this failed message, the error handler
                   gets triggered every time we call mbus_retransmit */
		while (m->waiting_ack->num_cmds > 0) {
		    m->waiting_ack->num_cmds--;
		    xfree(m->waiting_ack->cmd_list[m->waiting_ack->num_cmds]);
		    xfree(m->waiting_ack->arg_list[m->waiting_ack->num_cmds]);
		}
		xfree(m->waiting_ack->dest);
		xfree(m->waiting_ack);
		m->waiting_ack = NULL;
		return;
	} 
	/* Note: We only send one retransmission each time, to avoid
	 * overflowing the receiver with a burst of requests...
	 */
	if ((diff > 750) && (curr->retransmit_count == 2)) {
		resend(m, curr);
		return;
	} 
	if ((diff > 500) && (curr->retransmit_count == 1)) {
		resend(m, curr);
		return;
	} 
	if ((diff > 250) && (curr->retransmit_count == 0)) {
		resend(m, curr);
		return;
	}
}

void mbus_heartbeat(struct mbus *m, int interval)
{
	struct timeval	curr_time;
	char	*a = (char *) xmalloc(3);
	sprintf(a, "()");

	mbus_validate(m);

	gettimeofday(&curr_time, NULL);
	if (curr_time.tv_sec - m->last_heartbeat.tv_sec >= interval) {
		mb_header(++m->seqnum, (int) curr_time.tv_sec, 'U', m->addr, "()", -1);
		mb_add_command("mbus.hello", "");
		mb_send(m);

		m->last_heartbeat = curr_time;
		/* Remove dead sources */
		remove_inactiv_other_addr(m, curr_time, interval);
	}
	xfree(a);
}

int mbus_waiting_ack(struct mbus *m)
{
	mbus_validate(m);
	return m->waiting_ack != NULL;
}

int mbus_sent_all(struct mbus *m)
{
	mbus_validate(m);
	return (m->cmd_queue == NULL) && (m->waiting_ack == NULL);
}

struct mbus *mbus_init(void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat), 
		       void  (*err_handler)(int seqnum, int reason),
		       char  *addr)
{
	struct mbus		*m;
	struct mbus_key	 	 k;
	struct mbus_parser	*mp;
	int		 	 i;
	char            	*net_addr, *tmp;
	uint16_t         	 net_port;
	int              	 net_scope;

	m = (struct mbus *) xmalloc(sizeof(struct mbus));
	if (m == NULL) {
		debug_msg("Unable to allocate memory for mbus\n");
		return NULL;
	}

	m->cfg = mbus_create_config();
	mbus_lock_config_file(m->cfg);
	net_addr = (char *) xmalloc(20);
	mbus_get_net_addr(m->cfg, net_addr, &net_port, &net_scope);
	m->s		  = udp_init(net_addr, net_port, net_port, net_scope);
        if (m->s == NULL) {
                debug_msg("Unable to initialize mbus address\n");
                xfree(m);
                return NULL;
        }
	m->seqnum         = 0;
	m->cmd_handler    = cmd_handler;
	m->err_handler	  = err_handler;
	m->num_other_addr = 0;
	m->max_other_addr = 10;
	m->other_addr     = (char **) xmalloc(sizeof(char *) * 10);
	m->other_hello    = (struct timeval **) xmalloc(sizeof(struct timeval *) * 10);
	for (i = 0; i < 10; i++) {
		m->other_addr[i]  = NULL;
		m->other_hello[i] = NULL;
	}
	m->cmd_queue	  = NULL;
	m->waiting_ack	  = NULL;
	m->magic          = MBUS_MAGIC;
	m->index          = 0;
	m->index_sent     = 0;

	mp = mbus_parse_init(xstrdup(addr));
	if (!mbus_parse_lst(mp, &tmp)) {
		debug_msg("Invalid mbus address\n");
		abort();
	}
	m->addr = xstrdup(tmp);
	mbus_parse_done(mp);
	ASSERT(m->addr != NULL);

	gettimeofday(&(m->last_heartbeat), NULL);

	mbus_get_encrkey(m->cfg, &k);
	m->encrkey    = k.key;
	m->encrkeylen = k.key_len;

	mbus_get_hashkey(m->cfg, &k);
	m->hashkey    = k.key;
	m->hashkeylen = k.key_len;

	mbus_unlock_config_file(m->cfg);

	xfree(net_addr);

	return m;
}

void mbus_cmd_handler(struct mbus *m, void  (*cmd_handler)(char *src, char *cmd, char *arg, void *dat))
{
	mbus_validate(m);
	m->cmd_handler = cmd_handler;
}

static void mbus_flush_msgs(struct mbus_msg **queue)
{
        struct mbus_msg *curr, *next;
        int i;
	
        curr = *queue;
        while(curr) {
                next = curr->next;
                xfree(curr->dest);
                for(i = 0; i < curr->num_cmds; i++) {
                        xfree(curr->cmd_list[i]);
                        xfree(curr->arg_list[i]);
                }
		xfree(curr);