// TCPProtocol.java package protocol; // protocol package import java.util.*; // import Java utility classes import support.*; // import Jasper support classes /** This is the class for a TCP protocol entity. @author Iain A. Robin, Kenneth J. Turner @version 1.0 (1st September 1999, IAR): initial version
1.4 (9th March 2006, KJT): updated for JDK 1.5
1.5 (26th July 2010, KJT): minor tidying; addition of code to handle slow starts; empty PDU removed from receive buffer after reception; state TIME_WAIT now entered after CLOSING or FIN_WAIT2 and a maximum of two received PDUs allowed in this state before closing; FIN is now acknowledged in LISTEN state; a SYN in CLOSING or FIN_WAIT2 state now causes closing */ public class TCPProtocol implements ProtocolEntity, Timeouts { /** Debug flag */ private final static boolean DEBUG = false; // Protocol state constants /** Closed state */ private final static int CLOSED = 0; /** Listen state */ private final static int LISTEN = 1; /** Synchronise received state */ private final static int SYN_RCVD = 2; /** Synchronise sent state */ private final static int SYN_SENT = 3; /** Synchronise pending at closed peer state */ private final static int SYN_PEND = 4; /** Established state */ private final static int ESTABLISHED = 5; /** Finish waiting 1 state */ private final static int FIN_WAIT1 = 6; /** Finish waiting 2 state */ private final static int FIN_WAIT2 = 7; /** Closing state */ private final static int CLOSING = 8; /** Timed waiting state */ private final static int TIME_WAIT = 9; /** Close waiting state */ private final static int CLOSE_WAIT = 10; /** Closed state */ private final static int LAST_ACK = 11; // TCP segment control flag constants /** Urgent flag */ public final static int URG = TCPMessage.URG; /** Urgent flag */ public final static int ACK = TCPMessage.ACK; /** Push flag */ public final static int PSH = TCPMessage.PSH; /** Reset flag */ public final static int RST = TCPMessage.RST; /** Synchronise flag */ public final static int SYN = TCPMessage.SYN; /** Finish flag */ public final static int FIN = TCPMessage.FIN; /** Data flag */ private final static int DATA = 0; // Other protocol constants /** Maximum number of retransmission attempts (currently unused) */ private final static int RETRY_LIMIT = 10; /** Maximum number of PDUs to be received in TIME_WAIT state (the wait ought to be 2 maximum segment lifetimes, but this is not meaningful as the simulation does not use time) */ private final static int WAIT_LIMIT = 2; // Protocol variables /** Open confirmed message */ private final static String OPEN_CONFIRMED = TCPService.OPEN_CONFIRMED; /** Send message */ private final static String SEND = "Send"; /** Deliver message */ private final static String DELIVER = "Deliver"; /** Resend message */ private final static String RESEND = "Timeout - resend"; /** Maximum medium size */ private static int segmentSize; /** Peer protocol entity */ private ProtocolEntity peer; /** Peer protocol entity */ private TCPService user; /** Peer protocol entity */ private Medium medium; /** Peer protocol entity */ private String name; /** Events from entity */ private Vector events; /** Events from user */ private Vector userEvents; /** Segments for timeout */ private Vector timedSegments; /** Incoming segments */ private Vector recvBuffer; /** Segments for sending */ private Vector sendBuffer; /** Peer protocol entity */ private int role; /** Current state */ private int state; /** Previous state */ private int prevState = -1; /** Next send sequence number */ private int sendSeq; /** Acknowledgement sequence number to be sent */ private int sendAck; /** Next expected sequence number */ private int recvSeq; /** Acknowledgement sequence number received */ private int recvAck; /** Local receive window */ private int recvWindow; /** Peer receive window */ private int peerWindow; /** Original peer initial receive window */ private int peerWindowOriginal; /** Original peer initial receive sequence number */ private int peerSequenceOriginal; /** Initial sequence number */ private int initSeq; /** Local window size */ private int window; /** Congestion window size */ private int congestionWindow; /** Slow start threshold */ private int slowStartThreshold; /** Start sequence number */ private int deliverableSeq; /** Size of data */ private int deliverableSize; /** Whether protocol requests have previously been made */ private boolean laterRequest; /** Number of PDUs received in TIME_WAIT state */ private int waitCount; /** Number of octets sent but not acknowledged */ private int sentPending; /** Retry count (currently unused) */ private int retryCount; /** Constructor for a TCP protocol entity. @param medium protocol medium @param name protocol name @param role protocol role @param initSeq initial sequence number @param window initial window */ public TCPProtocol(Medium medium, String name, int role, int initSeq, int window) { this.name = name; this.medium = medium; this.role = role; this.initSeq = initSeq; this.window = window; initialise(); } /** Constructor for a TCP protocol entity (slow start subtype). @param medium protocol medium @param name protocol name @param role protocol role @param initSeq initial sequence number @param window initial window @param peerWindow initial peer window @param peerSequence initial peer sequence number */ public TCPProtocol(Medium medium, String name, int role, int initSeq, int window, int peerWindow, int peerSequence) { this.name = name; this.medium = medium; this.role = role; this.initSeq = initSeq; this.window = window; peerWindowOriginal = peerWindow; peerSequenceOriginal = peerSequence; initialise(); } /** Cancel retransmissions. @param pdu PDU whose retranmission is to be cancelled */ private void cancelRetransmissions(TCPMessage pdu) { if (pdu != null) { int una = oldestUnacked(); // get oldest unack'ed int ack = pdu.ack; // get ack seq no if (una >= 0 && una < ack && ack <= sendSeq) { // ack OK recvAck = ack; // note ack seq no // identify TCP segment(s) being acknowledged for (Enumeration enumeration = timedSegments.elements(); enumeration.hasMoreElements(); ) { TCPMessage tcpdu = (TCPMessage) enumeration.nextElement(); if (tcpdu.seq + tcpdu.size <= ack) { // segment fully acknowledged, remove from retransmission queue timedSegments.removeElement(tcpdu); sentPending -= tcpdu.size; // reduce amount of sent pending enumeration = timedSegments.elements(); } } } } } /** Close the protocol by setting CLOSE state and re-initialising medium. */ public void closeProtocol() { setState(CLOSED); // set state closed } /** Return a protocol event with the given comment. @param comment protocol comment @return protocol event */ public ProtocolEvent comment(String comment) { return(new ProtocolEvent(ProtocolEvent.COMMENT, this, comment)); } /** If this is protocol A, add to global protocol events (events) a comment with the given congestion comment. @return protocol event */ public void commentCongestion() { if (name.equals("Protocol A")) { // protocol A? String windowIncrease = // get window increase type congestionWindow < slowStartThreshold ? "exp." : "lin."; events.addElement( // add congestion comment comment("cwind " + congestionWindow + " (" + windowIncrease + ")")); } } /** Return the protocol name. @return The name value */ public String getName() { return(name); } /** Return protocol entity services. @return protocol entity services */ public Vector getServices() { Vector services = new Vector(); int size = 0; if (!sendBuffer.isEmpty()) { // send buffer non-empty? int firstSize = // get first send element size ((PDU) sendBuffer.firstElement()).size; int protocolWindow = getWindow(); // get window size = Math.min(firstSize, protocolWindow); if (protocolWindow > 0) { // can send to peer? services.addElement(SEND + " " + size + " octets to peer"); } } if (!recvBuffer.isEmpty()) { // receive buffer non-empty? sortBuffer(); // sort into ascending seq. nos. // check first sequence number in buffer is one we expect PDU pdu = (PDU) recvBuffer.firstElement(); int firstSeq = pdu.seq; if (firstSeq <= recvSeq) { // check contiguous data int seq = firstSeq; for (Enumeration enumeration = recvBuffer.elements(); enumeration.hasMoreElements(); ) { pdu = (PDU) enumeration.nextElement(); if (pdu.seq != seq) break; else seq += pdu.size; } size = seq - firstSeq; if (size > 0) // offer to deliver buffered data to user services.addElement(DELIVER + " " + size + " octets to user"); deliverableSeq = firstSeq; deliverableSize = size; } } // check for oldest segment that could have timed out for (Enumeration enumeration = timedSegments.elements(); enumeration.hasMoreElements(); ) { TCPMessage tcpdu = (TCPMessage) enumeration.nextElement(); if (tcpdu.seq == oldestUnacked()) services.addElement(RESEND + " " + tcpdu.getID()); } return(services); } /** Extract integer value from string s with prefix p, removing prefix p (including space character at end). @param s string @param p prefix @return extracted integer */ private int getSize(String s, String p) { String s1 = s.substring(p.length() + 1); return(Integer.parseInt(s1.substring(0, s1.indexOf(' ')))); } /** Return the available window. For client-server or peer-peer, this is the peer window. For slow start, this is the minimum of the peer window and the congestion window (less what has been sent but is outstanding). @return available window */ private int getWindow() { int availableWindow = // get available window Math.min(congestionWindow - sentPending, peerWindow); return(availableWindow); // return window } /** Return whether the protocol uses a timer. All TCP segments are of same type; the need for a timer depends on whether segment contains data. @param type type description @return true/false if protocol uses timer (always true) */ public boolean hasTimer(String type) { return(true); } /** Initialise the protocol. */ public void initialise() { events = new Vector(); // empty events list userEvents = new Vector(); // empty user events list timedSegments = new Vector(); // empty unack'ed segments list sendBuffer = new Vector(); // empty sent segments recvBuffer = new Vector(); // empty received segments recvWindow = window; // init received window size sendSeq = initSeq; // init send ack seq no recvAck = -1; // init received ack seq no waitCount = 0; // initialise waiting PDU count sentPending = 0; // initialise sent pending count retryCount = 0; // initialise retry count if (TCP.isSlowStart()) { // slow start? setState(ESTABLISHED); // start directly as established congestionWindow = segmentSize; // initialise congestion window slowStartThreshold = Integer.MAX_VALUE; // initialise threshold laterRequest = false; // initialise not later request peerWindow = peerWindowOriginal; // initialise peer window recvSeq = peerSequenceOriginal; // initialise peer seq. no. } else { // client-server/peer-peer setState(CLOSED); // start as closed congestionWindow = Integer.MAX_VALUE; // set no congestion window slowStartThreshold = Integer.MAX_VALUE; // set no threshold } } /** Check if a sequence number is a duplicate. @param seq sequence number @return true/false if sequence number is/is not a duplicate */ private boolean isDuplicate(int seq) { // return true if receive buffer already contains // a segment with sequence number seq if (recvBuffer.isEmpty()) return(false); for (Enumeration enumeration = recvBuffer.elements(); enumeration.hasMoreElements(); ) { PDU pdu = (PDU) enumeration.nextElement(); if (pdu.seq == seq) { return(true); } } return(false); } /** Check if the sequence number is within the window. @param seq sequence number @return true/false if sequence number is/is not within window */ private boolean isWithinWindow(int seq) { return(recvSeq <= seq && seq < recvSeq + window); } /** Return oldest unacknowledged sequence number. @return Return Value */ private int oldestUnacked() { if (timedSegments.isEmpty()) // no unacknowledged segments return(-1); int oldestSoFar = ((PDU) timedSegments.firstElement()).seq; for (Enumeration e = timedSegments.elements(); e.hasMoreElements(); ) { PDU pdu = (PDU) e.nextElement(); if (pdu.seq < oldestSoFar) oldestSoFar = pdu.seq; } return(oldestSoFar); } /** Perform given service. @param service service @return resulting protocol events */ public Vector performService(String service) { events.removeAllElements(); // remove previous prot. events userEvents.removeAllElements(); // remove previous user events if (service.startsWith(SEND)) { // send message to peer? // send buffered data to peer int size = getSize(service, SEND); if (sendBuffer.size() > 0) { // at least one element? TCPMessage tcpdu = (TCPMessage) sendBuffer.firstElement(); sendToPeer(tcpdu.flags, size); if (tcpdu.size > size) tcpdu.size -= size; else sendBuffer.removeElement(tcpdu); } else // send buffer is empty System.err.println("cannot send as send buffer is empty"); } else if (service.startsWith(DELIVER)) { // deliver message to user? // deliver buffered data to user int size = deliverableSize; int maxSeq = deliverableSeq + size; int lastElement = 0; // first PDU to stay for (Enumeration enumeration = recvBuffer.elements(); enumeration.hasMoreElements(); ) { PDU pdu = (PDU) enumeration.nextElement(); if (pdu.seq > maxSeq) break; else lastElement++; } while (lastElement-- > 0) recvBuffer.remove(0); recvWindow += size; // alter receive window sendToUser(TCPService.DELIVER + " (" + size + ")"); deliverableSize = 0; sendToPeer(0); // open up window } else if (service.startsWith(RESEND)) { // re-send on timeout? if (TCP.isSlowStart()) { // slow start? slowStartThreshold = // reset slow start theshold Math.max((int) (congestionWindow / 2), segmentSize); events.addElement( // add threshold comment comment("ssthresh " + slowStartThreshold)); congestionWindow = segmentSize; // reset congestion window commentCongestion(); // add congestion window comment } // identify segment to be retransmitted String label = service.substring(RESEND.length() + 1); for (Enumeration enumeration = timedSegments.elements(); enumeration.hasMoreElements(); ) { TCPMessage tcpdu = (TCPMessage) enumeration.nextElement(); if (label.equals(tcpdu.getLabel())) { transmitPDU(tcpdu, peer); timedSegments.removeElement(tcpdu); events.addElement(new ProtocolEvent(ProtocolEvent.TIMEOUT, tcpdu)); break; } } } for (Enumeration enumeration = userEvents.elements(); enumeration.hasMoreElements(); ) events.addElement((ProtocolEvent) enumeration.nextElement()); return(events); } /** Check for data waiting in the send buffer with the push flag set. */ private void pushData() { if (!sendBuffer.isEmpty()) { TCPMessage tcpdu = (TCPMessage) sendBuffer.firstElement(); int protocolWindow = getWindow(); // get window int size = Math.min(tcpdu.size, protocolWindow); if (tcpdu.isPsh()) { sendToPeer(tcpdu.flags, size); if (tcpdu.size > size) tcpdu.size -= size; else sendBuffer.removeElement(tcpdu); } } } /** Check that data segment from peer is within receive window and is not a duplicate of a segment already received. @param pdu PDU */ private void receiveData(PDU pdu) { int seq = pdu.seq; int size = pdu.size; if ((isWithinWindow(seq) || isWithinWindow(seq + size - 1)) && !isDuplicate(seq)) { recvBuffer.addElement(pdu); // store PDU in receive buffer recvWindow -= size; sortBuffer(); // sort buffer int firstSeq = ((PDU) recvBuffer.firstElement()).seq; if (firstSeq <= recvSeq) { deliverableSeq = firstSeq; for (Enumeration enumeration = recvBuffer.elements(); enumeration.hasMoreElements(); ) { pdu = (PDU) enumeration.nextElement(); if (pdu.seq > deliverableSeq) break; else if (pdu.seq == deliverableSeq) deliverableSeq += pdu.size; } deliverableSize = deliverableSeq - firstSeq; } else { deliverableSeq = recvSeq; deliverableSize = 0; } sendAck = deliverableSeq; // update ack seq no recvSeq = deliverableSeq; // update exp. seq no recvWindow = window - deliverableSize; // update receive window sendToPeer(ACK); // respond with ACK if (size == 0) // PDU contains no data? recvBuffer.removeElement(pdu); // remove from receive buffer // must call getServices to ensure a buffer check when undoing getServices(); // check buffers } else { // acknowledge duplicate events.addElement(comment("ignored")); // add ignored comment // sendAck = pdu.seq + pdu.size; sendToPeer(ACK); // respond with ACK } } /** Receive a PDU/SDU @param pdu PDU/SDU @return resulting protocol events */ public Vector receivePDU(PDU pdu) { events.removeAllElements(); // remove previous prot. events userEvents.removeAllElements(); // remove previous user events if (pdu == null) return(events); String pduType = pdu.type; int flags = -1; TCPMessage tcpduRecv = null; if (pdu instanceof TCPMessage) { tcpduRecv = (TCPMessage) pdu; flags = tcpduRecv.flags; peerWindow = tcpduRecv.win; } switch (state) { case CLOSED: // closed? if (pduType.equals(TCPService.ACTIVE_OPEN)) { // step 1 of three-way handshake sendToPeer(SYN); setState(SYN_SENT); } else if (pduType.equals(TCPService.PASSIVE_OPEN)) setState(LISTEN); else if (flags == SYN && role == TCP.PEER) { // closed peer gets SYN? recvSeq = pdu.seq + pdu.size; // note next seq no sendAck = recvSeq; // note seq no for ACK events.addElement(comment("open queued")); // add open queued comment setState(SYN_PEND); // note SYN pending } else if ((flags & RST) != 0) // reset? timedSegments = new Vector(); // cancel retransmissions else // stray segment reset(flags, pdu); // reset connection break; case LISTEN: // listening after passive open? if (flags == SYN) { // step 2 of three-way handshake recvSeq = pdu.seq + pdu.size; sendAck = recvSeq; sendToPeer(SYN + ACK); sendToUser(TCPService.OPEN_RECEIVED); prevState = LISTEN; setState(SYN_RCVD); } else if (flags == FIN) { sendAck = pdu.seq + 1; sendToPeer(ACK); } else if (pduType.equals(TCPService.CLOSE)) closeProtocol(); else if (pduType.equals(TCPService.SEND)) { sendToPeer(SYN); setState(SYN_SENT); } else { events.addElement(comment("ignored")); // add ignored comment reset(flags, pdu); // reset connection } break; case SYN_SENT: // SYN sent? if (flags == SYN + ACK) { // step 3 of three-way handshake cancelRetransmissions(tcpduRecv); recvSeq = pdu.seq + pdu.size; sendAck = recvSeq; sendToPeer(ACK); events.addElement(comment("established")); // add established comment sendToUser(TCPService.OPEN_SUCCESS); setState(ESTABLISHED); } else if (flags == SYN) { // SYN? recvSeq = pdu.seq + pdu.size; // get next seq no sendAck = recvSeq; // set ack seq no sendSeq--; // reset send seq no sendToPeer(SYN + ACK); // SYN and ACK prevState = SYN_SENT; // previously SYN sent setState(SYN_RCVD); // now SYN received } else if ((flags & RST) != 0) { // reset by peer? sendToUser(TCPService.OPEN_FAILURE); // report rejected timedSegments = new Vector(); // cancel retrans closeProtocol(); // now closed } else if ((flags & FIN) != 0) { // finish by peer? sendToUser(TCPService.OPEN_FAILURE); // report rejected timedSegments = new Vector(); // cancel retrans reset(flags, pdu); // reset connection closeProtocol(); // now closed } else if (pduType.equals(TCPService.CLOSE)) // user close? closeProtocol(); // now closed else events.addElement(comment("ignored"));// add ignored comment break; case SYN_RCVD: // SYN received? if (flags == ACK || flags == SYN + ACK) { // cancel timer on previously sent SYN + ACK segment: cancelRetransmissions(tcpduRecv); events.addElement(comment("established")); // add established comment if (prevState == LISTEN) // passive open confirm? transmitPDU(new PDU(OPEN_CONFIRMED), user); if (prevState == SYN_SENT) // was SYN sent? sendToUser(TCPService.OPEN_SUCCESS); if (prevState == SYN_PEND) // was SYN pending? sendToUser(TCPService.OPEN_SUCCESS); setState(ESTABLISHED); } else if ((flags & FIN) != 0) { // other user closed? sendToUser(TCPService.OPEN_FAILURE); // report open failure timedSegments = new Vector(); // cancel retrans recvSeq = pdu.seq + pdu.size; // set next seq expected sendAck = recvSeq; // set ack seq no sendToPeer(FIN + ACK); // acknowledge FIN prevState = SYN_RCVD; // was SYN received setState(LAST_ACK); // waiting for FIN ack } else if (pduType.equals(TCPService.CLOSE)) { sendToPeer(FIN); setState(FIN_WAIT1); // wait for FIN ack } else if (flags == SYN) { // duplicate SYN segment (due to retransmission) timedSegments.removeAllElements(); sendSeq--; // restore seq no peerWindow++; // restore peer window sendToPeer(SYN + ACK); // resend SYN + ACK } else events.addElement(comment("ignored"));// add ignored comment break; case SYN_PEND: // SYN for closed peer? if (pduType.equals(TCPService.ACTIVE_OPEN)) { // active open? sendToPeer(SYN + ACK); // SYN, ACK for prev SYN prevState = SYN_PEND; // was SYN pending state setState(SYN_RCVD); // now SYN rcvd. state } else if ((flags & RST) != 0) // reset? closeProtocol(); // back to closed else // not active open/reset events.addElement(comment("ignored"));// add ignored comment break; case ESTABLISHED: // established? int seq = pdu.seq; // seq. no. first octet int size = pdu.size; // received lengt if (pduType.startsWith(TCPService.SEND)) { // user data request if (TCP.isSlowStart()) { // slow start? if (!laterRequest) { // first request? laterRequest = true; // note now later request commentCongestion(); // add congestion comment } } transmitData(pduType, size); break; } else if (pduType.equals(TCPService.CLOSE)) { // user close sendToPeer(FIN); prevState = ESTABLISHED; setState(FIN_WAIT1); break; } if (tcpduRecv == null) // null PDU received? break; // just in case else if (flags == DATA || // DATA received or (flags == DATA + ACK && size > 0)) { // DATA and ACK received? if ((flags & ACK) != 0) // ACK? updateWindow(); // update congestion window receiveData(pdu); } else if (tcpduRecv.isPsh()) { // push data to user at once if (isWithinWindow(seq) || isWithinWindow(seq + size - 1)) { recvBuffer.addElement(pdu); // buffer to user, ack sortBuffer(); seq = ((PDU) recvBuffer.firstElement()).seq; PDU lastPDU = (PDU) recvBuffer.lastElement(); size = lastPDU.seq + lastPDU.size - seq; sendAck = seq + size; recvSeq += size; // update exp seq no recvBuffer.removeAllElements(); recvWindow = window; sendToPeer(ACK); // send ACK to peer sendToUser(TCPService.DELIVER + " (" + size + ")"); } else { // ack duplicate sendAck = pdu.seq + pdu.size; sendToPeer(ACK); } } else if (flags == SYN + ACK) { // duplicate SYN with ACK? cancelRetransmissions(tcpduRecv); // ack from peer sendAck = pdu.seq + pdu.size; // get next ack seq no sendToPeer(ACK); // acknowledge again } else if (flags == FIN) { // FIN received? sendAck = pdu.seq + 1; sendToPeer(ACK); sendToUser(TCPService.CLOSING); setState(CLOSE_WAIT); } else if ((flags & RST) != 0) { // RST received? if (TCP.isSlowStart()) // slow start? initialise(); // re-initialise protocol else // client-server/peer-peer ; // *** what should happen here? } else if ((flags & ACK) != 0) { // ACK received? updateWindow(); // update congestion window cancelRetransmissions(tcpduRecv); // ack from peer pushData(); // push data to user int oldest = oldestUnacked(); // get oldest unacknowledged if (oldest != -1) { // some data pending? int pending = sendSeq - oldest; // get number of octets pending peerWindow -= pending; // reduce window by pending } } break; case FIN_WAIT1: // FIN sent? if (flags == DATA || (flags == DATA + ACK && pdu.size > 0)) receiveData(pdu); else if (flags == ACK) { cancelRetransmissions(tcpduRecv); setState(FIN_WAIT2); } else if ((flags & SYN) != 0) // duplicate SYN? sendToPeer(ACK); // acknowledge it else if (flags == FIN && // FIN? pdu.seq == recvSeq) { // expected seq no? sendAck = pdu.seq + 1; // get seq no to ack sendToPeer(ACK); // acknowledge FIN setState(CLOSING); // now closing } else if (flags == FIN + ACK) { // FIN and ACK? sendAck = pdu.seq + 1; // get seq no to ack cancelRetransmissions(tcpduRecv); // cancel retrans sendToPeer(ACK); // acknowledge FIN if (prevState == ESTABLISHED) // was established? sendToUser(TCPService.CLOSED); // report closed closeProtocol(); // now closed } else if ((flags & RST) != 0) { // reset? cancelRetransmissions(tcpduRecv); // cancel retrans sendToUser(TCPService.CLOSED); // report closed closeProtocol(); } else events.addElement(comment("ignored"));// add ignored comment break; case FIN_WAIT2: // FIN received? if (flags == DATA) // DATA? receiveData(pdu); else if (flags == ACK) { // ACK? cancelRetransmissions(tcpduRecv); // cancel retrans setState(FIN_WAIT2); } else if (flags == SYN + ACK) // SYN and ACK? // duplicate segment due to retransmission sendToPeer(ACK); else if (flags == FIN) { // FIN? if (sendAck == pdu.seq) { // all segments sent? sendAck = pdu.seq + 1; if (deliverableSize > 0) // data to deliver? sendToUser(TCPService.DELIVER + " (" + deliverableSize + ")"); recvWindow = window; // open receive window recvBuffer.removeAllElements(); // empty receive buffer sendToPeer(ACK); // send ack sendToUser(TCPService.CLOSED); // notify user closed closeProtocol(); // note state closed } else { // outstanding segments sendToPeer(ACK); // send ack setState(TIME_WAIT); // set time wait state // sendToUser(TCPService.CLOSED); // closeProtocol(); // following buffered } } else if ((flags & RST) != 0 || // reset or (flags & SYN) != 0) { // synchronise? cancelRetransmissions(tcpduRecv); // cancel retrans sendToUser(TCPService.CLOSED); // report closed closeProtocol(); } else events.addElement(comment("ignored"));// add ignored comment break; case CLOSING: // closing? if (flags == ACK || // finish ack or (flags & RST) != 0) { // reset? cancelRetransmissions(tcpduRecv); // cancel retrans if (tcpduRecv.ack == sendSeq) { // all segments acknowledged? sendToUser(TCPService.CLOSED); // report closed closeProtocol(); // set closed state } else // enter final wait setState(TIME_WAIT); // set time wait state } else if ((flags & RST) != 0 || // reset or (flags & SYN) != 0) { // synchronise? cancelRetransmissions(tcpduRecv); // cancel retransmission sendToUser(TCPService.CLOSED); // report closed closeProtocol(); } else events.addElement(comment("ignored"));// add ignored comment break; case CLOSE_WAIT: // waiting to close? if (pduType.equals(TCPService.CLOSE)) { sendToPeer(FIN); setState(LAST_ACK); } if (pduType.startsWith(TCPService.SEND)) // user data request transmitData(pduType, pdu.size); if (tcpduRecv != null && tcpduRecv.isAck()) { // acknowledgement from peer cancelRetransmissions(tcpduRecv); // cancel retrans pushData(); } if (flags == DATA || (flags == DATA + ACK && pdu.size > 0)) receiveData(pdu); if (flags == FIN) { sendAck = pdu.seq + 1; sendToPeer(ACK); } break; case LAST_ACK: // last ACK expected? if (flags == -1) // primitive (open)? // sendToUser(TCPService.CLOSED); // report closed // closeProtocol(); ; else if (flags == ACK) { // ACK? cancelRetransmissions(tcpduRecv); // cancel retrans if (timedSegments.size() == 0) { // queue now empty? // prevState != SYN_RCVD) { // was not SYN received? sendToUser(TCPService.CLOSED); // report closed closeProtocol(); // now closed } } else if (flags == FIN) { // FIN? sendAck = pdu.seq + 1; sendToPeer(ACK); } else if (flags == SYN) { // new SYN from peer? sendToUser(TCPService.CLOSED); // report rejected timedSegments = new Vector(); // cancel retrans reset(flags, pdu); // reset connection closeProtocol(); // set closed } else if ((flags & RST) != 0) { // reset? cancelRetransmissions(tcpduRecv); // cancel retrans sendToUser(TCPService.CLOSED); // report closed closeProtocol(); // set closed } else events.addElement(comment("ignored"));// add ignored comment break; case TIME_WAIT: // final timed waiting? waitCount++; // increment wait count if (waitCount >= WAIT_LIMIT) { // wait limit reached? cancelRetransmissions(tcpduRecv); // cancel retrans sendToUser(TCPService.CLOSED); // report closed closeProtocol(); // set closed medium.initialise(); // re-initialise medium } } for (Enumeration enumeration = userEvents.elements(); enumeration.hasMoreElements(); ) events.addElement((ProtocolEvent) enumeration.nextElement()); return(events); } // end of receivePDU /** Handle Reset. @param flags protocol flags @param pdu PDU */ private void reset(int flags, PDU pdu) { // reset connection if ((flags & ACK) != 0) { // ACK flag set? sendSeq = recvSeq; // use incoming seq no sendToPeer(RST); // reset peer } else { // ACK flag unset sendSeq = 0; // default 0 seq no recvSeq = pdu.seq + pdu.size; // note next seq no sendAck = recvSeq; // note seq no for ACK sendToPeer(RST + ACK); // reset peer } } /** Send message to peer with the specified flags. @param flags protocol flags */ private void sendToPeer(int flags) { if (flags == 0) { // pure window change TCPMessage tcpdu = new TCPMessage(sendSeq, sendAck, 0, 0); tcpdu.setWindowSize(recvWindow); transmitPDU(tcpdu, peer); events.addElement(new ProtocolEvent(ProtocolEvent.TRANSMIT, tcpdu)); } else if (flags == ACK) { // pure ACK - no seq. TCPMessage tcpdu = new TCPMessage(sendSeq, sendAck, ACK, 0); tcpdu.setWindowSize(recvWindow); events.addElement(new ProtocolEvent(ProtocolEvent.TRANSMIT, tcpdu)); transmitPDU(tcpdu, peer); } else sendToPeer(flags, 1); // other - seq. 1 } /** Send message to peer with the specified flags and packet size. @param flags protocol flags @param packetSize packet size */ private void sendToPeer(int flags, int packetSize) { // send packet of given size to peer entity with specified flags set // if packet size too large for medium to handle, fragment it for (int p = 0; p < packetSize; p += segmentSize) { int size = Math.min(packetSize - p, segmentSize); TCPMessage tcpdu = new TCPMessage(sendSeq, sendAck, flags, size); if (tcpdu.isAck()) flags -= ACK; // ACK only in first tcpdu.setWindowSize(recvWindow); events.addElement(new ProtocolEvent(ProtocolEvent.TRANSMIT, tcpdu)); transmitPDU(tcpdu, peer); sendSeq += size; peerWindow -= size; sentPending += size; // increase pending sent data } } /** Send message to user. @param type SDU type */ private void sendToUser(String type) { PDU pdu = new PDU(type); transmitPDU(pdu, user); events.addElement(new ProtocolEvent(ProtocolEvent.DELIVER, pdu)); } /** Set protocol peer. @param peer protocol peer */ public void setPeer(ProtocolEntity peer) { this.peer = peer; } /** Set segment size (maximum send packet size). @param size maximum send packet size */ public static void setSegmentSize(int size) { segmentSize = size; } /** Set the protocol state. @param state protocol state */ public void setState(int state) { if (DEBUG) System.err.println("state (" + name + "): " + state); this.state = state; } /** Sets the timer for a PDU. @param pdu PDU @param enabled whether timer is enabled */ public void setTimer(PDU pdu, boolean enabled) { if (enabled && pdu.size > 0 && // enabled, non-empty? pdu.seq >= recvAck && // new segment? !((TCPMessage) pdu).isRst()) // no reset flag? timedSegments.addElement(pdu); } /** Set the protocol user. @param user protocol user (service entity) */ public void setUser(ProtocolEntity user) { this.user = (TCPService) user; } /** Set the protocol window size. @param winSize window size */ public void setWindowSize(int winSize) { recvWindow = winSize; } /** Set the protocol default window size. @param winSize default window size */ public void setWindowSizeDefault(int winSize) { window = winSize; recvWindow = winSize; } /** Description of the Method */ private void sortBuffer() { PDU pdu1; PDU pdu2; if (recvBuffer.isEmpty()) return; int size = recvBuffer.size(); // sort buffer in order if (size < 2) return; // no need to sort for (int i = 0; i < size; i++) for (int j = i + 1; j < size; j++) { pdu1 = (PDU) recvBuffer.elementAt(i); pdu2 = (PDU) recvBuffer.elementAt(j); if (pdu2.seq < pdu1.seq) { recvBuffer.setElementAt(pdu2, i); recvBuffer.setElementAt(pdu1, j); } } } /** Transmit protocol data. @param pduType PDU type @param size PDU size */ private void transmitData(String pduType, int size) { // check whether PUSH flag is to be set int sendFlags = pduType.indexOf(TCPService.PUSH) > 0 ? PSH : DATA; int protocolWindow = getWindow(); // get window if (size <= protocolWindow) { // can send all the data? if (deliverableSize > 0) { // piggyback ACK on data sendAck = deliverableSeq + deliverableSize; sendFlags += ACK; } sendToPeer(sendFlags, size); } else { // can send only some // buffer data to be sent when peer's receive buffer has space if (protocolWindow > 0) { size -= protocolWindow; // send as much as possible sendToPeer(sendFlags, protocolWindow); } // ... and buffer the rest TCPMessage tcpdu = new TCPMessage(sendSeq, sendAck, sendFlags, size); tcpdu.setWindowSize(recvWindow); sendBuffer.addElement(tcpdu); events.addElement(comment("buffered")); // add buffered comment } } /** Transmit PDU. @param pdu PDU @param destination destination */ public void transmitPDU(PDU pdu, ProtocolEntity destination) { pdu.setSource(this); // set PDU source pdu.setDestination(destination); // set PDU destination if (destination == peer) { // sending to peer? userEvents = medium.receivePDU(pdu); } else // sending to service userEvents = user.receivePDU(pdu); } /** Update congestion window on ACK if window management subtype. */ public void updateWindow() { if (TCP.isSlowStart()) { // slow start? congestionWindow += segmentSize; // augment congestion window commentCongestion(); // add congestion window comment } } }