tcp.java

一个小型网络仿真器的实现

/*
   JaNetSim  ---  Java Network Simulator
   -------------------------------------

   This software was developed at the Network Research Lab, Faculty of
   Computer Science and Information Technology (FCSIT), University of Malaya.
   This software may be used and distributed freely. FCSIT assumes no responsibility
   whatsoever for its use by other parties, and makes no guarantees, expressed or
   implied, about its quality, reliability, or any other characteristic.

   We would appreciate acknowledgement if the software is used.

   FCSIT ALLOWS FREE USE OF THIS SOFTWARE IN ITS "AS IS" CONDITION AND
   DISCLAIM ANY LIABILITY OF ANY KIND FOR ANY DAMAGES WHATSOEVER RESULTING
   FROM THE USE OF THIS SOFTWARE.
*/

package janetsim.component;

import janetsim.*;

/*
  Implementation of TCP for JaNetSim based on RFCs 793, 2581 & 2988
  -----------------------------------------------------------------
  Please send bug report to:
      Lim Shiau Hong
      [e-mail]

  Some notes on the implementation:
  1. No "quiet time" imposed (because no hosts can crash :)
  2. No security / precedence is implemented
  3. No other stuff can be included during connection setup handshake
      (in the SYN and associated ACKs)
  4. Passive sockets can not specify range of possible remote sockets
      (i.e. must accept all sockets)
  5. Urgent bit & pointer not implemented
  6. User timeout not implemented
  7. Receiver window size is always advertised as the maximum window size
     (as specified by the TCPProvider)
*/

public class TCP implements SimListener,java.io.Serializable {

  private class TCBKey implements java.io.Serializable {
    int source_ip,dest_ip;
    int source_port,dest_port;
    TCBKey(int srcip,int srcport,int destip,int destport) {
      source_ip=srcip;
      source_port=srcport;
      dest_ip=destip;
      dest_port=destport;
    }
    public boolean equals(Object o) {
      if(!(o instanceof TCBKey)) return false;
      TCBKey k=(TCBKey)o;
      if(k.source_ip==source_ip && k.source_port==source_port &&
          k.dest_ip==dest_ip && k.dest_port==dest_port)
        return true;
      return false;
    }
    public int hashCode() {
      return (source_ip ^ source_port ^ dest_ip ^ dest_port);
    }
  }

  private class TCB implements java.io.Serializable {
    //socket info
    TCBKey key=null;
    boolean passive=false;

    //user
    TCPUser user=null;
    boolean userclosing=false;

    //user buffers
    java.util.List input_q=null; //outgoing data (from user to interface)
    java.util.List output_q=null; //incoming data (from interface to user)
    int userbuf_len=0; //size of receiving user buffer
    int userbuf_seq=0; //beginning seq # of output_q

    //timers
    VirtualTimer retrans_timer=null;
    VirtualTimer timewait_timer=null;

    //retransmission & resequencing queue
    java.util.List retrans_q=null;
    java.util.List reseq_q=null;

    //status info
    SimParamIntTag status;
    SimParamInt snd_una, snd_nxt, snd_wnd;
    SimParamInt snd_wl1, snd_wl2;
    SimParamInt iss;
    SimParamInt rcv_nxt, rcv_wnd;
    SimParamInt irs;
    SimParamInt cwnd;
    SimParamInt ssthresh;
    SimParamDouble rttvar,srtt,rto;
    int rtt_seq=0;
    long rtt_tick=0;
    boolean rtt_on=false;
    int dup_ack=0; //duplicate ACK counter
    boolean fast_retrans_recovery=false;
    long last_sent_tick=0;
  }

  private class VirtualTimer implements java.io.Serializable {
    private int eventtype;
    private double tempo;
    private SimListener listener;
    private Object params;
    private SimEvent olde=null;

