worker.c

滑动窗口协议

#include <sys/types.h>
#include <sys/stat.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/time.h>
#include <sys/select.h>
#include "common.h"

#define NR_TIMERS 8		/* number of timers */
#define MAX_QUEUE 100000	/* max number of buffered frames */
#define NO_EVENT -1		/* no event possible */
#define FRAME_SIZE (sizeof(frame))
#define BYTE 0377		/* byte mask */
#define UINT_MAX  0xFFFFFFFF	/* maximum value of an unsigned 32-bit int */
#define INTERVAL 100000		/* interval for periodic printing */
#define AUX 2			/* aux timeout is main timeout/AUX */

/*added by theNeo@THU */
#define BUFFER_SIZE  40960

/* DEBUG MASKS */
#define SENDS        0x0001	/* frames sent */
#define RECEIVES     0x0002	/* frames received */
#define TIMEOUTS     0x0004	/* timeouts */
#define PERIODIC     0x0008	/* periodic printout for use with long runs */

/* Simulation parameters. */
int protocol;			/* protocol we are simulating */
bigint timeout_interval;	/* timeout interval in ticks */
int pkt_loss;			/* controls packet loss rate: 0 to 990 */
int garbled;			/* control cksum error rate: 0 to 990 */
int debug_flags;		/* debug flags */ 

/* File descriptors for pipes. */
int r1, w1, r2, w2, r3, w3, r4, w4, r5, w5, r6, w6;

/* Filled in by main to tell each worker its id. */
int id;				/* 0 or 1 */

bigint zero;

int mrfd, mwfd, prfd;



/* Status variables used by the workers, M0 and M1. */
bigint ack_timer[NR_TIMERS];	/* ack timers */
unsigned int seqs[NR_TIMERS];	/* last sequence number sent per timer */
bigint lowest_timer;		/* lowest of the timers */
bigint aux_timer;		/* value of the auxiliary timer */
int network_layer_status;	/* 0 is disabled, 1 is enabled */
unsigned int next_net_pkt;	/* seq of next network packet to fetch */
unsigned int last_pkt_given= 0xFFFFFFFF;	/* seq of last pkt delivered*/
frame last_frame;		/* arrive frames are kept here */
int offset;			/* to prevent multiple timeouts on same tick*/
bigint tick;			/* current time */
int retransmitting;		/* flag that is set on a timeout */
int nseqs = -1;			/* must be MAX_SEQ + 1 after startup */
extern unsigned int oldest_frame;	/* tells protocol 6 which frame timed out */

char *badgood[] = {"bad ", "good"};
char *tag[] = {"Data", "Ack ", "Nak "};

/* Statistics */
int data_sent;			/* number of data frames sent */
int data_retransmitted;		/* number of data frames retransmitted */
int data_lost;			/* number of data frames lost */
int data_not_lost;		/* number of data frames not lost */
int good_data_recd;		/* number of data frames received */
int cksum_data_recd;		/* number of bad data frames received */

int acks_sent;			/* number of ack frames sent */
int acks_lost;			/* number of ack frames lost */
int acks_not_lost;		/* number of ack frames not lost */
int good_acks_recd;		/* number of ack frames received */
int cksum_acks_recd;		/* number of bad ack frames received */

int payloads_accepted;		/* number of pkts passed to network layer */
int timeouts;			/* number of timeouts */
int ack_timeouts;		/* number of ack timeouts */

/* Incoming frames are buffered here for later processing. */
frame queue[MAX_QUEUE];		/* buffered incoming frames */
frame *inp = &queue[0];		/* where to put the next frame */
frame *outp = &queue[0];	/* where to remove the next frame from */
int nframes;			/* number of queued frames */

/* Prototypes. */
int ready_test(int fd);
void wait_for_event(event_type *event);
void queue_frames(void);
int pick_event(void);
event_type frametype(void);
void from_network_layer(packet *p);
void to_network_layer(packet *p);
void from_physical_layer(frame *r);
void to_physical_layer(frame *s);
void start_timer(seq_nr k);
void stop_timer(seq_nr k);
void start_ack_timer(void);
void stop_ack_timer(void);
void enable_network_layer(void);
void disable_network_layer(void);
int check_timers(void);
int check_ack_timer(void);
unsigned int pktnum(packet *p);
void fr(frame *f);
void recalc_timers(void);
void print_statistics(void);
void sim_error(char *s);


