tcp-sock.cc

升级版的p2p文件共享linux环境ns2仿真软件

/*-------------------------------------------------------------------------*/
/* SocketTcp: subclass of AdvwTcpAgent that implements the services        */
/* provided by real system TCP socket, e.g.,                               */
/*      limited sender buffer,                                             */
/*      spawning new connections on request,                               */
/*      real payload transfer.                                             */ 
/* Author: Qi He <http://www.cc.gatech.edu/~qhe> 01Aug2003                 */
/* $Revision:$ $Name:$ $Date:$                                             */
/*-------------------------------------------------------------------------*/
#include "tcp-sock.h"
#include "ip.h"
#include "flags.h"

#include "nssocket.h"

#ifdef PDNS
extern NsObject* GetLocalIP(ipaddr_t ipaddr);
#endif

#define MIN(a, b)	((a) < (b) ? (a) : (b))
static int pfcnt=0, pdfcnt=0, pcnt=0, opcnt=0, apcnt=0, dropcnt=0;

static class SocketTcpClass: public TclClass {
public:
	SocketTcpClass(): TclClass("Agent/TCP/FullTcp/AdvwTcp/SocketTcp") {}
	TclObject* create(int, const char*const*argv) {
		return (new SocketTcp());
	}
} class_socket_tcp;

PktDataEntry::PktDataEntry(int seqno, PacketData *p) {
  key_ = seqno;

  if(p)
    pktdata_ = (PacketData *)p->copy(); 
  else
    pktdata_ = NULL;

}

SocketTcp::SocketTcp(): AdvwTcpAgent() {
	bind("inf_rcv_", &infinite_rcv_);
	bind("snd_wnd_", &snd_wnd_);
	listen_only_ = FALSE; 
}

int SocketTcp::command(int argc, const char*const* argv) {
	if(argc==3) {
		if(strcmp(argv[1], "listen")==0) {
			listen(atoi(argv[2]));
			return TCL_OK;
		}
		if(strcmp(argv[1], "send-dummy")==0) {
			send_dummy(atoi(argv[2]));
			return TCL_OK;
		}
	}

	return AdvwTcpAgent::command(argc, argv);
}	

int SocketTcp::send_dummy(int nbytes) {
	PacketData *p = new PacketData(nbytes);
	sendmsg(p, nbytes);	
}

int SocketTcp::sendmsg(PacketData *p, int len) {
  int res;

  res = r_sendmsg(p, 0);
  if(p)
    delete p;
  return res;
}

int SocketTcp::r_send(int bytes) {
  return r_sendmsg(bytes, NULL);
}

int SocketTcp::r_send(PacketData *p) {
  return r_sendmsg(p, 0);
}


int SocketTcp::r_sendmsg(int nbytes, const char *flags = 0) {
	if (flags && strcmp(flags, "MSG_EOF") == 0) 
		close_on_empty_ = TRUE;
	if (nbytes == -1) {
		infinite_send_ = TRUE;
		return r_advance_bytes(0);
	} else
		return r_advance_bytes(nbytes);
}

int SocketTcp::r_sendmsg(PacketData *p, const char *flags = 0 ) {
	if (flags && strcmp(flags, "MSG_EOF") == 0) 
		close_on_empty_ = TRUE;
	
	return r_advance_pkt(p);
}

void SocketTcp::newack(Packet* pkt)
{
	hdr_tcp *tcph = hdr_tcp::access(pkt);

	register int ackno = tcph->ackno();
	int progress = (ackno > highest_ack_);

	if (ackno == maxseq_) {
		cancel_rtx_timer();	// all data ACKd
	} else if (progress) {
		set_rtx_timer();
	}

	//advance the ack number if this is for new data
	if (progress) {
		highest_ack_ = ackno;

		Islist_iter<PktDataEntry> pkt_iter(snd_buf_);
		PktDataEntry *cur, *last=NULL;

		while((cur=pkt_iter())!=NULL) {
		  if(cur->key_ < ackno) {
		    delete cur->pktdata_;
		    snd_buf_.remove(cur, last);
		  } else
		    break;
		}
		if(app_) {
		  ((NSSocket *)app_)->upcall_send();
		}

	}

	// if we have suffered a retransmit timeout, t_seqno_
	// will have been reset to highest_ ack.  If the
	// receiver has cached some data above t_seqno_, the
	// new-ack value could (should) jump forward.  We must
	// update t_seqno_ here, otherwise we would be doing
	// go-back-n.

	if (t_seqno_ < highest_ack_)
		t_seqno_ = highest_ack_; // seq# to send next

        /*
         * Update RTT only if it's OK to do so from info in the flags header.
         * This is needed for protocols in which intermediate agents
         * in the network intersperse acks (e.g., ack-reconstructors) for
         * various reasons (without violating e2e semantics).
         */
        hdr_flags *fh = hdr_flags::access(pkt);
        if (!fh->no_ts_) {
                if (ts_option_) {
			recent_age_ = now();
			recent_ = tcph->ts();
                        rtt_update(now() - tcph->ts_echo());
		} else if (rtt_active_ && ackno > rtt_seq_) {
			// got an RTT sample, record it
                        t_backoff_ = 1;
                        rtt_active_ = FALSE;
			rtt_update(now() - rtt_ts_);
                }
        } else {
		printf("no_ts\n");
	}
	return;
}