    //tempo in seconds
    VirtualTimer(int evtype,double atempo,SimListener lst,Object prms) {
      eventtype=evtype;
      tempo=atempo;
      listener=lst;
      params=prms;
    }
    void setTempo(double t) { //tempo in seconds
      tempo=t;
    }
    //start also same as restart
    void start() {
      if(olde!=null) { //there is an old timer running, need to stop it
        comp.getSim().dequeue(olde);
        olde=null;
      }
      olde=new SimEvent(eventtype,listener,listener,comp.getSim().now() +
                          SimClock.Sec2Tick(tempo),params);
      comp.getSim().enqueue(olde);
    }
    void stop() {
      if(olde!=null) {
        comp.getSim().dequeue(olde);
        olde=null;
      }
    }
    //must call this when receive the timer event
    void done() {
      olde=null;
    }
    boolean isRunning() {
      return (olde!=null);
    }
  }

  private class ProcessEntry implements java.io.Serializable {
    IPPacket packet;
    boolean incoming;
    SimComponent fromlink;

    ProcessEntry(IPPacket p,boolean in,SimComponent frmlink) {
      packet=p;
      incoming=in;
      fromlink=frmlink;
    }
  }

  //packet processing queue
  private java.util.List processing_q=null;
  private boolean busy;

  private long init_seq;
  private java.util.Random randgen;
  private java.util.Map tcbs=null;
  private SimComponent comp=null;
  private TCPProvider provider=null;

/////////// default timeouts ///////////////

  private static final int DEFAULT_RETRANS_TIMEOUT = 3; //seconds
  private static final int DEFAULT_TIMEWAIT_TIMEOUT = 120; //seconds

///////////// status constants /////////////

  private static final int STAT_CLOSED = 0;
  private static final int STAT_LISTEN = 1;
  private static final int STAT_SYN_RCVD = 2;
  private static final int STAT_SYN_SENT = 3;
  private static final int STAT_ESTABLISHED = 4;
  private static final int STAT_FIN_WAIT1 = 5;
  private static final int STAT_FIN_WAIT2 = 6;
  private static final int STAT_CLOSING = 7;
  private static final int STAT_TIME_WAIT = 8;
  private static final int STAT_CLOSE_WAIT = 9;
  private static final int STAT_LAST_ACK = 10;

//private events
  private static final int MY_READY_DEMUX = SimProvider.EV_PRIVATE + 1;
  private static final int MY_RECEIVE_BUF = SimProvider.EV_PRIVATE + 2;
  private static final int MY_RETRANS_TIMEOUT = SimProvider.EV_PRIVATE + 3;
  private static final int MY_TIMEWAIT_TIMEOUT = SimProvider.EV_PRIVATE + 4;
  private static final int MY_NOTIFY_USER = SimProvider.EV_PRIVATE + 5;


  public TCP(SimComponent owner,TCPProvider prov) {
    randgen=new java.util.Random(owner.getName().hashCode());

    //initialized the init_seq
    init_seq=randgen.nextLong();

    processing_q=new java.util.LinkedList();
    busy=false;

    tcbs=new java.util.HashMap();
    comp=owner;
    provider=prov;
  }

  public void reset() {
    randgen.setSeed(comp.getName().hashCode());
    init_seq=randgen.nextLong();

    processing_q.clear();
    busy=false;

    tcbs.clear();
    provider.statusChanged();
  }

  public void action(SimEvent e) {
    switch(e.getType()) {
      case MY_READY_DEMUX:
        my_ready_demux();
        break;
      case MY_RECEIVE_BUF:
        receive_buf(e);
        break;
      case MY_RETRANS_TIMEOUT:
        retrans_timeout(e);
        break;
      case MY_TIMEWAIT_TIMEOUT:
        timewait_timeout(e);
        break;
      case MY_NOTIFY_USER:
        Object [] params=(Object [])e.getParams();
        TCB tcb=(TCB)params[0];
        int ev=((Integer)params[1]).intValue();
        tcb.user.notify(ev,params[2]);
        break;
    }
  }

  //IMPORTANT: this method is called by provider
  public void receive_ip(IPPacket packet,SimComponent fromlink) {
    processing_q.add(new ProcessEntry(packet,true,fromlink));
    check_processing_q();
  }

  private void send_ip(IPPacket packet) { //called by tcp
    processing_q.add(new ProcessEntry(packet,false,null));
    check_processing_q();
  }