void wait_for_event(event_type *event)
{
/* Wait_for_event reads the pipe from main to get the time.  Then it
 * fstat's the pipe from the other worker to see if any
 * frames are there.  If so, if collects them all in the queue array.
 * Once the pipe is empty, it makes a decision about what to do next.
 */
 
 bigint ct, word = OK;

  if (nseqs < 0) nseqs = oldest_frame;	/* need MAX_SEQ+1 for protocol 6 */
  offset = 0;			/* prevents two timeouts at the same tick */
  retransmitting = 0;		/* counts retransmissions */
  while (true) {
	queue_frames();		/* go get any newly arrived frames */
	if (write(mwfd, &word, TICK_SIZE) != TICK_SIZE) print_statistics();
	if (read(mrfd, &ct, TICK_SIZE) != TICK_SIZE) print_statistics();
	if (ct == 0) print_statistics();
	tick = ct;		/* update time */
	if ((debug_flags & PERIODIC) && (tick%INTERVAL == 0))
		printf("Tick %u. Proc %d. Data sent=%d  Payloads accepted=%d  Timeouts=%d\n", 
		       tick/DELTA, id, data_sent, payloads_accepted, timeouts);

	/* Now pick event. */
	*event = pick_event();
	if (*event == NO_EVENT) {
		word = (lowest_timer == 0 ? NOTHING : OK);
		continue;
	}
	word = OK;
	if (*event == timeout) {
		timeouts++;
		retransmitting = 1;	/* enter retransmission mode */
		if (debug_flags & TIMEOUTS)
		      printf("Tick %u. Proc %d got timeout for frame %d\n",
					       tick/DELTA, id, oldest_frame);
	}

	if (*event == ack_timeout) {
		ack_timeouts++;
		if (debug_flags & TIMEOUTS)
		      printf("Tick %u. Proc %d got ack timeout\n",
					       tick/DELTA, id);
	}
	return;
  }
}

char buffer[BUFFER_SIZE]; /* store incoming datas */
int startpos = 0;  /*if no data is available, startpos = 0
		     if unconsecutive data is available, startpos is
		     where the broken point is */

void queue_frames(void)
{
/* See if any frames from the peer have arrived; if so get and queue them.
 * Queue_frames() sucks frames out of the pipe into the circular buffer,
 * queue[]. It first fstats the pipe, to avoid reading from an empty pipe and
 * thus blocking.  If inp is near the top of queue[], a single call here
 * may read a few frames into the top of queue[] and then some more starting
 * at queue[0].  This is done in two read operations.
 *
 *  this function was updated by theNeo@NET9/THU, using select() instead of
 *  fstat(), to avoid malfunction with 2.6.x and mach kernel.
 */

  int prfd, frct, k;
  frame *top;
  struct stat statbuf;
  

  ssize_t nread; /* number of bytes read from pipe */
  int n;
  

  prfd = (id == 0 ? r2 : r1);	/* which file descriptor is pipe on */

  
  /* if (fstat(prfd, &statbuf) < 0) sim_error("Cannot fstat peer pipe");*/
  if (ready_test(prfd)) {
    /* pipe is available, first we read the data into buffer */
    nread = read(prfd, buffer + startpos, BUFFER_SIZE - startpos);
    
    frct = (nread + startpos) / FRAME_SIZE; /* number of complete frames */
    n = (nread + startpos) % FRAME_SIZE; /* length of the incomplete frame 
					    at the end of buffer*/

    if (frct > 0) { 
      /* there's at least one complete frame */
      top = (outp <= inp ? &queue[MAX_QUEUE] : outp); /* how far can we rd?*/
      k = top - inp;	/* number of frames that can be read consecutively */
      if (k > frct) k = frct;	/* how many frames to read from peer */
      memcpy(inp, buffer, FRAME_SIZE * k); /* fill the queue */
      frct -= k;
      inp += k;
      if (inp == &queue[MAX_QUEUE]) inp = queue;
      nframes += k;
      if (frct > 0) { 
	/* still some frames left in buffer*/
	memcpy(inp, buffer + k * FRAME_SIZE, frct * FRAME_SIZE);
	nframes += frct;
	inp = &queue[frct];
      } 

      /* deal with the tail, move incomplete frame to 
	 the beginning of the buffer*/
      if ( n == 0 ) /* no tail */
	startpos = 0;
      else {
	memcpy(buffer, buffer + startpos + nread - n, n);
	startpos = n;
      }
    } else {
      /* still cannot form a complete frame, so just update startpos*/
      startpos += nread;
    }
  } else {
    frct = 0;
  }

 
}