int SocketTcp::foutput(int seqno, int reason)
{
	int quiet = (highest_ack_ == maxseq_);
	int pflags = outflags();
	int syn = (seqno == iss_);
	int emptying_buffer = FALSE;
	int win = window() * maxseg_;	// window (in bytes)
	int datalen;
	PacketData *data = NULL;

	PktDataEntry *pe = snd_buf_.find(seqno);
	if(pe) {
	  data = pe->pktdata_;
	}

	if ((syn && !data_on_syn_) || pe==NULL || data==NULL) 
	  datalen = 0;
	else
	  datalen = data->size();
	//
	// this is an option that causes us to slow-start if we've
	// been idle for a "long" time, where long means a rto or longer
	// the slow-start is a sort that does not set ssthresh
	//

	if (slow_start_restart_ && quiet && datalen > 0) {
		if (idle_restart()) {
			slowdown(CLOSE_CWND_INIT);
		}
	}

	//
	// see if sending this packet will empty the send buffer
	// a dataless SYN packet counts also
	//
	if (!infinite_send_ && ((seqno + datalen) > curseq_ || 
	    (syn && datalen == 0))) {
		emptying_buffer = TRUE;
		//
		// if not a retransmission, notify application that 
		// everything has been sent out at least once.
		//
		if (!syn) {
			idle();
			if (close_on_empty_ && quiet) {
				flags_ |= TF_NEEDCLOSE;
			}
		}
		pflags |= TH_PUSH;
		//
		// if close_on_empty set, we are finished
		// with this connection; close it
		//
	} else  {
		/* not emptying buffer, so can't be FIN */
		pflags &= ~TH_FIN;
	}

	pflags |= TH_PUSH;

	/* sender SWS avoidance (Nagle) */
send:

	syn = (pflags & TH_SYN) ? 1 : 0;
	int fin = (pflags & TH_FIN) ? 1 : 0;

	sendpacket(seqno, rcv_nxt_, pflags, datalen, data, reason);
        /*      
         * Data sent (as far as we can tell).
         * Any pending ACK has now been sent.
         */      
	flags_ &= ~(TF_ACKNOW|TF_DELACK);

	/*
	 * if we have reacted to congestion recently, the
	 * slowdown() procedure will have set cong_action_ and
	 * sendpacket will have copied that to the outgoing pkt
	 * CACT field. If that packet contains data, then
	 * it will be reliably delivered, so we are free to turn off the
	 * cong_action_ state now  If only a pure ACK, we keep the state
	 * around until we actually send a segment
	 */

	int reliable = datalen + syn + fin; // seq #'s reliably sent
	if (cong_action_ && reliable > 0)
		cong_action_ = FALSE;

	/*
	 * SYNs and FINs each use up one sequence number, but
	 * there's no reason to advance t_seqno_ by one for a FIN
	 */

	if (!fin && seqno == t_seqno_) {
		t_seqno_ += reliable;
	}

	// highest: greatest sequence number sent + 1
	//	and adjusted for SYNs and FINs which use up one number

	int highest = seqno + reliable;
	if (highest > maxseq_) {
		maxseq_ = highest;
		//
		// if we are using conventional RTT estimation,
		// establish timing on this segment
		//
		if (!ts_option_ && rtt_active_ == FALSE) {
			rtt_active_ = TRUE;	// set timer
			rtt_seq_ = seqno; 	// timed seq #
			rtt_ts_ = now();	// when set
		}
	}

	/*
	 * Set retransmit timer if not currently set,
	 * and not doing an ack or a keep-alive probe.
	 * Initial value for retransmit timer is smoothed
	 * round-trip time + 2 * round-trip time variance.
	 * Future values are rtt + 4 * rttvar.
	 */
	if (rtx_timer_.status() != TIMER_PENDING && reliable) {
		set_rtx_timer();  // no timer pending, schedule one
	}

	if (flags_ & TF_NEEDCLOSE) {
		flags_ &= ~TF_NEEDCLOSE;
		if (state_ <= TCPS_ESTABLISHED && state_ != TCPS_CLOSED)
			usrclosed();
	}

	return reliable;
}