  private void my_ready_demux() {
    //ready to process the next packet (process delay already accounted for)

    if(processing_q.isEmpty()) {
      System.out.println("Warning: tcp: demux called when processing_q is empty!");
      return;
    }

    ProcessEntry entry=(ProcessEntry)processing_q.remove(0);
    if(entry.incoming)
      process_packet(entry.packet,entry.fromlink);
    else
      transmit_packet(entry.packet);

    busy=false;
    check_processing_q();
  }

  private void check_processing_q() {
    if(busy) return;

    if(!processing_q.isEmpty()) {
      long ticks=compute_processing_delay();
      comp.getSim().enqueue(new SimEvent(MY_READY_DEMUX,this,this,comp.getSim().now()+ticks,null));
      busy=true;
    }
  }

////helpers for mod 2^32 operations
  private boolean modLE(int a,int b) {
    int delta=b-a;
    return (delta>=0);
  }
  private boolean modLT(int a,int b) {
    int delta=b-a;
    return (delta>0);
  }
  private boolean modGE(int a,int b) {
    int delta=a-b;
    return (delta>=0);
  }
  private boolean modGT(int a,int b) {
    int delta=a-b;
    return (delta>0);
  }
////end helpers

  private void send_new_data(TCB tcb) {
    if(tcb.input_q.isEmpty()) { //nothing to send, must send ack
      if(tcb.userclosing)
        send_fin(tcb);
      else
        output_pkt(tcb,tcb.snd_nxt.getValue(),0,null,false);
      return;
    }

    //test if not been sending for more than RTO
    if(SimClock.Tick2Sec(comp.getSim().now()-tcb.last_sent_tick) >
        tcb.rto.getValue()) {
      //set cwnd back to initial window size
      tcb.cwnd.setValue(provider.getMaxSegmentSize(),comp.getSim().now(),provider.getLogFactor());
    }

    boolean push;
    int len,offset;
    int max_to_send_offset = tcb.snd_una.getValue() + Math.min(tcb.cwnd.getValue(),tcb.snd_wnd.getValue()) - 1;

    if (modLE(tcb.snd_nxt.getValue(),max_to_send_offset)) {
      if(!tcb.rtt_on) { //start rtt measurement
        tcb.rtt_seq=tcb.snd_nxt.getValue();
        tcb.rtt_tick=comp.getSim().now();
        tcb.rtt_on=true;
      }
      tcb.last_sent_tick=comp.getSim().now(); //record last sending time
      // send all we can
      while (modLE(tcb.snd_nxt.getValue(),max_to_send_offset) && !tcb.input_q.isEmpty()) {
        push=false;
        len = Math.min(provider.getMaxSegmentSize(),
                        max_to_send_offset - tcb.snd_nxt.getValue() + 1);
        java.util.List payload=new java.util.LinkedList();
        int buflen=0;
        while (buflen<len) {
          TCPBuffer buf=(TCPBuffer)tcb.input_q.get(0);
          if(buflen+buf.len>len) { //this buffer too big, chop it...
            //must add the first half in the retrans q
            TCPBuffer tempbuf=new TCPBuffer();
            tempbuf.len=len-buflen;
            tempbuf.push=buf.push;
            tcb.retrans_q.add(tempbuf);
            payload.add(tempbuf.makeclone());

            buf.len-=tempbuf.len; //update this buf
            if(buf.push) push=true;
            break;
          }
          tcb.retrans_q.add(tcb.input_q.remove(0));
          payload.add(buf.makeclone());
          buflen+=buf.len;
          if(buf.push) push=true;

          if(tcb.input_q.isEmpty()) { //all sent
            len=buflen;
            break;
          }
        }

        offset = tcb.snd_nxt.getValue();
        tcb.snd_nxt.setValue(offset+len,comp.getSim().now(),provider.getLogFactor());
        output_pkt(tcb,offset,len,payload,push);
      }
    }
    else { //send ACK if can't send data
      output_pkt(tcb,tcb.snd_nxt.getValue(),0,null,false);
    }