int pick_event(void)
{
/* Pick a random event that is now possible for the process.
 * The set of legal events depends on the protocol number and system state.
 * A timeout is not possible, for example, if no frames are outstanding.
 * For each protocol, events from 0 to some protocol-dependent maximum
 * are potentially allowed.  The maximum is given by highest_event.  The
 * events that are theoretically possible are given below.
 *
 *  # Event		Protocols:  1 2 3 4 5 6
 *  0 frame_arrival                 x x x x x x
 *  1 chksum_err                        x x x x
 *  2 timeout                           x x x x
 *  3 network_layer_ready                   x x 
 *  4 ack_timeout                             x (e.g. only 6 gets ack_timeout)
 *
 * Note that the order in which the tests is made is critical, as it gives
 * priority to some events over others.  For example, for protocols 3 and 4
 * frames will be delivered before a timeout will be caused.  This is probably
 * a reasonable strategy, and more closely models how a real line works.
 */

  switch(protocol) {
    case 2:			/* {frame_arrival} */
	if (nframes == 0 && lowest_timer == 0) return(NO_EVENT);
	return(frametype());

    case 3:			/* {frame_arrival, cksum_err, timeout} */
    case 4:
	if (nframes > 0) return((int)frametype());
	if (check_timers() >= 0) return(timeout);	/* timer went off */
	return(NO_EVENT);

    case 5:	/* {frame_arrival, cksum_err, timeout, network_layer_ready} */
	if (nframes > 0) return((int)frametype());
	if (network_layer_status) return(network_layer_ready);
	if (check_timers() >= 0) return(timeout);	/* timer went off */
	return(NO_EVENT);

    case 6:	/* {frame_arrival, cksum_err, timeout, net_rdy, ack_timeout}*/
	if (check_ack_timer() > 0) return(ack_timeout);
	if (nframes > 0) return((int)frametype());
	if (network_layer_status) return(network_layer_ready);
	if (check_timers() >= 0) return(timeout);	/* timer went off */
	return(NO_EVENT);
  }
}


event_type frametype(void)
{
/* This function is called after it has been decided that a frame_arrival
 * event will occur.  The earliest frame is removed from queue[] and copied
 * to last_frame.  This copying is needed to avoid messing up the simulation
 * in the event that the protocol does not actually read the incoming frame.
 * In protocols 2 and 3, the senders do not call from_physical_layer() to
 * collect the incoming frame.  If frametype() did not remove incoming frames
 * from queue[], they never would be removed.  Of course, one could change
 * sender2() and sender3() to have them call from_physical_layer(), but doing
 * it this way is more robust.
 *
 * This function determines (stochastically) whether the arrived frame is good
 * or bad (contains a checksum error).
 */

  int n, i;
  event_type event;

  /* Remove one frame from the queue. */
  last_frame = *outp;		/* copy the first frame in the queue */
  outp++;
  if (outp == &queue[MAX_QUEUE]) outp = queue;
  nframes--;

  /* Generate frames with checksum errors at random. */
  n = rand() & 01777;
  if (n < garbled) {
	/* Checksum error.*/
	event = cksum_err;
	if (last_frame.kind == data) cksum_data_recd++;
	if (last_frame.kind == ack) cksum_acks_recd++;
	i = 0;
  } else {
	event = frame_arrival;
	if (last_frame.kind == data) good_data_recd++;
	if (last_frame.kind == ack) good_acks_recd++;
	i = 1;
  }

  if (debug_flags & RECEIVES) {
	printf("Tick %u. Proc %d got %s frame:  ",