/* mcl.c Reliable file multicast subscriber */ /****************************************************************************** This code is 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 #include #include #include #include #ifdef IMM #include #endif extern int SendDeclarePacket(); extern int CreateControlSocket(); extern int createsocket(); extern int readbuf(); extern int cmdbuf(); extern int dumpbuf(); extern int serveradd(); extern int memberadd(); extern int xlinit(); extern int xlopen(); extern int xlclose(); extern void abortimage(); extern int err_block_size; extern int err_in_block; struct nodetype *checkhost(); void sendack(int); void RecData(); int recoverbufs(); int seed; struct stattype statnum; struct sockaddr_in addr_dist; int len_addr_dist; extern int locktransmit; extern char *localhost; struct hosttype *hosts_tail; struct hosttype *hosts_head; /* initialization routine */ char *OpenRecvFD(portgrp,hostname,archfilenum) char *portgrp; /* socket to open */ char *hostname; /* ip address */ int archfilenum; /* number of files to archive- not implemented yet */ { char MC_GROUP[16]; char myname[80]; char myhostname[80]; unsigned int i1, i2, i3, i4; struct hostent *hp; checkhost(RECOV_INIT,NULL, NULL); recoverbufs(NULL, 0,0, RECOV_INIT); if (getenv("DISPLAY") == NULL) xlinit(archfilenum,":0.0"); else xlinit(archfilenum,getenv("DISPLAY")); strcpy(statnum.portgrp, portgrp); /* get the user name */ if (statnum.name_user[0] == NULL) { /* get the user name */ /* cuserid(myname); */ strcpy(myname,(char *) getlogin()); /* get the hostname of this machine */ if (localhost == NULL) { gethostname(myhostname, sizeof(myhostname)); } else { strcpy(myhostname,localhost); } hp = gethostbyname(myhostname); strcpy(myhostname, hp->h_name); if (strchr(myhostname,'.') != NULL) sprintf(statnum.name_user, "%s@%s", myname,myhostname); else strcpy(statnum.name_user, myname); } /* set up the response socket to the server */ if (hostname != NULL) { if(sscanf(hostname, "%u.%u.%u.%u", &i4, &i3, &i2, &i1) != 4){ if((hp=gethostbyname(hostname)) == 0){ warn("%d-Unknown host %s\n", getpid(), hostname); exit(-1); } memcpy((char *)&addr_dist.sin_addr,hp->h_addr, 4); addr_dist.sin_family = AF_INET; strcpy(MC_GROUP, inet_ntoa(addr_dist.sin_addr)); }else{ strcpy(MC_GROUP, hostname); addr_dist.sin_family = AF_INET; addr_dist.sin_addr.s_addr = (u_long)inet_addr(hostname); } } /* Initialize the send socket */ xmitsock=CreateSendSocket(&addr_dist,&len_addr_dist,portgrp,MC_GROUP,&ttl); if (xmitsock < 0) fatal("Improper port number or IP address\n"); /* Initializations receive port */ sock=createsocket(portgrp,MC_GROUP); if (sock < 0) fatal("Improper port number or IP address\n"); #ifdef IMM fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NDELAY); Tk_CreateFileHandler(sock, TK_READABLE, RecData, NULL); #endif if (debug) fprintf(logfp, "%s host %s port %s\n", myhostname, MC_GROUP,portgrp); return(statnum.name_user); } /***************************************************************************/ /* receive multicasted packet handler */ void RecData() { struct sockaddr_in srcaddr; /* receiving address */ int srcaddrlen = sizeof(struct sockaddr_in); time_t currtime; int lr,pos, bytes_written; struct bufdef buf; struct nodetype *nodeptr; char butline[80]; /* text line for button */ /* define received packet types */ static struct bufid *bufid_ptr; static struct bufcmd *bufcmd_ptr; static struct bufmissdef *bufmissdef_ptr; static struct bufstatdef *bufstatdef_ptr; static struct bufmemberdef *bufmemberdef_ptr; /* the real engine - receive the packets and send them off */ while ((lr = recvfrom(sock, (char *) &buf,sizeof(buf), 0, (struct sockaddr *) &srcaddr, (int *) &srcaddrlen)) > 0) { buf.head.fid = (unsigned short)ntohs(buf.head.fid); /* set pointer to correct file */ nodeptr = checkhost(RECOV_PUT,&buf, &srcaddr); /* special case - toss the record but keep user informed */ if (nodeptr == NULL) { switch (buf.head.type) { case BUF_TYPE_ID: /* file id */ bufid_ptr = (struct bufid *) &buf; sprintf(butline,"%d",ntohs(bufid_ptr->hold_duration)); buttset(BUF_TYPE_MISS_REQ,butline,NULL); continue; case BUF_TYPE_DATA: sprintf(butline,"Data %u", ntohl(buf.bytepos)); buttset(BUF_TYPE_ID, butline, "Receiving"); continue; default: continue; } } switch (buf.head.type) { /* id packet identifies file - comes from server */ case BUF_TYPE_ID: /* file id */ bufid_ptr = (struct bufid *) &buf; *nodeptr->sendername = NULL; nodeptr->hold = ntohs(bufid_ptr->hold_duration); nodeptr->recovery_cycle_time = (nodeptr->hold > MAXWAIT) ? MAXWAIT : nodeptr->hold; if (sscanf((char *) bufid_ptr->data,"%s %s", nodeptr->currfile, nodeptr->sendername) < 2) continue; if (debug) fprintf(logfp,"Id %d %s %s %d %d\n", bufid_ptr->head.fid,nodeptr->currfile, nodeptr->sendername, nodeptr->hold, bufid_ptr->head.flags); strcat(nodeptr->sendername,"@"); strcat(nodeptr->sendername, inet_ntoa(srcaddr.sin_addr)); nodeptr->total = (unsigned int)ntohl(bufid_ptr->file_size); nodeptr->serverflags = bufid_ptr->head.flags; if (nodeptr->transmitcode == TRANSMISSION_START) { if (debug) sprintf(butline,"%s %u %d", nodeptr->currfile, nodeptr->total, bufid_ptr->head.fid); else sprintf(butline,"%s %u", nodeptr->currfile,nodeptr->total); buttset(BUF_TYPE_ID, butline, nodeptr->sendername); } if (nodeptr->transmitcode == TRANSMISSION_START) { nodeptr->transmitcode = TRANSMISSION_OK; } else if ((nodeptr->count != 0 ) && (nodeptr->transmitcode != TRANSMISSION_REST)) nodeptr->transmitcode = TRANSMISSION_OK; sprintf(butline,"%d",nodeptr->hold); buttset(BUF_TYPE_MISS_REQ,butline,NULL); break; /* specific request for an ack */ case BUF_TYPE_CMD: /* ack request */ bufcmd_ptr = (struct bufcmd *) &buf; bufcmd_ptr->cycle_duration = ntohl(bufcmd_ptr->cycle_duration); if (debug) fprintf(logfp,"recv ack request from %s %x\n", inet_ntoa(srcaddr.sin_addr),bufcmd_ptr->head.flags); /*check case if we send response to server only if file not received*/ if (bufcmd_ptr->head.flags & BUF_FLG_NACK) { if (nodeptr->transmitcode != TRANSMISSION_REST) { /* send only if we did not receive file */ setackalarm(BUF_TYPE_CMD, bufcmd_ptr->cycle_duration, nodeptr,bufcmd_ptr->head.flags); } } lr -= BUFCMD_SIZE; if (PackAttack(&buf,lr)) { if (debug) printf("ack confirm\n"); nodeptr->serverPackAck = BUF_FLG_PACK; } if (nodeptr->serverPackAck == BUF_FLG_PACK) { continue; } /* if we are looking for a positive ack send it as well if file received */ if ((bufcmd_ptr->head.flags & BUF_FLG_PACK) && (nodeptr->transmitcode == TRANSMISSION_REST)) { setackalarm(BUF_TYPE_CMD, bufcmd_ptr->cycle_duration, nodeptr,bufcmd_ptr->head.flags); } /* NACKs require almost immediate response - no more processing */ continue; /* statistical info from clients */ case BUF_TYPE_STAT: statinfo(&buf,&srcaddr); continue; /* report from other client - stating new membership */ case BUF_TYPE_NEW_MEM: bufmemberdef_ptr = (struct bufmemberdef *) &buf; bufmemberdef_ptr->myversion = ntohs(bufmemberdef_ptr->myversion); if (debug) { strcpy(butline, inet_ntoa(srcaddr.sin_addr)); fprintf(logfp,"new member %s ver %d\n", inet_ntoa(srcaddr.sin_addr),bufmemberdef_ptr->myversion); } continue; /* report from another client - stating missing packets */ case BUF_TYPE_MISS_REQ: /* missing packet request */ case BUF_TYPE_MISS_ANS: /* answer to missing packet request */ bufmissdef_ptr = (struct bufmissdef *) &buf; bufmissdef_ptr->num_requests = ntohl(bufmissdef_ptr->num_requests); if (debug) { fprintf(logfp,"%s #%d miss report %d \n", inet_ntoa(srcaddr.sin_addr), bufmissdef_ptr->head.type,bufmissdef_ptr->num_requests); } lr -= BUFMISSDEF_SIZE; if (nodeptr->transmitcode != TRANSMISSION_REST) missingreport(nodeptr, &buf,lr); break; /* data packet from server */ case BUF_TYPE_DATA: /* data packet */ lr -= BUFDEF_SIZE; buf.bytepos = ntohl(buf.bytepos); if (debug) fprintf(logfp,"data %u pos %d %d %u %u\n", buf.head.fid, buf.bytepos,lr, nodeptr->count, (((buf.bytepos * lr) + lr) * 100 / nodeptr->total)); if (nodeptr->transmitcode == TRANSMISSION_REST) break; scaleset(((buf.bytepos * lr) + lr) * 100 / nodeptr->total); if (nodeptr->buf.maxsize < lr) nodeptr->buf.maxsize = lr; pos = buf.bytepos; if (recoverbufs(nodeptr, pos, lr, RECOV_PUT)) { if (debug) fprintf(logfp,"dup %d\n", buf.bytepos); break; } /* write the data to file */ if ( nodeptr->bytes_written != (buf.bytepos * nodeptr->buf.maxsize)) { if (debug>6) printf("lseek %d %d\n",nodeptr->bytes_written, buf.bytepos); lseek(nodeptr->ofp, (buf.bytepos * nodeptr->buf.maxsize), SEEK_SET); bytes_written = write(nodeptr->ofp, buf.data, lr); } else bytes_written = write(nodeptr->ofp, buf.data, lr); if (bytes_written == -1) { perror("write failed"); } nodeptr->count += lr; if (debug) fprintf(logfp,"pos %d %d %u\n", buf.bytepos,lr, nodeptr->count); nodeptr->bytes_written = (buf.bytepos * lr) + bytes_written; statnum.pack_recv++; break; default: /* ignore this packet - not recognizable */ continue; } if (nodeptr->count != nodeptr->total) { if (buf.head.flags & BUF_FLG_EOF) { if (!(nodeptr->serverflags & BUF_FLG_EOF)) { sprintf(butline,"%d",nodeptr->hold); buttset(BUF_TYPE_MISS_REQ,butline,NULL); nodeptr->serverflags |= BUF_FLG_EOF; } /* eof received but file received is not complete */ if (nodeptr->transmitcode != TRANSMISSION_REST) { if (nodeptr->toggleflag == (buf.head.flags & BUF_FLG_TOGGLE)){ setackalarm(BUF_TYPE_MISS_ANS, nodeptr->recovery_cycle_time, nodeptr,buf.head.flags); } else { nodeptr->toggleflag = buf.head.flags & BUF_FLG_TOGGLE; setackalarm(BUF_TYPE_MISS_REQ, nodeptr->recovery_cycle_time, nodeptr,buf.head.flags); } } } } /* new file */ if (( nodeptr->count < nodeptr->total) || (nodeptr->transmitcode != TRANSMISSION_OK)) continue; currtime = time(NULL); if (debug) fprintf(logfp, "Received %s %d %d %u %s", nodeptr->currfile, nodeptr->hold,nodeptr->transmitcode, nodeptr->count, ctime(&currtime)); sprintf(butline,"%d",nodeptr->hold); serveradd(nodeptr->sendername,nodeptr->currfile); buttset(BUF_TYPE_MISS_REQ,butline,NULL); if (nodeptr->total == nodeptr->count) statnum.goodtrans++; if (xlclose(nodeptr)) { if (buf.head.flags & BUF_FLG_PACK) { if (debug) fprintf(logfp,"send ack\n"); setackalarm(BUF_TYPE_CMD,MINWAIT * 100, nodeptr,buf.head.flags); } recoverbufs(nodeptr,0,0,RECOV_FREE); checkhost(RECOV_FREE,NULL, NULL); } } } /******************************************************************************/ /* recovery system */ /* the system implements a selective request repeat missing packets are requested from the server - with missing packet requests delay timed. Packets requested from other clients are not supposed to be requested. server periodically resets this mechanism */ int recoverbufs(nodeptr, pos,bytes,commnd) struct nodetype *nodeptr; unsigned int pos; int bytes; int commnd; { u_long *bufsize; unsigned long mask, ring; unsigned long seq; static int counter = 0; static int current_count = 0; struct window_packets *curr_window_ptr; struct window_packets *tmp_window_ptr; static struct hosttype *hostptr; ring = (pos >> 5) >> ((MAXPACKETS/ 2) - 1); seq = (pos >> 5) & (MAXPACKETS - 1); mask = 0x01 << (pos & (MAXPACKETSHIFT - 1)); switch (commnd) { case RECOV_PUT: /* store another buffer */ if (bytes == 0) return TRUE; /* determine if we need to allocate another window */ curr_window_ptr = nodeptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->BasePosition != ring)) { curr_window_ptr = curr_window_ptr->prev; } if (curr_window_ptr == NULL) { if (debug) fprintf(logfp,"init RING %x %d\n",mask,seq); /* FIX */ curr_window_ptr = 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; curr_window_ptr->counter = counter++; nodeptr->age = time(NULL); /* insert into link list */ curr_window_ptr->next = NULL; if (nodeptr->buf.tail == NULL) { curr_window_ptr->next = NULL; curr_window_ptr->prev = NULL; nodeptr->buf.tail = curr_window_ptr; nodeptr->buf.head = curr_window_ptr; } else { nodeptr->buf.tail->next = curr_window_ptr; curr_window_ptr->prev = nodeptr->buf.tail; nodeptr->buf.tail = curr_window_ptr; } } bufsize = &curr_window_ptr->RecPegPosition[ seq]; /* test if we already received the packet */ if (*bufsize & mask) { if (debug) fprintf(logfp,"recov dup %u %u %u %x\n", pos,ring,seq,mask); return TRUE; } if (debug > 3) fprintf(logfp,"recov ins %u %u %u %x\n", pos,ring,seq,mask); /* didn't receive it. mark it received now */ *bufsize |= mask; if (!(nodeptr->serverflags & BUF_FLG_AUTOREQ)) return FALSE; if (curr_window_ptr->counter <= current_count) { current_count = curr_window_ptr->counter; return FALSE; } current_count = curr_window_ptr->counter; setackalarm(BUF_TYPE_MISS_REQ, nodeptr->recovery_cycle_time, nodeptr,0); if (debug) fprintf(logfp,"%d auto req. pos %d\n", nodeptr->fid, pos); case RECOV_MISS_FREE: /* free up all the allocated buffers */ for (hostptr = hosts_tail;hostptr != NULL; hostptr = (struct hosttype *) hostptr->prev) { nodeptr = &hostptr->node; if (nodeptr->transmitcode == TRANSMISSION_REST) continue; nodeptr->flag = FALSE; curr_window_ptr = nodeptr->buf.tail; while (curr_window_ptr != NULL) { memset(&curr_window_ptr->MissPegPosition[0],0,MAXPACKETS * 4); curr_window_ptr = curr_window_ptr->prev; } } return FALSE; case RECOV_INIT: /* initialize the buffer holder */ signal(SIGALRM, sendack); signal(SIGKILL, abortimage); signal(SIGTERM, abortimage); signal(SIGINT, abortimage); signal(SIGQUIT, abortimage); seed = time(NULL) % 65536; #if defined SOLARIS || defined(hpux) srand(seed); #else srandom(seed); #endif statnum.alarmoff = 0; statnum.goodtrans = 0; statnum.pack_recv = 0; statnum.badtrans = 0; statnum.lastreceived = NULL; statnum.retries = 0; return TRUE; case RECOV_FREE: /* free up all the allocated buffers */ nodeptr->buf.maxsize = 1; nodeptr->max = 0; nodeptr->min = 0; nodeptr->bytes_written = 0; curr_window_ptr = nodeptr->buf.tail; while (curr_window_ptr != NULL) { tmp_window_ptr = curr_window_ptr->prev; if (tmp_window_ptr != NULL) tmp_window_ptr->next = NULL; nodeptr->buf.tail = tmp_window_ptr; free(curr_window_ptr); curr_window_ptr = tmp_window_ptr; } nodeptr->buf.tail = NULL; return TRUE; case RECOV_MISS: /* mark packet as missing packet requested */ curr_window_ptr = nodeptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->BasePosition != ring)) { curr_window_ptr = curr_window_ptr->prev; } if (curr_window_ptr == NULL) return TRUE; bufsize = &curr_window_ptr->MissPegPosition[seq]; *bufsize |= mask; return TRUE; case RECOV_TEST: /* test packet received or not */ curr_window_ptr = nodeptr->buf.tail; while ((curr_window_ptr != NULL) && (curr_window_ptr->BasePosition != ring)) { curr_window_ptr = curr_window_ptr->prev; } if (curr_window_ptr == NULL) return 0; bufsize = &curr_window_ptr->RecPegPosition[seq]; if (*bufsize & mask) return 1; bufsize = &curr_window_ptr->MissPegPosition[seq]; if (*bufsize & mask) return -1; return 0; } } /******************************************************************************/ /* send an acknowledgement to the server on how we are doing */ void sendack(int nop) { struct bufmissdef buf; struct nodetype *nodeptr; struct bufstatdef *bufstat; struct bufmemberdef *newmember; struct hosttype *hostptr; unsigned long *data_ptr; unsigned long max = MAXPACKETS; int j; unsigned long counter; switch (statnum.alarmoff) { case BUF_TYPE_CMD: if (debug) fprintf(logfp,"alarm ack sent %x\n",statnum.heard_nack_flag); if ((statnum.last_buf_flag & BUF_FLG_NACK) && (statnum.heard_nack_flag)) break; bufstat = (struct bufstatdef *)&buf; bufstat->head.version = VERSION; bufstat->head.type = BUF_TYPE_STAT; bufstat->head.fid = htons(statnum.file_received_flag); bufstat->head.flags = statnum.last_buf_flag; bufstat->good = htons(statnum.goodtrans); bufstat->bad = htons(statnum.badtrans); bufstat->retries = htons(statnum.retries); bufstat->packet_recv = htons(statnum.pack_recv); memcpy(bufstat->server,statnum.server,4); strcpy(bufstat->data, statnum.name_user); counter = strlen(bufstat->data) + 1; bufstat->data[counter++] = NULL; counter += BUFSTATDEF_SIZE; sendbuf( &buf, counter); statnum.alarmoff = 0; break; case BUF_TYPE_NEW_MEM: newmember = (struct bufmemberdef *) &buf; newmember->head.type = statnum.alarmoff; newmember->head.version = VERSION; newmember->head.fid = 0; newmember->head.flags = 0; counter = sizeof(struct bufmemberdef); newmember->myversion = VERSION; if (!locktransmit) { sendbuf( &buf, counter); } statnum.alarmoff = 0; break; default: buf.head.type = statnum.alarmoff; for (hostptr = hosts_tail; hostptr != NULL; hostptr = (struct hosttype *) hostptr->prev) { nodeptr = &hostptr->node; if (nodeptr->flag) continue; nodeptr->flag = TRUE; if (nodeptr->transmitcode == TRANSMISSION_REST) continue; memcpy(&buf.server[0], &nodeptr->ipaddr[0],4); data_ptr = &buf.data[0]; buf.head.version = VERSION; buf.head.fid = htons(nodeptr->fid); buf.num_requests = 0; buf.head.flags = 0; /* define the upper limit */ if (!(nodeptr->serverflags & BUF_FLG_EOF)) max = (nodeptr->bytes_written + nodeptr->buf.maxsize - 1) / nodeptr->buf.maxsize; else if (nodeptr->total != MAXFILESIZE) max = (nodeptr->total + nodeptr->buf.maxsize - 1) / nodeptr->buf.maxsize; if (debug) fprintf(logfp,"%d alarm response %d min %u max %u\n", buf.head.fid,statnum.alarmoff,nodeptr->min, max); /* setup retransmit packet */ for (counter = nodeptr->min; counter < max; counter++) { j = recoverbufs(nodeptr,counter,0,RECOV_TEST); if (j == 0) { if (buf.num_requests >= (T_MAX / sizeof(long))) break; *data_ptr++ = (u_long)htonl(counter); if ( ! buf.num_requests++) nodeptr->min = counter; } } if (buf.num_requests == 0) { continue; } /* send packet */ counter = BUFMISSDEF_SIZE + (sizeof(long) * buf.num_requests); statnum.retries += buf.num_requests; buf.num_requests = (u_long)htonl(buf.num_requests); sendbuf( &buf, counter); } statnum.alarmoff = 0; break; } } /******************************************************************************/ /* set the alarm clock to go off within the specified duration */ int setackalarm(cmd, durat,nodeptr, current_server_flag) int cmd; int durat; struct nodetype *nodeptr; { int timedur_miss = MINWAIT; static int repeatanswer = 0; int i; switch (cmd) { case BUF_TYPE_CMD: /* SEND STATUS PACKET */ repeatanswer = 0; if (statnum.alarmoff == BUF_TYPE_CMD) return FALSE; statnum.file_received_flag = nodeptr->fid; if (nodeptr->transmitcode == TRANSMISSION_REST) statnum.last_buf_flag = current_server_flag | BUF_FLG_PACK; else statnum.last_buf_flag = current_server_flag & ~BUF_FLG_PACK; /* do not allow response period greater than MAXWAIT */ if (durat > (MAXWAIT * 1000)) durat = MAXWAIT * 1000; #ifdef SOLARIS i = ((double) durat / 1000.0) * (rand_r(&seed) >> 5); #else #ifdef hpux i = ((double) durat / 1000.0) * (rand() >> 5); #else i = ((double) durat / 1000.0) * (random() % 1024); #endif #endif if (debug) fprintf(logfp,"cmd alarm set %lf sec\n",(double) (i+1)/ 1000.0); statnum.heard_nack_flag = FALSE; memcpy(statnum.server, nodeptr->ipaddr,4); signal(SIGALRM, sendack); alarm_clock((i+1)); statnum.alarmoff = BUF_TYPE_CMD; return i+1; case BUF_TYPE_MISS_REQ: /* SEND MISSING PACKET LIST */ repeatanswer = 0; recoverbufs(nodeptr, 0,0, RECOV_MISS_FREE); default: if (nodeptr != NULL) nodeptr->flag = FALSE; if ( durat != 0) timedur_miss = durat; if (durat > MAXWAIT) durat = MAXWAIT; if ((statnum.alarmoff == BUF_TYPE_MISS_REQ) || (statnum.alarmoff == BUF_TYPE_MISS_ANS)) return FALSE; if (repeatanswer++) return FALSE; #ifdef SOLARIS i = timedur_miss * (rand_r(&seed) >> 5); #else #ifdef hpux i = timedur_miss * (rand() >> 5); #else i = timedur_miss * (random() % 1024); #endif #endif if (debug) { if (nodeptr!=NULL) fprintf(logfp,"missing alarm ans id %d %lf sec\n", nodeptr->fid,(double)(i+1)/1000.0); else fprintf(logfp,"missing alarm ans %lf sec\n", (double)(i+1)/ 1000.0); } signal(SIGALRM, sendack); alarm_clock((i+1)); statnum.alarmoff = cmd; return i+1; } } /*************************************************************************/ int newmember(durat) int durat; { return setackalarm(BUF_TYPE_NEW_MEM,durat,NULL,0); } /*************************************************************************/ /* missing packet report - update recoverbuf array */ /* return variable indicates if it found redundant packets sent */ int missingreport(nodeptr, buf, lr) struct nodetype *nodeptr; struct bufmissdef *buf; int lr; { u_long i; u_long *data_ptr = &buf->data[0]; /* buf->num_requests = (unsigned long) ntohl(buf->num_requests); */ if (debug) fprintf(logfp,"detect missing packets lr %d\n",lr); for (i=0; i < buf->num_requests; i++) { /* mark buf as already reported - also discover if someelse already did */ if (debug) printf(" %d",ntohl(*data_ptr)); recoverbufs( nodeptr, ntohl(*data_ptr++),-1, RECOV_MISS); } if (debug) fprintf(logfp,"\n"); } /*************************************************************************/ /* Server Positive ack report - Just looking for my own IP */ int PackAttack(buf, datasize) struct bufdef *buf; int datasize; { u_long i; int addr_len = 4; for (i = 0; i < (datasize / addr_len); i++) { if (memcmp(myipaddress,(char *) &buf->data[i * addr_len],addr_len) == 0) return TRUE; } return FALSE; } /*************************************************************************/ /* information packet from another client server */ int statinfo(bufstat,srcaddr) struct bufstatdef *bufstat; struct sockaddr_in *srcaddr; { char ipaddress[20]; char user[MAXUSERNAME]; bufstat->good = (unsigned short)ntohs(bufstat->good); bufstat->packet_recv = (unsigned short)ntohs(bufstat->packet_recv); bufstat->bad = (unsigned short)ntohs(bufstat->bad); bufstat->retries = (unsigned short)ntohs(bufstat->retries); if ((bufstat->head.flags & BUF_FLG_NACK) && (memcmp(&statnum.server[0], &bufstat->server[0],4) == 0)) statnum.heard_nack_flag = TRUE; strncpy(ipaddress, inet_ntoa(srcaddr->sin_addr), 16); strncpy(user,bufstat->data,MAXUSERNAME); if (strchr(user,'@') == NULL) { sprintf(user,"%s@%s", bufstat->data,ipaddress); } if (debug) fprintf(logfp,"%s %s g %u b %u r %u rec %u flag %x conf %d %d\n", user, ipaddress, bufstat->good, bufstat->bad, bufstat->retries, bufstat->packet_recv, bufstat->head.flags, bufstat->head.fid, statnum.heard_nack_flag); memberadd(user, ipaddress, bufstat->good, bufstat->bad, bufstat->retries, bufstat->packet_recv); } /*************************************************************************/ /* ticket issuing system to insure each file coming in is given a unique id for this client. A simple mapping system - it returns a pointer to where all the data for that server is kept */ struct nodetype *checkhost(cmd,bufptr,srcaddr) int cmd; struct bufdef *bufptr; struct sockaddr_in *srcaddr; { static int newid = 0; static struct hosttype *hostptr; static struct hosttype *tmp_hostptr; static struct nodetype empty_node; static struct nodetype *nodeptr; static struct bufmissdef *bufmissdef_ptr; switch (cmd) { case RECOV_INIT: memset(&empty_node,0, sizeof(struct nodetype)); hosts_tail = NULL; hosts_head = NULL; return &empty_node; case RECOV_MISS_FREE: hostptr = hosts_tail; while (hostptr != NULL) { nodeptr = &hostptr->node; recoverbufs(nodeptr,0,0,RECOV_FREE); if (debug) printf("check miss free %d %s %s\n",hostptr->fid, nodeptr->currfile,nodeptr->sendername); if (nodeptr->tmpfile[0] != NULL) close(nodeptr->ofp); tmp_hostptr = hostptr; hostptr = (struct hosttype *) hostptr->prev; free(tmp_hostptr); } /* empty list */ hosts_head = NULL; hosts_tail = NULL; return &empty_node; case RECOV_FREE: hostptr = hosts_tail; while (hostptr != NULL) { nodeptr = &hostptr->node; /* resource will go away in five minutes if no activity */ if ((nodeptr->age + (5 * 60)) > time(NULL)) { hostptr = (struct hosttype *) hostptr->prev; continue; } recoverbufs(nodeptr,0,0,RECOV_FREE); if (debug) printf("check free %d %s %s\n",hostptr->fid, nodeptr->currfile,nodeptr->sendername); if (nodeptr->tmpfile[0] != NULL) { close(nodeptr->ofp); unlink(nodeptr->tmpfile); } if (hostptr->prev == NULL) { if (hostptr->next == NULL) { /* empty list */ hosts_head = NULL; hosts_tail = NULL; } else { /* at beginning of list */ tmp_hostptr = hostptr->next; tmp_hostptr->prev = NULL; hosts_head = tmp_hostptr; } }else { if (hostptr->next == NULL) { /* at end of list */ tmp_hostptr = hostptr->prev; tmp_hostptr->next = NULL; hosts_tail = tmp_hostptr; } else { /* delete from middle of list */ tmp_hostptr = hostptr->prev; tmp_hostptr->next = hostptr->next; tmp_hostptr = hostptr->next; tmp_hostptr->prev = hostptr->prev; } } tmp_hostptr = hostptr; hostptr = (struct hosttype *) hostptr->prev; free(tmp_hostptr); } return &empty_node; default: switch (bufptr->head.type) { case BUF_TYPE_ID: case BUF_TYPE_DATA: case BUF_TYPE_CMD: for (hostptr = hosts_tail; hostptr != NULL; hostptr = (struct hosttype *) hostptr->prev) { if (debug > 4) printf("check %d %d\n",hostptr->fid,bufptr->head.fid); if ((hostptr->fid == bufptr->head.fid) && (! memcmp(&hostptr->address, &srcaddr->sin_addr,4))) { return &hostptr->node; /* found id */ } } /* return NULL if EOF and we are seeing a new file */ if (bufptr->head.flags & BUF_FLG_EOF) { if (bufptr->head.type != BUF_TYPE_DATA) return NULL; if (bufptr->bytepos > (PACKETSIZE * MAXPACKETS * MAXPACKETSHIFT)) return NULL; } if ( ++newid >= MAXFILES) newid = 0; /* insert into link list */ hostptr = malloc(sizeof(struct hosttype)); if (hostptr == NULL) fatal("unable to malloc"); memset(hostptr,0,sizeof(struct hosttype)); /* insert into link list */ hostptr->next = NULL; if (hosts_tail == NULL) { hostptr->next = NULL; hostptr->prev = NULL; hosts_tail = hostptr; hosts_head = hostptr; } else { hosts_tail->next = hostptr; hostptr->prev = hosts_tail; hosts_tail = hostptr; } hostptr->fid = bufptr->head.fid; hostptr->id = newid; memcpy(hostptr->address, &srcaddr->sin_addr,4); if (debug) fprintf(logfp,"checkhost:new file %s %d %d\n", inet_ntoa(srcaddr->sin_addr), bufptr->head.fid,newid); /* initialize the new file */ nodeptr = &hostptr->node; memcpy((char *)nodeptr->ipaddr, &srcaddr->sin_addr,4); nodeptr->fid = bufptr->head.fid; nodeptr->flag = TRUE; nodeptr->hold = MINWAIT; nodeptr->recovery_cycle_time = MINWAIT; nodeptr->toggleflag = BUF_FLG_TOGGLE; nodeptr->buf.maxsize = 1; nodeptr->buf.tail = NULL; nodeptr->buf.head = NULL; nodeptr->age = time(NULL); nodeptr->total = MAXFILESIZE; xlopen(0,nodeptr); nodeptr->transmitcode = TRANSMISSION_START; return nodeptr; case BUF_TYPE_NEW_MEM: case BUF_TYPE_STAT: return &empty_node; case BUF_TYPE_MISS_REQ: case BUF_TYPE_MISS_ANS: bufmissdef_ptr = (struct bufmissdef *) bufptr; for (hostptr = hosts_tail; hostptr != NULL; hostptr = (struct hosttype *) hostptr->prev) { if ((hostptr->fid == bufmissdef_ptr->head.fid) && (! memcmp(&hostptr->address, &bufmissdef_ptr->server[0],4))) { if (debug > 4) printf("checkhost: %d -> %d\n", bufmissdef_ptr->head.fid, hostptr->id); return &hostptr->node; /* found id */ } }; } /* could not find anything - kicked it out*/ if (debug) fprintf(logfp,"ignore type %d id %d\n", bufptr->head.type,bufptr->head.fid); bufptr->head.type = BUF_TYPE_IGNORE; return &empty_node; } } /******************************************************************************/ /* 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 */ lwrite = sendto(xmitsock, (char *) buf,len,0, (struct sockaddr *) &addr_dist, len_addr_dist); if (lwrite >= 0) return lwrite; warn("sendto: Cannot send\n"); } /******************************************************************************/ int alarm_clock( wait) unsigned long wait; { static struct itimerval value; if (wait == 0) wait = 1; value.it_interval.tv_sec = 0; value.it_interval.tv_usec = 0; value.it_value.tv_sec = wait / 1000; value.it_value.tv_usec = wait % 1000; setitimer(ITIMER_REAL, &value, NULL); }