tcp.java

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

            newpacket.destIP=packet.sourceIP;
            newpacket.protocol=IPPacket.PRO_TCP;
            newpacket.len=46;
            newpacket.payload=newseg;
            send_ip(newpacket);

            // SH added: I think the initial window size should be recorded
            tcb.snd_wnd.setValue(seg.seg_wnd,comp.getSim().now(),provider.getLogFactor());
            // SH end

            //must notify user about this!
            Object [] msg=new Object[3];
            msg[0]=new Integer(packet.sourceIP);
            msg[1]=new Integer(seg.sourceport);
            msg[2]=fromlink;
            tcb.user.notify(TCPUser.CON_ESTABLISHED,msg);

            provider.statusChanged();
          }
          else {
            tcb.status.setValue(STAT_SYN_RCVD);

            TCPSegment newseg=new TCPSegment();
            newseg.sourceport=seg.destport;
            newseg.destport=seg.sourceport;
            newseg.seg_seq=tcb.iss.getValue();
            newseg.seg_ack=tcb.rcv_nxt.getValue();
            newseg.syn=true;
            newseg.ack=true;
            newseg.seg_wnd=tcb.rcv_wnd.getValue();
            newseg.seg_len=1;
            IPPacket newpacket=new IPPacket();
            newpacket.sourceIP=packet.destIP;
            newpacket.destIP=packet.sourceIP;
            newpacket.protocol=IPPacket.PRO_TCP;
            newpacket.len=46;
            newpacket.payload=newseg;
            send_ip(newpacket);

            provider.statusChanged();
          }
        }
        return;
    }

  //From here on, the state should be all except CLOSED, LISTEN and SYN_SENT
    //segment acceptability test
    if(!isAcceptable(seg,tcb)) {
      if(!seg.rst) {
        TCPSegment newseg=new TCPSegment();
        newseg.sourceport=seg.destport;
        newseg.destport=seg.sourceport;
        newseg.seg_seq=tcb.snd_nxt.getValue();
        newseg.seg_ack=tcb.rcv_nxt.getValue();
        newseg.ack=true;
        newseg.seg_wnd=tcb.rcv_wnd.getValue();
        newseg.seg_len=0;
        IPPacket newpacket=new IPPacket();
        newpacket.sourceIP=packet.destIP;
        newpacket.destIP=packet.sourceIP;
        newpacket.protocol=IPPacket.PRO_TCP;
        newpacket.len=46;
        newpacket.payload=newseg;
        send_ip(newpacket);
      }
      return;
    }

    //now chop until the packet is "idealized"
    chop_segment_to_window(seg,tcb);

    //ok, if sequence is in future, queue it in reseq_q
    if(modGT(seg.seg_seq,tcb.rcv_nxt.getValue())) {
      //send a intentional duplicate ACK before that
      TCPSegment newseg=new TCPSegment();
      newseg.sourceport=seg.destport;
      newseg.destport=seg.sourceport;
      newseg.seg_seq=tcb.snd_nxt.getValue();
      newseg.ack=true;
      newseg.seg_ack=tcb.rcv_nxt.getValue();
      newseg.seg_wnd=tcb.rcv_wnd.getValue();
      newseg.seg_len=0;
      IPPacket newpacket=new IPPacket();
      newpacket.sourceIP=packet.destIP;
      newpacket.destIP=packet.sourceIP;
      newpacket.protocol=IPPacket.PRO_TCP;
      newpacket.len=46;
      newpacket.payload=newseg;
      send_ip(newpacket);

      java.util.ListIterator i=tcb.reseq_q.listIterator();
      while(i.hasNext()) {
        TCPSegment segin_q=(TCPSegment)i.next();
        if(modLT(seg.seg_seq,segin_q.seg_seq)) {
          i.previous();
          i.add(seg);
          return;
        }
      }
      i.add(seg);
      return;
    }

  ////now, must process each segment in the reseq_q (if possible)
    boolean ack_pending=false;
    do {
      //second step, check RST bit
      switch(tcb.status.getValue()) {
        case STAT_SYN_RCVD:
          if(seg.rst) {
            tcb.retrans_timer.stop();
            tcb.retrans_q.clear();

            //SH added: these should also be cleared
            tcb.input_q.clear();
            tcb.output_q.clear();
            tcb.reseq_q.clear();
            //SH end

            if(tcb.passive) {
              tcb.status.setValue(STAT_LISTEN);
              //change back the key
              tcbs.remove(tcb.key);
              tcb.key.source_ip=packet.destIP;
              tcb.key.source_port=seg.destport;
              tcb.key.dest_ip=0;
              tcb.key.dest_port=0;
              if(tcbs.get(tcb.key)==null) //extra check, shouldn't be false...
                tcbs.put(tcb.key,tcb);
              provider.statusChanged();
            }
            else {
              tcb.status.setValue(STAT_CLOSED);
              tcbs.remove(tcb.key);
              tcb.user.notify(TCPUser.CON_REFUSED,"connection refused");
              provider.statusChanged();
            }
            return;
          }
          break;
  
        case STAT_ESTABLISHED:
        case STAT_FIN_WAIT1:
        case STAT_FIN_WAIT2:
        case STAT_CLOSE_WAIT:
          if(seg.rst) {
            tcb.retrans_timer.stop(); //safety
            tcb.status.setValue(STAT_CLOSED);
            tcbs.remove(tcb.key);
            tcb.user.notify(TCPUser.CON_RESET,"connection reset");
            provider.statusChanged();
            return;
          }
          break;
  
        case STAT_CLOSING:
        case STAT_LAST_ACK:
        case STAT_TIME_WAIT:
          if(seg.rst) {
            tcb.retrans_timer.stop(); //safety
            tcb.timewait_timer.stop(); //safety
            tcb.status.setValue(STAT_CLOSED);
            tcbs.remove(tcb.key);
            provider.statusChanged();
            return;
          }
          break;
      }

      //third step, check security and precedence (skipped)

      //fourth step, check SYN bit
      if(seg.syn) {
//NOTE: by law should send a reset, but i think it can be omitted ;)
        tcb.retrans_timer.stop(); //safety
        tcb.timewait_timer.stop(); //safety
        tcb.status.setValue(STAT_CLOSED);
        tcbs.remove(tcb.key);
        tcb.user.notify(TCPUser.CON_RESET,"connection reset");
        provider.statusChanged();
        return;
      }

      //fifth step, check ACK field
      if(!seg.ack) continue;
      switch(tcb.status.getValue()) {
        case STAT_SYN_RCVD:
          if(modLE(tcb.snd_una.getValue(),seg.seg_ack) &&
              modLE(seg.seg_ack,tcb.snd_nxt.getValue())) {
            tcb.status.setValue(STAT_ESTABLISHED);

            //must notify user about this!
            Object [] msg=new Object[3];
            msg[0]=new Integer(packet.sourceIP);
            msg[1]=new Integer(seg.sourceport);
            msg[2]=fromlink;
            Object [] params=new Object[3];
            params[0]=tcb;
            params[1]=new Integer(TCPUser.CON_ESTABLISHED);
            params[2]=msg;
            comp.getSim().enqueue(new SimEvent(MY_NOTIFY_USER,this,this,comp.getSim().now(),params));
            provider.statusChanged();
          }
          else {
            //ack not acceptable, send reset
            TCPSegment newseg=new TCPSegment();
            newseg.sourceport=seg.destport;
            newseg.destport=seg.sourceport;
            newseg.seg_seq=seg.seg_ack;
            newseg.rst=true;
            newseg.seg_len=0;
            IPPacket newpacket=new IPPacket();
            newpacket.sourceIP=packet.destIP;
            newpacket.destIP=packet.sourceIP;
            newpacket.protocol=IPPacket.PRO_TCP;
            newpacket.len=46;
            newpacket.payload=newseg;
            send_ip(newpacket);
            continue;
          }

        //flows through intended here.
  
        case STAT_ESTABLISHED:
        case STAT_FIN_WAIT1:
        case STAT_FIN_WAIT2:
        case STAT_CLOSE_WAIT:
        case STAT_CLOSING:
          if(modGT(seg.seg_ack,tcb.snd_nxt.getValue())) {
            //receiver acks something in the future, send an ack back
            TCPSegment newseg=new TCPSegment();
            newseg.sourceport=seg.destport;
            newseg.destport=seg.sourceport;
            newseg.seg_seq=tcb.snd_nxt.getValue();
            newseg.seg_ack=tcb.rcv_nxt.getValue();
            newseg.ack=true;
            newseg.seg_wnd=tcb.rcv_wnd.getValue();
            newseg.seg_len=0;
            IPPacket newpacket=new IPPacket();
            newpacket.sourceIP=packet.destIP;
            newpacket.destIP=packet.sourceIP;
            newpacket.protocol=IPPacket.PRO_TCP;
            newpacket.len=46;
            newpacket.payload=newseg;
            send_ip(newpacket);
            continue;
          }
          if(modLT(tcb.snd_una.getValue(),seg.seg_ack)) {

//Major ACK processing here...

            int acklen=seg.seg_ack-tcb.snd_una.getValue();
            tcb.snd_una.setValue(seg.seg_ack,comp.getSim().now(),provider.getLogFactor());
            if(tcb.rtt_on && modLT(tcb.rtt_seq,tcb.snd_una.getValue())) updateRTT(tcb);

            int userack=acklen; //number of user data to sent_ack

            //should remove acked buffer in retrans_q
            java.util.Iterator i=tcb.retrans_q.iterator();
            while(i.hasNext()) {
              TCPBuffer buf=(TCPBuffer)i.next();
              if(buf.len<=acklen) {
                acklen-=buf.len;
                i.remove();
                if(buf.syn || buf.fin) userack--; //exclude SYN/FIN from user data!
              }
              else {
                buf.len-=acklen;
                break;
              }
            }

            if(userack>0) {
              Object [] params=new Object[3];
              params[0]=tcb;
              params[1]=new Integer(TCPUser.SENT_ACK);
              params[2]=new Integer(userack);
              comp.getSim().enqueue(new SimEvent(MY_NOTIFY_USER,this,this,comp.getSim().now(),params));
            }

            //congestion window management
            if(tcb.fast_retrans_recovery==true) {
              tcb.fast_retrans_recovery=false;
              tcb.dup_ack=0;
              if(provider.getTCPCongestionControl()==2) //Reno
                tcb.cwnd.setValue(tcb.ssthresh.getValue(),comp.getSim().now(),provider.getLogFactor());
              else //Tahoe
                tcb.cwnd.setValue(provider.getMaxSegmentSize(),comp.getSim().now(),provider.getLogFactor());
            }
            else {
              if(tcb.cwnd.getValue()<=tcb.ssthresh.getValue()) {
                //in slow start state
                tcb.cwnd.setValue(tcb.cwnd.getValue()+provider.getMaxSegmentSize(),comp.getSim().now(),provider.getLogFactor());
              }
              else {
                //in congesetion avoidance state
                tcb.cwnd.setValue(tcb.cwnd.getValue()+(int)Math.ceil(
                                  (double)provider.getMaxSegmentSize() *
                                    provider.getMaxSegmentSize() /
                                    tcb.cwnd.getValue()),comp.getSim().now(),provider.getLogFactor());
              }
            }

            //retrans_timer
            tcb.retrans_timer.stop();
            if(!tcb.retrans_q.isEmpty()) tcb.retrans_timer.start();

            //update send window
            if(modLT(tcb.snd_wl1.getValue(),seg.seg_seq) ||
                ((tcb.snd_wl1.getValue()==seg.seg_seq) &&
                  modLE(tcb.snd_wl2.getValue(),seg.seg_ack))) {
              tcb.snd_wnd.setValue(seg.seg_wnd,comp.getSim().now(),provider.getLogFactor());
              tcb.snd_wl1.setValue(seg.seg_seq);
              tcb.snd_wl2.setValue(seg.seg_ack);
            }

            //SH Added: if can send new data then mark to send
            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) &&
                !tcb.input_q.isEmpty()) {
              ack_pending=true;
              if(!tcb.retrans_timer.isRunning()) tcb.retrans_timer.start();
            }
          }
          else if(seg.seg_len==0) {
//Duplicate ack...
            if(provider.getTCPCongestionControl()==1 || //Tahoe or Reno
                provider.getTCPCongestionControl()==2) {
              tcb.dup_ack++;
              if(tcb.fast_retrans_recovery==true) {
                tcb.cwnd.setValue(tcb.cwnd.getValue()+provider.getMaxSegmentSize(),comp.getSim().now(),provider.getLogFactor());
  
                //if can send new data then mark to send
                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) &&
                    !tcb.input_q.isEmpty()) {
                  ack_pending=true;
                  if(!tcb.retrans_timer.isRunning()) tcb.retrans_timer.start();
                }
              }
              else if(tcb.dup_ack==3) {
                tcb.fast_retrans_recovery=true;
                //queue an immediate timeout
                tcb.retrans_timer.stop();
                comp.getSim().enqueue(new SimEvent(MY_RETRANS_TIMEOUT,this,this,comp.getSim().now(),tcb));
              }
            }
//System.out.println(SimClock.Tick2Sec(comp.getSim().now())+" duplicate ack: src="+SimParamIP.IP2String(packet.sourceIP)+
//                    " "+seg.sourceport+" seq="+seg.seg_seq+" ack="+seg.seg_ack);
          }

          //additional processing for these states
          switch(tcb.status.getValue()) {
            case STAT_FIN_WAIT1:
              if(tcb.snd_una.getValue()==tcb.snd_nxt.getValue()) {
                //all are acked => FIN is acked
                Object [] params=new Object[3];
                params[0]=tcb;
                params[1]=new Integer(TCPUser.CON_CLOSED);
                params[2]=null;
                comp.getSim().enqueue(new SimEvent(MY_NOTIFY_USER,this,this,comp.getSim().now(),params));
                tcb.status.setValue(STAT_FIN_WAIT2);
                provider.statusChanged();
              }
              else break;
  
              //flows through intentional
            case STAT_FIN_WAIT2:
              //SH NOTE: RFC793 requires check for retrans_q empty and
              //         acknowledge user's CLOSE command, this is
              //         redundant and will cause multiple acknowledgement