/* mnmserv.c Server program to broadcast images */ /****************************************************************************** Portions of this code are derived from work originally done within IMM 3.0 Written by Winston Dang * Copyright (c) University of Hawaii 1993. All rights reserved. * * License is granted to copy, to use, and to make and to use derivative * works for research, educational, government or evaluation purposes, provided * that the University of Hawaii is acknowledged in all documentation pertaining * to any such copy or derivative work. University of Hawaii grants no other * licenses expressed or imrplied. The University of Hawaii trade name should * not be used in any advertising without its written permission. * * UNIVERSITY OF HAWAII MAKES NO REPRESENTATIONS CONCERNING EITHER THE * MERCHANTABILITY OF THIS SOFTWARE OR THE SUITABILITY OF THIS SOFTWARE * FOR ANY PARTICULAR PURPOSE. The software is provided "as is" without * express or implied warranty of any kind. * * These notices must be retained in any copies of any part of this software. ******************************************************************************/ /****************************************************************************** * This code release was developed and written by * Joe Macker and Winston Dang, * 1996, 1997 ******************************************************************************/ #define MAIN #include "protocol.h" #include #if defined(sgi) || defined (ultrix) || defined (__alpha) || defined(hpux) || defined(__NetBSD__) #include #else #include #include #endif #include "immserv.h" #include #include #include #include #include #include #ifdef __NetBSD__ #include #endif #ifdef direct #undef direct #endif direct static struct sockaddr_in addr; static int len_addr; int seed; char deffile[] = ".mnm.def"; /* default file name */ char *default_file = deffile; unsigned txrate= DEFAULTRATE; unsigned int transmittime = TRANSMITTIME; /* interval between packet transmits */ unsigned int newtransmittime; /* transmit interval between packet transmission */ unsigned long lastdatathreshold = 0; /* clear to send gate flag */ int holdtime = MINWAIT; /* default hold between transmits*/ char directname[256] = {NULL}; /* directory name used in walkdir subroutine */ int directlen; /* length of directory name */ void closeall(int); void resetalrm(), closealrm(); void *recdata(); void *signal_hand(); char *localhost = NULL; struct key directlist[MAXENT]; /* list of directories to scan - setup by user */ int dir_number=0; /* number of directories in list above */ char portnum[6]; int maxread = PACKETSIZE; /* default size of packet data */ unsigned int mintransmittime = MINTRANSMITTIME; unsigned int maxtransmittime = MAXTRANSMITTIME; int heard_nack_flag = FALSE; /* set only if I heard a nack */ int request_ack_flag = FALSE; /* set by initialize routine to request Packs */ struct infotype infoclient = {32000,1,15,0}; struct servetype serveFile[MAXFILES]; /* list of multicasted file - carries status */ unsigned long lasttime(); /* latest file date timestamp */ unsigned int num_passes = 1; /* number of passes throught a directory 0-repeat */ extern char *queueit(); extern char *liststrdup(); extern struct listdef *talloc(); extern int err_block_size; extern int err_in_block; /******************************************************************************/ int main(argc, argv) int argc; char **argv; { int dirsel; struct key *dir_ptr; int i; int deleteflag = FALSE; thread_t recdata_process; sigset_t set; /* initialize the command line options */ deleteflag = initialize(argc,argv); newtransmittime= (((float)maxread)/((float)(txrate))*8000000); /*time between packets in usecs = (size(Byte)/(bytes/sec))^10e6 */ if (debug) printf ("transmit rate = %d psize= %d slottimecalc =%d\n", txrate,maxread,newtransmittime); if (newtransmittime > maxtransmittime) { newtransmittime = maxtransmittime; if (debug) printf ("lower rate limit MAXTRANSMITTIME exceeded\n"); } if (newtransmittime < mintransmittime) { newtransmittime = mintransmittime; if (debug) printf ("upper rate limit MINTRANSMITTIME exceeded\n"); } if (debug) printf ("transmit rate = %d psize= %d slottime =%d\n", txrate,maxread,newtransmittime); if (debug) printf("default is %s multicast %s port %s %d\n", default_file,group,portnum, BUF_HDR_SZE); /* test if the mnmserv session exists already */ i = ProcExists(portnum,group); /* Initializations receive port */ sock=createsocket(portnum,group); if (sock < 0) fatal("Improper port number or IP address assigned\n"); /* Initializations send port */ xmitsock=CreateSendSocket(&addr,&len_addr,portnum,group,&ttl); signal(SIGTERM, closeall); sigfillset(&set); thr_sigsetmask(SIG_SETMASK, &set, NULL); thr_create(NULL,0, signal_hand, 0, THR_NEW_LWP | THR_DAEMON | THR_DETACHED, NULL); /* Create the threads. */ if (thr_create(NULL,0,recdata,NULL,THR_NEW_LWP,&recdata_process) != 0) { fatal("thread creation error"); } /* wait a few seconds to make sure we can transmit */ if (deleteflag) newmember(BUF_TYPE_BOMB); newmember(BUF_TYPE_NEW_MEM); /* walk through each specified directory */ for (;;) { cleartosend(1); /* wait for no traffic before multicasting */ for (dirsel = 0; dirsel < dir_number; dirsel++) { dir_ptr = &directlist[dirsel]; if (debug) { printf("WALK %s rep=%d ", dir_ptr->direct, dir_ptr->repeatflag); if (dir_ptr->time != 0) printf("%s",ctime(&(dir_ptr->time))); else printf("\n"); } *directname = NULL; directlen = strlen(dir_ptr->direct); if (dir_ptr->IsDirectory) { walkdir(dir_ptr->direct,dir_ptr); dir_ptr->time = dir_ptr->cutoff; } else { infoclient.lastid = deliver(dir_ptr->direct,"", dir_ptr->size, dir_ptr->repeatflag, dir_ptr->holdtime); } } /* check the request for files queue */ requestfile(RECOV_GET,dir_number,NULL); if (num_passes) { if (--num_passes == 0) { if (debug) printf("exiting procedure\n"); delaytime(infoclient.lastid, 0); break; } } usleep(newtransmittime); } close(xmitsock); } /******************************************************************************/ /* simple routine to recursively walk through a directory */ int walkdir(direct,dir_ptr) char *direct; struct key *dir_ptr; { int num_entries; DIR *dirp; struct dirent *dp; struct listdef *filelist = NULL; int transmitone; struct stat dirinfo; u_int dirsize; char *filename; if ( chdir(direct) ) { if (debug > 4) printf("fail to chdir to %s\n",direct); return FALSE; } else { if (*direct != '/') strcat(directname,"/"); strcat(directname,direct); if (debug) printf(" chdir to %s\n",directname); } transmitone = 0; /* get a directory listing of the current subdirectory */ num_entries = 0; dirp = opendir("."); if (dirp == NULL) { warn("directory %s inacessible \n",direct); return; } while ((dp = readdir(dirp)) != NULL) { if (dp->d_name[0] == '.') continue; /* check if we need to get only the latest files */ if ((dir_ptr->repeatflag == 0) && (stat( dp->d_name, &dirinfo) == 0)) { if ( dirinfo.st_ctime < dir_ptr->cutoff) continue; dir_ptr->cutoff = dirinfo.st_ctime; } queueit(RECOV_PUT, &filelist, dp->d_name); num_entries++; } (void) closedir(dirp); /* walk through the entries in the directory */ if (debug > 2) printf("Number of Entries is %d\n", num_entries); while (filelist != NULL) { /* check the stats of particular files */ if (stat( filelist->name, &dirinfo) == 0) { dirsize = dirinfo.st_size; if ((S_ISDIR(dirinfo.st_mode) == TRUE) && (dirinfo.st_mode & S_IRUSR)) { if (debug > 2) printf("direct "); walkdir(filelist->name,dir_ptr); if ( chdir("..")) { if (debug) warn("failed to chdir to %s\n",direct); return FALSE; } filename = strrchr(directname,'/'); *filename = NULL; queueit(RECOV_GET, &filelist, NULL); continue; } } else if (debug) warn("stat fail "); if (debug > 2) printf(" %s %d\n",filelist->name,dirsize); /* check the minimum requirements */ if (dirsize < MINFILESIZE) { if (debug > 2) printf(" fail min req\n"); queueit(RECOV_GET, &filelist, NULL); continue; } /* check if we need to delete the old files */ if (dir_ptr->repeatflag == 0) { if ( dirinfo.st_ctime <= dir_ptr->time) { queueit(RECOV_GET, &filelist, NULL); continue; } if ( dirinfo.st_ctime != dir_ptr->cutoff) { queueit(RECOV_GET, &filelist, NULL); continue; } if (debug > 2) showtime(dirinfo.st_ctime); } /* begin transmission - multicast it now */ transmitone++; infoclient.lastid = deliver(filelist->name, &directname[directlen], dirsize,dir_ptr->repeatflag, dir_ptr->holdtime); queueit(RECOV_GET, &filelist, NULL); } if (!transmitone) delaytime(infoclient.lastid, dir_ptr->holdtime); } /******************************************************************************/ /* shutdown everything - close all files */ void closeall(int nop) { /* finished - send end of file marker */ exit(0); } /******************************************************************************/ /* multicast an ID packet */ int deliver_id(fileptr, toggle) struct servetype *fileptr; int toggle; { struct bufid buf; int byteswrit; int hold; if (fileptr->inquiryflag & BUF_FLG_REPEAT) return FALSE; if (fileptr->inquiryflag & BUF_FLG_EOF) hold = fileptr->waittime - time(NULL); else hold = fileptr->hold; if (hold <= 0) return FALSE; /* the following flag simple toggles back and forth with each new send */ if (toggle) fileptr->inquiryflag ^= BUF_FLG_TOGGLE; buf.head.version = VERSION; buf.head.type = BUF_TYPE_ID; buf.head.flags = fileptr->inquiryflag; buf.head.fid = (unsigned short)htons(fileptr->fid); buf.head.bytepos = (unsigned int)htonl(fileptr->dirsize); buf.hold_duration = (unsigned short) htons(hold); if (strlen(fileptr->border) < 1) sprintf((char *) buf.data,"%s %s", fileptr->filename,getlogin()); else sprintf((char *) buf.data,"%s/%s %s", fileptr->border,fileptr->filename,getlogin()); if (debug) printf("sendid:%d %s %u %x\n",buf.head.fid,buf.data, fileptr->dirsize, buf.head.flags); byteswrit = sendbuf(&buf,(strlen((char *)buf.data) + (2 * sizeof(short)) + BUF_HDR_SZE + 1)); return TRUE; } /******************************************************************************/ /* multicast an new member ID packet or a request for servers to drop */ int newmember(message) int message; { struct bufdef buf; int byteswrit; int counter = BUF_HDR_SZE + 4; if (debug) printf("send new mem %d\n",message); buf.head.version = VERSION; buf.head.fid = 0; buf.head.type = message; buf.head.flags = 0; buf.head.bytepos = (unsigned int)htonl(VERSION); byteswrit = sendbuf(&buf, counter); } /******************************************************************************/ /* take care of repeating all requests for packet retransmits */ int deliverRepeat(idsend) int idsend; { struct servetype *fileptr; struct bufdef buf; int ring,len,byteswrit; unsigned long i, minimum,maximum; int found = 0; for (ring = 0; ring < MAXFILES; ring++) { fileptr = &serveFile[ring]; if (fileptr->dirsize == 0) continue; if ( missingbufs(RECOV_TEST,fileptr,0) == FALSE) continue; minimum = missingbufs(RECOV_MISS_FREE,fileptr,0); found++; /* there are packets to retransmit - iterate through it now */ buf.head.version = VERSION; buf.head.flags = fileptr->inquiryflag; buf.head.fid = fileptr->fid; buf.head.type = BUF_TYPE_DATA; maximum = (fileptr->pos + maxread - 1) / maxread; buf.head.fid = (unsigned short)htons(buf.head.fid); for (i = minimum; i <= maximum; i++) { if (missingbufs(RECOV_GET,fileptr,i)) { buf.head.bytepos = i * maxread; lseek(fileptr->ifp, buf.head.bytepos, SEEK_SET); len = read(fileptr->ifp, buf.data, maxread); buf.head.bytepos = (unsigned int)htonl(buf.head.bytepos); byteswrit = sendbuf(&buf, len + BUF_HDR_SZE ); fileptr->packets_sent++; } } lseek(fileptr->ifp, fileptr->pos, SEEK_SET); if (idsend) deliver_id(fileptr, TRUE); usleep(newtransmittime); } return found; } /******************************************************************************/ /* multicast the specified file */ int deliver(name,directory, dirsize, repeat,waittime) char *name; /* file name */ char *directory; /* directory we are in */ unsigned int dirsize; /* size of file to be transmitted */ int repeat; /* number of times to repeat */ int waittime; /* suggested client response time allowed */ { struct servetype *fileptr; static u_short fid = 0; int len; struct bufdef buf; int byteswrit; unsigned int bytepos; if (debug) printf("wait to send %s %s %d %d %d\n", directory, name,dirsize, repeat, waittime); if (fid == 0) fid = time(NULL) % 86400; /* one day in seconds */ fid++; fileptr = &serveFile[fid % MAXFILES]; /* close the previously opened file */ if (fileptr->dirsize != 0) { /* adjust packet transmission rate from historical data */ if (fileptr->packets_sent > 100) { if (((fileptr->packets_requested * 25) < fileptr->packets_sent) && (newtransmittime > mintransmittime)) { newtransmittime *= 0.9; } else if (((fileptr->packets_requested * 10) > fileptr->packets_sent) && (newtransmittime < maxtransmittime)) { newtransmittime *= 1.07; } } if (waittime < MAXWAIT) delaytime(fileptr->fid,0); close(fileptr->ifp); free(fileptr->filename); free(fileptr->border); fileptr->dirsize = 0; fileptr->packets_sent = 0; fileptr->packets_requested = 0; } /* wait for a clear time to transmit - probably want to randomize this */ /* newtransmittime = transmittime; reset the transmit time */ buf.head.version = VERSION; buf.head.fid = (unsigned short)htons(fid); fileptr->fid = fid; fileptr->filename = liststrdup(name,NULL); fileptr->border = liststrdup(directory,NULL); fileptr->dirsize = dirsize; fileptr->hold = waittime; fileptr->pos = 0; fileptr->waittime = waittime + time(NULL); if (debug) printf("new file %s %u\n", fileptr->filename, dirsize); fileptr->inquiryflag = 0; /* set Positive ack flag if initialization flag set */ if (request_ack_flag == TRUE) fileptr->inquiryflag |= BUF_FLG_PACK; /* if the file size is too large to handle set the auto request bit */ if (dirsize > (MAXPACKETS * MAXPACKETSHIFT * PACKETSIZE)) fileptr->inquiryflag |= BUF_FLG_AUTOREQ; /* check if the file is executable */ if (access(fileptr->filename,X_OK) == 0) { fileptr->inquiryflag |= BUF_FLG_EXECUT; } deliver_id(fileptr,TRUE); missingbufs(RECOV_FREE,fileptr,0); buf.head.flags = fileptr->inquiryflag; /* begin transmission */ if ((fileptr->ifp = open(fileptr->filename, O_RDONLY)) == -1) { warn("%s open failed\n",fileptr->filename); fid--; return infoclient.lastid; } /* iterate through the file */ buf.head.type = BUF_TYPE_DATA; bytepos = 0; /* broadcast the entire file once only */ while ((len = read(fileptr->ifp, buf.data, maxread)) > 0) { buf.head.bytepos = (unsigned int)htonl(bytepos); byteswrit = sendbuf(&buf, len + BUF_HDR_SZE); bytepos += len; fileptr->pos = bytepos; fileptr->packets_sent++; if (debug > 1) printf("%u %d\n",fid,bytepos); if ( missingbufs(RECOV_TEST,fileptr,0)) { deliverRepeat(FALSE); } } fileptr->inquiryflag |= BUF_FLG_EOF; if (debug > 1) printf("wrote %lu bytes\n",bytepos); fileptr->waittime = waittime + time(NULL); deliver_id(fileptr,TRUE); if (waittime >= MAXWAIT) { delaytime(fid,waittime); fileptr->inquiryflag |= BUF_FLG_REPEAT; } return fid; } /*****************************************************************************/ /* check required packet recovery period */ int delaytime(fid, addwait) int fid; int addwait; { int i; u_long timeout; struct servetype *fileptr; fileptr = &serveFile[fid % MAXFILES]; if (fid != fileptr->fid) { if (debug > 3) printf("%d delaytime fid %d\n", fid,fileptr->fid); if (addwait == 0) timeout = time(NULL) + fileptr->hold; else timeout = time(NULL) + addwait; while (timeout > time(NULL)) { usleep(newtransmittime); if ( missingbufs(RECOV_TEST,fileptr,0)== TRUE) { timeout = time(NULL) + fileptr->hold; deliverRepeat(TRUE); } } usleep(newtransmittime); return; } if (addwait) { timeout = time(NULL) + MAXSILENCE; while (fileptr->waittime > time(NULL)) { usleep(newtransmittime); if (recdata()) { /* we got a request - wait for one more period */ if (missingbufs(RECOV_TEST,fileptr,0) == TRUE) fileptr->waittime += MINWAIT; deliverRepeat(TRUE); usleep(newtransmittime); sendcmd_alrm(BUF_FLG_PACK, ((MINWAIT * 1000) / 5), fileptr); usleep(newtransmittime); deliver_id(fileptr,TRUE); } if (timeout < time(NULL)) { sendcmd_alrm(BUF_FLG_NACK,200, fileptr); timeout = time(NULL) + MAXSILENCE; } } return; } /* new routine to use direct nack - maximum 100 retries */ for (i = 0; i < 100; i++) { /* transmit a NACK request for three seconds - if no response assume that nobody is interested in more packets */ heard_nack_flag = FALSE; timeout = time(NULL) + MINWAIT; while ((timeout > time(NULL)) && (! heard_nack_flag)) { sendcmd_alrm(BUF_FLG_NACK,200, fileptr); usleep(newtransmittime); if (missingbufs(RECOV_TEST,fileptr,0)) { fileptr->waittime += MINWAIT; deliverRepeat(TRUE); } } if (heard_nack_flag == FALSE) break; /* someone wants more information - send request to send selective requests */ fileptr->waittime = time(NULL) + MINWAIT; deliver_id(fileptr,TRUE); while (fileptr->waittime > time(NULL)) { usleep(newtransmittime); if (missingbufs(RECOV_TEST,fileptr,0)) { fileptr->waittime += MINWAIT; deliverRepeat(TRUE); } sendcmd_alrm(BUF_FLG_PACK, ((MINWAIT * 1000) / 5), fileptr); } } return; } /*****************************************************************************/ /* minimum wait time between server transmits */ int cleartosend(minhold) int minhold; { unsigned long currenthold; do { lastdatathreshold = FALSE; currenthold = time(NULL) + minhold; do { usleep(newtransmittime); } while (time(NULL) < currenthold); } while (lastdatathreshold == TRUE); return TRUE; } /*****************************************************************************/ /* recovery system - pegboard to mark packet to rebroadcast */ int missingbufs(commnd, bufptr,pos) int commnd; struct servetype *bufptr; unsigned int pos; { static mutex_t mutex; struct window_packets *curr_window_ptr; u_long *bufsize; int i; u_int seq,mask; int ring; /* the following three statements define which pegboard to use */ ring = (pos >> 5) >> ((MAXPACKETS/ 2) - 1); seq = (pos >> 5) & (MAXPACKETS - 1); mask = 0x01 << (pos & (MAXPACKETSHIFT - 1)); switch (commnd) { case RECOV_INIT: if(mutex_init(&mutex, USYNC_THREAD, NULL) != 0) { printf("Error initializing Mutex.\n"); exit(1); } return; case RECOV_FREE: /* free up all the allocated buffers */ mutex_lock(&mutex); curr_window_ptr = bufptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->prev != NULL)) { curr_window_ptr = curr_window_ptr->prev; bufptr->buf.tail = curr_window_ptr; free(curr_window_ptr->next); curr_window_ptr->next = NULL; } if (curr_window_ptr != NULL) { free(curr_window_ptr); bufptr->buf.tail = NULL; bufptr->buf.head = NULL; } bufptr->min_test_pos = -1; bufptr->buf.updateflag = FALSE; mutex_unlock(&mutex); return TRUE; case RECOV_PUT: /* mark packet as needed to be rebroadcated */ /* determine if we need to allocate another window */ mutex_lock(&mutex); curr_window_ptr = bufptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->BasePosition != ring)){ curr_window_ptr = curr_window_ptr->prev; } if (curr_window_ptr == NULL) { curr_window_ptr = (struct window_packets *) malloc(sizeof(struct window_packets)); memset(&curr_window_ptr->RecPegPosition[0],0, MAXPACKETS * sizeof(long)); memset(&curr_window_ptr->MissPegPosition[0],0, MAXPACKETS * sizeof(long)); curr_window_ptr->BasePosition = ring; /* insert into link list */ curr_window_ptr->next = NULL; if (bufptr->buf.tail == NULL) { curr_window_ptr->next = NULL; curr_window_ptr->prev = NULL; bufptr->buf.tail = curr_window_ptr; } else { bufptr->buf.tail->next = curr_window_ptr; curr_window_ptr->prev = bufptr->buf.tail; bufptr->buf.tail = curr_window_ptr; } if (debug > 2) printf("Missingbuf: pos %d ring %d\n", pos,ring); } if (bufptr->min_test_pos > pos) bufptr->min_test_pos = pos; bufsize = &curr_window_ptr->RecPegPosition[ seq]; /* test if we already received the packet */ if (*bufsize & mask) { mutex_unlock(&mutex); return TRUE; } /* didn't receive it. mark it received now */ *bufsize |= mask; if (debug && (bufptr->fid != 0)) printf(" %d ",pos); bufptr->buf.updateflag = TRUE; mutex_unlock(&mutex); return; case RECOV_TEST: mutex_lock(&mutex); if (bufptr->buf.updateflag) { mutex_unlock(&mutex); return TRUE; } mutex_unlock(&mutex); return FALSE; case RECOV_MISS_FREE: mutex_lock(&mutex); bufptr->buf.updateflag = FALSE; mutex_unlock(&mutex); return(bufptr->min_test_pos); case RECOV_GET: /* test packet received or not */ mutex_lock(&mutex); curr_window_ptr = bufptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->BasePosition != ring)) { curr_window_ptr = curr_window_ptr->prev; } if (curr_window_ptr == NULL) { mutex_unlock(&mutex); return 0; } bufsize = &curr_window_ptr->RecPegPosition[seq]; if (*bufsize & mask) { *bufsize &= ~mask; bufptr->min_test_pos = -1; if (debug > 1) printf("retrans %d %d %d %x %d\n",bufptr->fid,pos,seq,mask,i); mutex_unlock(&mutex); return (TRUE); } mutex_unlock(&mutex); return FALSE; } } /******************************************************************************/ /* multicast the buffer of length len*/ int sendbuf(buf,len) char *buf; int len; { int lwrite; #ifdef ERROR_TEST if (random_drop(err_block_size,err_in_block)) { if (debug) printf ("dropped packet\n"); return lwrite; } #endif /* ERROR_TEST */ for (;;) { lwrite=sendto(xmitsock, (char *) buf,len , 0, (struct sockaddr *) &addr, len_addr); usleep(newtransmittime); if (lwrite >= 0) return lwrite; warn("sendto: Cannot send %d\n", errno); } } /******************************************************************************/ /* send command packet to set alarm */ int sendcmd_alrm(cmd,dur,fileptr) int cmd; int dur; struct servetype *fileptr; { struct bufdef bufcmd; int byteswrit; switch (cmd) { case BUF_FLG_NACK: bufcmd.head.flags = fileptr->inquiryflag | BUF_FLG_NACK; break; case BUF_FLG_PACK: bufcmd.head.flags = fileptr->inquiryflag; break; case BUF_FLG_TOGGLE: fileptr->inquiryflag ^= BUF_FLG_TOGGLE; bufcmd.head.flags = fileptr->inquiryflag | BUF_FLG_PACK; } /* send request */ bufcmd.head.version = VERSION; bufcmd.head.type = BUF_TYPE_CMD; bufcmd.head.fid = (unsigned short) htons(fileptr->fid); bufcmd.head.bytepos = (unsigned int) htonl(dur); byteswrit = PackAck(RECOV_GET,NULL, &bufcmd.data[0]); if (debug) printf("cmd %d\n",dur); sendbuf(&bufcmd,BUF_HDR_SZE + byteswrit); return TRUE; } /******************************************************************************/ /* read the configuration file */ int initialize(argc, argv) int argc; char *argv[]; { FILE *ifp; char token[90]; char tokval[90]; char buf[180]; char *ptr; int i=0; int repeatflag = 0; int deleteflag = FALSE; struct stat info; struct key *dir_ptr; infoclient.lastid = -1; /* initialize the server file name */ for (i=0; i < MAXFILES; i++) { serveFile[i].lastfid = 999; serveFile[i].inquiryflag = 0; serveFile[i].fid = 0; serveFile[i].dirsize = 0; serveFile[i].buf.head = NULL; serveFile[i].buf.tail = NULL; } seed = time(NULL); #if defined(SOLARIS) || defined(hpux) srand(seed); #else srandom(seed); #endif missingbufs(RECOV_INIT,NULL,0); requestfile(RECOV_INIT,0,NULL); i = 1; while (i < argc) { if (*argv[i] == '-') { switch (*(argv[i] + 1)) { case 'a': request_ack_flag = TRUE; break; case 'd': i++; debug = atoi(argv[i]); break; case 'p': /* take the port number */ i++; strcpy(portnum,argv[i]); break; case 'I': case 'i': /* change the MULTICAST address */ i++; strcpy(group,argv[i]); break; case 'r': i++; /* number of passes to repeat transmission */ num_passes = atoi(argv[i]); break; case 'K': /* remove all servers on line */ deleteflag = TRUE; break; case 'l': /* specify the time to hold image */ i++; holdtime = atoi(argv[i]); break; case 'f': /* take another default */ i++; default_file = argv[i]; break; case 't': /* alter the time to live */ i++; ttl = atoi(argv[i]); break; #ifdef ERROR_TEST case 'e': /* set number of simulated dropped packets */ i++; err_in_block = atoi (argv[i]); break; #endif /* rate parameter in bits/sec */ case 's': /* s for speed */ i++; txrate = atoi (argv[i]); break; /*end of rate */ case 'L': /* change local host name */ i++; localhost = argv[i]; if (debug) printf("set hostname %s\n",localhost); break; case 'h': #ifdef ERROR_TEST printf("mnmserver [-d] [-p ] [-i ] [-r] [-s ] [-e ] [-f ] [-t ] [-h] [ ] [-i ] [-r] [-s ] [-f ] [-t ] [-h] [ IsDirectory = (info.st_mode & S_IFDIR); dir_ptr->size = info.st_size; if (dir_ptr->IsDirectory) { dir_ptr->direct = (char *) malloc(strlen(default_file) + 1); if (dir_ptr->direct != NULL) /* multicast a directory */ strcpy(dir_ptr->direct, default_file); } else { /* multicast a file */ strcpy(token,default_file); /* test if directory specified */ ptr = strchr(token,'/'); if (ptr != NULL) { *ptr = 0; chdir(token); ptr = strchr(default_file,'/'); ptr++; strcpy(token,ptr); } dir_ptr->direct = (char *) malloc(strlen(token) + 1); if (dir_ptr->direct != NULL) strcpy(dir_ptr->direct, token); } dir_ptr->repeatflag = TRUE; dir_ptr->holdtime = holdtime; dir_ptr->time = 0; dir_ptr->cutoff = dir_ptr->time; newtransmittime = transmittime; if (debug) printf("direct multicast:%s port:%s\n", dir_ptr->direct, portnum); dir_number++; return deleteflag; } } /* look for default file to multicast from */ dir_ptr = &directlist[0]; if (stat(default_file,&info) != 0) fatal("cannot stat file %s\n",default_file); /* test if directory */ if (info.st_mode & S_IFDIR) { dir_ptr->direct = (char *) malloc(strlen(default_file) + 1); if (dir_ptr->direct != NULL) strcpy(dir_ptr->direct, default_file); dir_ptr->repeatflag = repeatflag; dir_ptr->holdtime = holdtime; dir_ptr->time = lasttime(dir_ptr->direct) - 1; dir_ptr->IsDirectory = TRUE; dir_ptr->size = info.st_size; dir_ptr->cutoff = dir_ptr->time; newtransmittime = transmittime; if (debug) printf("direct:%s port:%s\n", dir_ptr->direct, portnum); dir_number++; return deleteflag; } if (debug) printf("default file is %s\n",default_file); if ((ifp = fopen(default_file,"r")) == NULL) { warn("unable to open %s\n",default_file); exit(0); } do { if (fgets(buf,80,ifp) == NULL) break; if (*buf == '#') continue; ptr = strchr(buf,'='); if (ptr == NULL) continue; ptr += 1; sscanf(buf, "%s", token); if (strcmp(token,"repeat") == 0) { sscanf(ptr,"%d",&repeatflag); } else if (strcmp(token,"min_transmit") == 0) { sscanf(ptr,"%d",&mintransmittime); if (debug) printf("min transmit = %d\n",mintransmittime); } else if (strcmp(token,"max_transmit") == 0) { sscanf(ptr,"%d",&maxtransmittime); if (debug) printf("max transmit = %d\n",maxtransmittime); } else if (strcmp(token,"hold") == 0) { sscanf(ptr,"%d",&holdtime); } else if (strcmp(token,"port") == 0) { *(ptr + (strlen(ptr) - 1)) = NULL; sscanf(ptr,"%s",portnum); } else if (strcmp(token,"transmit") == 0) { sscanf(ptr,"%u",&transmittime); if (debug) printf("start transmit = %d\n",transmittime); } else if (strcmp(token, "ttl") == 0) { sscanf(ptr, "%d", &ttl); } else if (strcmp(token, "group") == 0) { sscanf(ptr, "%s", group); } else if (strcmp(token, "maxread") == 0) { sscanf(ptr, "%d", &maxread); if (maxread > T_MAX ) maxread = T_MAX; } else if (strcmp(token, "requests") == 0) { sscanf(ptr, "%d", &infoclient.requests); } else if (strcmp(token, "frequency") == 0) { sscanf(ptr, "%d", &infoclient.frequency); } else if (strcmp(token, "duration") == 0) { sscanf(ptr, "%d", &infoclient.duration); } else if (strcmp(token,"direct") == 0) { sscanf(ptr,"%s",tokval); dir_ptr->direct = (char *) malloc(strlen(tokval) + 1); if (dir_ptr->direct != NULL) strcpy(dir_ptr->direct, tokval); dir_ptr->repeatflag = repeatflag; dir_ptr->holdtime = holdtime; dir_ptr->time = lasttime(dir_ptr->direct) - 1; dir_ptr->cutoff = dir_ptr->time; if (stat(dir_ptr->direct,&info) == 0) { /* test if directory */ dir_ptr->IsDirectory = (info.st_mode & S_IFDIR); dir_ptr->size = info.st_size; } if (debug) printf("direct:%s port:%s\n", dir_ptr->direct, portnum); dir_ptr++; if (dir_number++ >= MAXENT) break; } else printf("unknown token %s\n",token); } while (!feof(ifp)); fclose(ifp); newtransmittime = transmittime; return deleteflag; } /******************************************************************************/ /* routine to simply display the time in ASCII */ int showtime( disptime) time_t disptime; { time_t currtime = time(NULL); char messtime[80]; strcpy(messtime,ctime(&currtime)); messtime[ strlen(messtime) - 6] = NULL; if (debug) printf("%s ",messtime); if (debug > 1) { strcpy(messtime,ctime(&disptime)); messtime[ strlen(messtime) - 9] = NULL; printf("%s ",&messtime[4]); } } /******************************************************************************/ /* get a directory listing of the current subdirectory and return the latest time stamp found of all the files discovered in the root direct */ unsigned long lasttime(indir) char *indir; { DIR *dirp; /* directory pointer */ struct dirent *dp; /* directory file pointer */ unsigned long last = 0; struct stat dirinfo; if (chdir(indir)) { if (debug) warn("failed to chdir to %s\n",indir); return 0; } dirp = opendir("."); while ((dp = readdir(dirp)) != NULL) { if ( dp->d_name[0] == '.') continue; /* get the file size */ if (stat(dp->d_name, &dirinfo) != 0) continue; if (dirinfo.st_size == 0) continue; if ( dirinfo.st_ctime > last) { last = dirinfo.st_ctime; } } closedir(dirp); return last; } /******************************************************************************/ /* check for multicasted packets received - updates missingbufs */ void *recdata() { static struct bufdef buffer; static struct bufid *buffer_id; static struct servetype *fileptr; static struct bufstatdef *statptr; static struct bufmissdef *missptr; static unsigned long *data_ptr; static int lr; static int i; static int forme; static struct sockaddr_in srcaddr; int srcaddrlen = sizeof(srcaddr); memset((char *) &srcaddr, 0, sizeof(srcaddr)); for (;;) { lr = recvfrom(sock, (char *) &buffer,sizeof(buffer), 0, (struct sockaddr *) &srcaddr, &srcaddrlen); buffer.head.fid = (unsigned short)ntohs(buffer.head.fid); buffer.head.bytepos = (unsigned int)ntohl(buffer.head.bytepos); lr -= BUF_HDR_SZE; /* length of packet less header */ switch (buffer.head.type) { case BUF_TYPE_NEW_MEM: if (debug) printf("rd new mem %s ver %d\n", inet_ntoa(srcaddr.sin_addr),buffer.head.bytepos); for (i=0; i < MAXFILES; i++) { fileptr = &serveFile[i]; /* respond with a ID packet if we are the active server */ if ((fileptr->inquiryflag & BUF_FLG_EOF) && (fileptr->pos != 0)) deliver_id(fileptr,TRUE); } break; case BUF_TYPE_STAT: statptr = (struct bufstatdef *)&buffer; forme = memcmp(myipaddress, (char *) &srcaddr.sin_addr, 4); if ((forme == 0) && (buffer.head.flags & BUF_FLG_PACK)) { PackAck(RECOV_PUT,&srcaddr.sin_addr,NULL); printf("Received %s %d %x\n", inet_ntoa(srcaddr.sin_addr), statptr->fid, buffer.head.flags, statptr->retries, statptr->pack_recv); } else { if (debug) printf("rd stat %s %d %x g %u b %u r %u rec %u\n", inet_ntoa(srcaddr.sin_addr), statptr->fid, buffer.head.flags, statptr->good,statptr->bad, statptr->retries, statptr->pack_recv); } heard_nack_flag = TRUE; continue; case BUF_TYPE_MISS_REQ: case BUF_TYPE_MISS_ANS: /* missing packet requests */ forme = memcmp(myipaddress, buffer.data, 4); if (forme) continue; heard_nack_flag = TRUE; missptr = (struct bufmissdef *) &buffer; fileptr = &serveFile[(buffer.head.fid % MAXFILES)]; if (fileptr->fid != buffer.head.fid) break; /* set the transmit time interval - a crude test */ if (debug) { printf("rd %s id %u type %d pos %d lr %d ", inet_ntoa(srcaddr.sin_addr), missptr->head.fid,missptr->head.type, missptr->head.bytepos,lr); } /* for transmission control do not count packets that are large continuous runs */ i = missptr->head.bytepos - 1; if ((buffer.head.bytepos < (T_MAX / sizeof(long))) || ((missptr->data[i] - missptr->data[0]) == i)) fileptr->packets_requested += missptr->head.bytepos; data_ptr = &missptr->data[0]; for ( i = 0; i < missptr->head.bytepos; i++) { missingbufs(RECOV_PUT,fileptr, ntohl(*data_ptr++)); } if (debug) printf("\n"); break; case BUF_TYPE_DATA: /* data packet incoming */ lastdatathreshold = TRUE; if (debug > 3) printf("in data %s %d %d\n", inet_ntoa(srcaddr.sin_addr), buffer.head.fid, buffer.head.bytepos); break; case BUF_TYPE_ID: /* id packet incoming */ buffer_id = (struct bufid *) &buffer; forme = memcmp(myipaddress, (char *) &srcaddr.sin_addr, 4); if (!forme) continue; if (debug) printf("remote server id %s %d\n", inet_ntoa(srcaddr.sin_addr), buffer.head.fid); if (!(buffer.head.flags & BUF_FLG_EOF)) { lastdatathreshold = TRUE; } break; case BUF_TYPE_BOMB: forme = memcmp(myipaddress, (char *) &srcaddr.sin_addr, 4); if (!forme) continue; printf("remote server requested you stop transmitting %s\n", inet_ntoa(srcaddr.sin_addr)); closeall(0); case BUF_TYPE_REQUEST: requestfile(RECOV_PUT,buffer.head.bytepos,buffer.data); break; } } } /******************************************************************************/ /* Process requests from clients to multicast files */ requestfile(cmd,entries,data) int cmd; int entries; char *data; { static struct listdef *top; static mutex_t mutex; struct listdef *curr, *prev; struct stat dirinfo; struct key *dir_ptr; char *token; char *entry; char *name_ptr; int i,found; switch (cmd) { case RECOV_INIT: /* initialize file request queue */ top = NULL; workingdir = (char *) getenv("PWD"); if(mutex_init(&mutex, USYNC_THREAD, NULL) != 0) { printf("Error initializing Mutex.\n"); exit(1); } return TRUE; case RECOV_PUT: /* push entries on file request queue via insertion sort */ if (debug) printf("request PUT %d %s\n",entries,data); mutex_lock(&mutex); token = strtok(data," "); entry = liststrdup(token,NULL); while ((token = strtok(NULL," ")) != NULL) { curr = talloc(); curr->name = liststrdup(entry,token); if (top == NULL) { curr->next = NULL; top = curr; if (debug) printf("request queue %s\n",curr->name); continue; } prev = top; while ((prev->next != NULL) && (strcmp(prev->name,curr->name) < 0)) prev = prev->next; found = strcmp(prev->name, curr->name); if (found > 0) { /* insert into list */ curr->next = prev->next; prev->next = curr; name_ptr = curr->name; curr->name = prev->name; prev->name = name_ptr; if (debug) printf("alpha insert in queue %s %s\n", prev->name,curr->name); } else if (found == 0) { /* duplicate entry */ if (debug) printf("request duplicate in queue %s\n",curr->name); free(curr->name); free(curr); } else { /* insert at end of list */ curr->next = NULL; prev->next = curr; if (debug) printf("request insert queue %s\n",curr->name); } } free(entry); mutex_unlock(&mutex); return TRUE; case RECOV_TEST: mutex_lock(&mutex); prev = top; while (prev != NULL) { if (debug) printf("retrans queue %s\n",prev->name); prev = prev->next; } mutex_unlock(&mutex); return TRUE; case RECOV_GET: /* pop entry off of file request queue */ while (top != NULL) { /* extract name and directory */ /* pop off stack */ mutex_lock(&mutex); curr = top; top = top->next; mutex_unlock(&mutex); if (debug) printf("retransmit %s\n", curr->name); for (i= 0; i < entries; i++) { dir_ptr = &directlist[i]; if (chdir(dir_ptr->direct)) { warn("failed to chdir to %s\n",dir_ptr->direct); continue; } if (stat(curr->name, &dirinfo) == 0) { /* found it retransmit the file */ if (debug) printf("retransmit %s %u\n", curr->name, dirinfo.st_size); infoclient.lastid = deliver(curr->name,"", dirinfo.st_size, dir_ptr->repeatflag, dir_ptr->holdtime); break; } } if (i == entries) warn("retrans: unable to find %s in %s\n", curr->name, dir_ptr->direct ); free(curr->name); free(curr); } if (chdir(workingdir)) if (debug) warn("failed to chdir to %s\n",workingdir); return FALSE; } } /******************************************************************************/ /* positive acknowledge queue - holds stats heard by clients */ int PackAck(int cmd, struct sockaddr_in *srcaddr, char *target) { static struct listdef *top; struct listdef *curr, *prev; static mutex_t mutex; char *entry; int i; int addr_len = 4; switch (cmd) { case RECOV_INIT: /* initialize file request queue */ top = NULL; if(mutex_init(&mutex, USYNC_THREAD, NULL) != 0) { printf("Error initializing Mutex.\n"); exit(1); } return TRUE; case RECOV_PUT: mutex_lock(&mutex); entry = (char *) malloc(addr_len); memcpy(entry, srcaddr,addr_len); curr = talloc(); curr->name = entry; if (top == NULL) { curr->next = NULL; top = curr; } else { /* insert at end of list */ curr->next = top; top = curr; } mutex_unlock(&mutex); return TRUE; case RECOV_GET: /* pop entry off of file request queue */ i = 0; mutex_lock(&mutex); while (top != NULL) { /* pop off stack */ curr = top; top = top->next; /* copy ip address */ memcpy(target, curr->name,addr_len); if (debug > 3) printf("%s ",inet_ntoa(curr->name)); target += addr_len; free(curr->name); free(curr); if ( ++i >= (T_MAX / addr_len)) break; } if ((debug > 3) && i ) printf(" total acks %d\n",i); mutex_unlock(&mutex); return (i * addr_len); } } /******************************************************************************/ /* make sure that a duplicate server process does not exist */ ProcExists(port,ip) char *port; char *ip; { FILE *ifp; char cmd[120]; int myid; int i; myid = getpid(); #if defined sgi || defined SOLARIS || defined(hpux) sprintf(cmd,"/bin/ps -ef | /bin/grep mnmserv"); #else #if defined (ultrix) sprintf(cmd,"/bin/ps -ax | /usr/ucb/grep mnmserv"); #else sprintf(cmd,"/bin/ps -ax |/usr/bin/grep mnmserv"); #endif #endif ifp = popen(cmd,"r"); do { fgets(cmd,120,ifp); if (debug > 3) printf("ps shows: %s\n", cmd); if ((strstr(cmd,"grep") == NULL) && (strstr(cmd,port) != NULL) && (strstr(cmd,ip) != NULL)) { /* found existing port */ #if defined sgi || defined SOLARIS || defined(hpux) sscanf(cmd,"%*s %d", &i); #else sscanf(cmd,"%d",&i); #endif if (i != myid) { if (debug) printf("MNM server already running on: %s\n", cmd); pclose(ifp); kill(i, SIGKILL); return TRUE; } } } while (!feof(ifp)); pclose(ifp); return FALSE; } /******************************************************************************/ #ifndef __NetBSD__ /* routine to put process to sleep */ int usleep( wait) u_int wait; { struct timespec value; value.tv_sec = wait / 1000000; value.tv_nsec = (wait % 1000000) * 1000; nanosleep(&value,NULL); } #endif /*****************************************************************************/ void *signal_hand(void *arg) { sigset_t set; int sig; sigfillset(&set); while (1) { printf("Signal handling thread started\n"); switch (sig = sigwait(&set)) { case SIGINT: case SIGKILL: case SIGFPE: case SIGTERM: printf("Signal handling thread interrupted\n"); printf("Processing Spurious Asyn Signal %d Exiting...\n",sig); closeall(0); break; default: printf("Processing Other Signal - %d\n",sig); } } } /*****************************************************************************/