void SocketTcp::recvBytes(int nbytes)
{
        int rcv_buf_free, byte_read=0, seqno, acksent = FALSE;
	PktDataEntry *cur, *last=NULL;
	PacketData *pkt;

	num_bytes_avail_ += nbytes;
	Islist_iter<PktDataEntry> pkt_iter(rcv_buf_);
	while(((cur=pkt_iter())!=NULL) && num_bytes_avail_ > 0) {
	  pkt = cur->pktdata_;
	  
	  if(app_) {
	    if(pkt==NULL) {
	      printf("ERROR: Rcv Buffer Entry without Pkt\n");
	      break;
	    }   
	    byte_read = ((NSSocket*)app_)->upcall_recv(pkt);
	  } else
	    byte_read = pkt->size();

	  if(byte_read==0) 
	    break;

	  num_bytes_avail_ -= byte_read;
	  num_bytes_req_ -= byte_read;
	  if (byte_read > 0) {
	    delete pkt;
	    rcv_buf_.remove(cur, last);
	  }
	
	  if(num_bytes_avail_<0) {
	    fprintf(stderr,"SocketTcp::recvBytes(): agent %d %d num_bytes_avail_ %d < 0\n", here_.addr_, here_.port_, num_bytes_avail_);
	    break;
	  }
	
	  rcv_buf_free = rcv_buff_ - num_bytes_avail_;
	  if(rcv_buf_free >= rcv_wnd_ + MIN (rcv_buff_/2, maxseg_) && !acksent){
	    rcv_wnd_ = rcv_buf_free;
	    acksent = TRUE;
	    sendpacket(0, rcv_nxt_, TH_ACK, 0, NULL, REASON_NORMAL);
	  }

	};
}

void SocketTcp::insert(Islist<PktDataEntry> *list, PktDataEntry *pe) {
  Islist_iter<PktDataEntry> pkt_iter(*list);
  PktDataEntry *last=NULL, *cur;
  
  while((cur=pkt_iter())!=NULL && cur->key_ < pe->key_) {
    last = cur;
  };
  
  if(last!=NULL) {
    if(cur==NULL) {
      list->append(pe);
      return;
    }
    if(cur->key_==pe->key_) {
      free(pe);
      return;
     }
    pe->next_ = last->next_;
     last->next_ = pe;
  } else {
    if(cur && cur->key_==pe->key_) {
      free(pe);
      return;
    }
    list->insert(pe);
  }
}




void SocketTcp::recv(Packet *pkt, Handler* handler)
{

       if(handler) { 
	 int nb = ((PacketData *)pkt->userdata())->size();
	 if(r_send(((PacketData *)pkt->userdata()))>0)
	   handler->handle(NULL);
	 return;
       }

        hdr_ip *iph = hdr_ip::access(pkt);
  	hdr_tcp *tcph = hdr_tcp::access(pkt);
	hdr_cmn *th = hdr_cmn::access(pkt);
	hdr_flags *fh = hdr_flags::access(pkt);

	int needoutput = FALSE;
	int ourfinisacked = FALSE;
	int dupseg = FALSE;
	int todrop = 0;

	last_state_ = state_;

	int datalen = th->size() - tcph->hlen(); // # payload bytes
	int ackno = tcph->ackno();		 // ack # from packet
	int tiflags = tcph->flags() ; 		 // tcp flags from packet
	
	if (state_ == TCPS_CLOSED)
		goto drop;

	if (state_ != TCPS_LISTEN)
		dooptions(pkt);

	if(!listen_only_) {
	  if (delack_interval_ > 0.0 &&
	      (delack_timer_.status() != TIMER_PENDING)) {
	    int last = int(now() / delack_interval_);
	    delack_timer_.resched(delack_interval_ * (last + 1.0) - now());
	  }
	}

	wnd_ = (double)tcph->advwin()/(double)size_;

	if (fh->ce() && !fh->ect()) {
	    fprintf(stderr,
	    "%f: FullTcpAgent::recv(%s): warning: CE bit on, but ECT false!\n",
		now(), name());
	}

	if(listen_only_ && state_ != TCPS_LISTEN) {
	  printf("listening socket not in TCPS_LISTEN\n");
	  return;
	} 
	switch (state_) {

        /*
         * If the segment contains an ACK then it is bad and do reset.
         * If it does not contain a SYN then it is not interesting; drop it.
         * Otherwise initialize tp->rcv_nxt, and tp->irs, iss is already
	 * selected, and send a segment:
         *     <SEQ=ISS><ACK=RCV_NXT><CTL=SYN,ACK>
         * Initialize tp->snd_nxt to tp->iss.
         * Enter SYN_RECEIVED state, and process any other fields of this
         * segment in this state.
         */