UDT协议-基于UDP的可靠数据传输协议的实现分析(4)-发送和接收的算法

0. 计时器

udt有四种计时器：

ACK, NAK, EXP and SND

1. 发送端的算法

数据结构和变量：
1）SenderLossList： 记录发送方丢失的包的列表，根据序号升序排列
2）sendBuffer： 记录发送过的包和序号

发送算法：

1）如果丢失列表不为空，则重传这些packet包，并从丢失列表中移出，到5）
2）若应用层有数据，则执行发送
3) 进行检查
   a. 若未确认的包的数量超过流量窗口的大小，则回到1）
   b. 组织一个新的数据包，之后发送出去
4） 若当前的包的序号是16n（16的倍数），回到2），此处的作用是pair packet，可用于估计带宽的
5) 记录包的发送时间到SND PKT History Window中
6）如果属于发送速率降低后发送的第一个包，则等待SYN的时间
7) 得到STP-t的时间， t是step 1 to step 4 花费的时间. 回到 1).

算法的核心：
（1）为了保证可靠性，没有发送成功的包，需要重传
（2）可以发送的包数量受拥塞窗口和滑动窗口的限制
（3）根据发送速率的控制，两个包之间的发送时间间隔需要控制

udt-java版本实现的算法，基本一致
   

/**
     * sender algorithm
     */
    long iterationStart;
    public void senderAlgorithm()throws InterruptedException, IOException{
        while(!paused){
            // 本次迭代的时间
            iterationStart=Util.getCurrentTime();
            // 步骤1)：if the sender's loss list is not empty
            // 移出
            Long entry=senderLossList.getFirstEntry();
            if(entry!=null){
                // 执行重传
                handleRetransmit(entry);
            }
            else
            {
                // 步骤2) 若应用层有数据，则执行发送
                // 步骤3) 进行检查
                //if the number of unacknowledged data packets does not exceed the congestion
                //and the flow window sizes, pack a new packet
                int unAcknowledged=unacknowledged.get();
               
                // 未确认的包的数量小于拥塞窗口大小和流控制窗口
                if(unAcknowledged<session.getCongestionControl().getCongestionWindowSize()
                        && unAcknowledged<session.getFlowWindowSize()){
                    //check for application data
                    DataPacket dp=flowWindow.consumeData();
                    if(dp!=null){
                        send(dp);
                        largestSentSequenceNumber=dp.getPacketSequenceNumber();
                    }
                    else{
                        statistics.incNumberOfMissingDataEvents();
                    }
                }else{
                    //congestion window full, wait for an ack
                    if(unAcknowledged>=session.getCongestionControl().getCongestionWindowSize()){
                        statistics.incNumberOfCCWindowExceededEvents();
                    }
                   
                    // udt-java自己加入的处理方式
                    waitForAck();
                }
            }

            //wait
            if(largestSentSequenceNumber % 16 !=0){
                // 步骤7)
                long snd=(long)session.getCongestionControl().getSendInterval();
                long passed=Util.getCurrentTime()-iterationStart;
                int x=0;
                while(snd-passed>0){
                    //can't wait with microsecond precision :(
                    if(x==0){
                        statistics.incNumberOfCCSlowDownEvents();
                        x++;
                    }
                    passed=Util.getCurrentTime()-iterationStart;
                    if(stopped)return;
                }
            }
        }
    }

 
2. 接收端的算法

数据结构和变量
1）ReceiverLossListEntry和ReceiverLossList: ReceiverLossListEntry记录了丢失的packet的序号和反馈的时间，
ReceiverLossList是ReceiverLossListEntry的列表，根据序号升序排列
2）AckHistoryEntry和AckHistoryWindow：AckHistoryEntry记录发送的确认序列号和发送时间信息，AckHistoryWindow
是AckHistoryEntry的循环队列
3）PacketHistoryWindow：记录到达的package的信息
4）PacketPairWindow：记录两个probing packet pair之间的时间间隔
5) largestReceivedSeqNumber（LRSN）: 接收到的最大的确认的序号
6) exp-count: A variable to record number of continuous EXP timeout
   events and its initial value is 1.

接收算法：
1）检查ACK, NAK, or EXP timer是否到期可以处理了
2) 执行带时间限制的poll操作接收包，若没有到达的包，回到步骤1）
3) 设置exp-count=1,更新ETP: ETP = RTT + 4 * RTTVar + ATP.
4) 当所有的包都确认后，重置EXP的过期时间
5) 检查是否是控制包，是的话，处理完之后，回到1）
6) 若当前接收到的包序号是16n + 1, 在Packet Pair Window中保存此包和上一个包之间的时间间隔
7) 记录到达包到PKT History Window中.
8) a. 若当前序号大于上次LRSN+1，则将LRSN到本序号之间的包放到接收丢失列表中，并发送NAK包
   b. 若当前序号小于LRSN，则从接receiver's loss list中移出
9) 更新LRSN. 回到1).

udt-java的实现：

 */
    public void receiverAlgorithm()throws InterruptedException,IOException{
        // 1) 步骤1，检查各种定时器
        //check ACK timer
        long currentTime=Util.getCurrentTime();
        if(nextACK<currentTime){
            nextACK=currentTime+ackTimerInterval;
            processACKEvent(true);
        }
        //check NAK timer
        if(nextNAK<currentTime){
            nextNAK=currentTime+nakTimerInterval;
            processNAKEvent();
        }

        //check EXP timer
        if(nextEXP<currentTime){
            nextEXP=currentTime+expTimerInterval;
            processEXPEvent();
        }
        // 步骤2）perform time-bounded UDP receive
        UDTPacket packet=handoffQueue.poll(Util.getSYNTime(), TimeUnit.MICROSECONDS);
        if(packet!=null){
            // 有数据，步骤3）
            //reset exp count to 1
            expCount=1;
            // 步骤4），实现有偏差
            //If there is no unacknowledged data packet, or if this is an
            //ACK or NAK control packet, reset the EXP timer.
            boolean needEXPReset=false;
            if(packet.isControlPacket()){
                ControlPacket cp=(ControlPacket)packet;
                int cpType=cp.getControlPacketType();
                if(cpType==ControlPacketType.ACK.ordinal() || cpType==ControlPacketType.NAK.ordinal()){
                    needEXPReset=true;
                }
            }
           
            if(needEXPReset){
                nextEXP=Util.getCurrentTime()+expTimerInterval;
            }
            if(storeStatistics)processTime.begin();
            // 处理packet
            processUDTPacket(packet);
           
            if(storeStatistics)processTime.end();
        }
       
        Thread.yield();
    }

    // 处理数据包
    protected void onDataPacketReceived(DataPacket dp)throws IOException{
        long currentSequenceNumber = dp.getPacketSequenceNumber();
       
        //for TESTING : check whether to drop this packet
//        n++;
//        //if(dropRate>0 && n % dropRate == 0){
//            if(n % 1111 == 0){   
//                logger.info("**** TESTING:::: DROPPING PACKET "+currentSequenceNumber+" FOR TESTING");
//                return;
//            }
//        //}
        // 将数据放到接收缓存中
        boolean OK=session.getSocket().getInputStream().haveNewData(currentSequenceNumber,dp.getData());
        if(!OK){
            //need to drop packet...
            return;
        }
       
        long currentDataPacketArrivalTime = Util.getCurrentTime();

        /* 步骤6）(4).if the seqNo of the current data packet is 16n+1,record the
        time interval between this packet and the last data packet
        in the packet pair window*/
        if((currentSequenceNumber%16)==1 && lastDataPacketArrivalTime>0){
            long interval=currentDataPacketArrivalTime -lastDataPacketArrivalTime;
            packetPairWindow.add(interval);
        }
       
        // 步骤7）(5).record the packet arrival time in the PKT History Window.
        packetHistoryWindow.add(currentDataPacketArrivalTime);

       
        //store current time
        lastDataPacketArrivalTime=currentDataPacketArrivalTime;

       
        // 步骤8）
        //(6).number of detected lossed packet
        /*(6.a).if the number of the current data packet is greater than LSRN+1,
            put all the sequence numbers between (but excluding) these two values
            into the receiver's loss list and send them to the sender in an NAK packet*/
        if(SequenceNumber.compare(currentSequenceNumber,largestReceivedSeqNumber+1)>0){
            sendNAK(currentSequenceNumber);
        }
        else if(SequenceNumber.compare(currentSequenceNumber,largestReceivedSeqNumber)<0){
                /*(6.b).if the sequence number is less than LRSN,remove it from
                 * the receiver's loss list
                 */
                receiverLossList.remove(currentSequenceNumber);
        }

        statistics.incNumberOfReceivedDataPackets();
       
        // 步骤9），更新LRSN
        //(7).Update the LRSN
        if(SequenceNumber.compare(currentSequenceNumber,largestReceivedSeqNumber)>0){
            largestReceivedSeqNumber=currentSequenceNumber;
        }

        //(8) need to send an ACK? Some cc algorithms use this
        if(ackInterval>0){
            if(n % ackInterval == 0)processACKEvent(false);
        }
    }

 

ACK Event处理:
1）确定ACK number： 如果receiver's loss list为空，ACK number为LRSN + 1；否则
ACK number为receiver's loss list中的最小值
2) ACK number大于ACK2，或者ACK number等于上次的ACK number并且时间间隔小于2RTT，则结束这个ACK的发送
3) 设置ACK sequence number
4) 计算包到达的速率，算法如下：计算PKT History Window中最后16个到达间隔的均值（AI），
移除间隔大于AI*8或者小于AI/8的均值，若剩下的数量大于8， 重新计算AI'，the packet arrival speed is
1/AI' (number of packets per second); 否则返回0.
5） 计算滑动窗口effective flow window size as: max(min(W, available receiver buffer size), 2)
6) 计算estimated link capacity，算法如下：如果是quick start phase阶段，返回0，否则计算Packet Pair Window
中的最后16个packet pair的到达均值（PI），link capacity is 1/PI (number of packets per second).
7) 组装ACK包并发送：Pack the ACK sequence number, ACK number, RTT, RTT variance,
   effective flow window size, and estimated link capacity into the
   ACK packet and send it out.
8) 记录发送的ACK信息到ACK History Window中

          

     protected void processACKEvent(boolean isTriggeredByTimer)throws IOException{
        //步骤1）(1).Find the sequence number *prior to which* all the packets have been received
        final long ackNumber;
        ReceiverLossListEntry entry=receiverLossList.getFirstEntry();
        if (entry==null) {
            ackNumber = largestReceivedSeqNumber + 1;
        } else {
            ackNumber = entry.getSequenceNumber();
        }
        // 步骤2）(2).a) if ackNumber equals to the largest sequence number ever acknowledged by ACK2
        if (ackNumber == largestAcknowledgedAckNumber){
            //do not send this ACK
            return;
        }else if (ackNumber==lastAckNumber) {
            //or it is equals to the ackNumber in the last ACK 
            //and the time interval between these two ACK packets
            //is less than 2 RTTs,do not send(stop)
            long timeOfLastSentAck=ackHistoryWindow.getTime(lastAckNumber);
            if(Util.getCurrentTime()-timeOfLastSentAck< 2*roundTripTime){
                return;
            }
        }
        final long ackSeqNumber;
        //if this ACK is not triggered by ACK timers,send out a light Ack and stop.
        if(!isTriggeredByTimer){
            ackSeqNumber=sendLightAcknowledgment(ackNumber);
            return;
        }
        else{
            // 步骤7）
            //pack the packet speed and link capacity into the ACK packet and send it out.
            //(7).records  the ACK number,ackseqNumber and the departure time of
            //this Ack in the ACK History Window
            ackSeqNumber=sendAcknowledgment(ackNumber);
        }
       
        // 步骤8）
        AckHistoryEntry sentAckNumber= new AckHistoryEntry(ackSeqNumber,ackNumber,Util.getCurrentTime());
        ackHistoryWindow.add(sentAckNumber);
        //store ack number for next iteration
        lastAckNumber=ackNumber;
    }

 
NAK Event处理:

找到feedback time大于k*RTT的包，发送丢失信息
   

protected void processNAKEvent()throws IOException{
        //find out all sequence numbers whose last feedback time larger than is k*RTT
        List<Long>seqNumbers=receiverLossList.getFilteredSequenceNumbers(roundTripTime,true);
        sendNAK(seqNumbers);
    }

 

EXP Event处理:
1）将所有未确认的包放到发送丢失列表中
2）如果ExpCount > 16 并且具体距离上次操作的时间超过3s 或者 3min已过，则关闭UDT并退出
3) 如果发送丢失列表为空，则发送keep-alive packet包
4) Increase ExpCount by 1.

  

  protected void processEXPEvent()throws IOException{
        if(session.getSocket()==null || !session.getSocket().isActive())return;
        UDTSender sender=session.getSocket().getSender();
        //put all the unacknowledged packets in the senders loss list
        sender.putUnacknowledgedPacketsIntoLossList();
        if(expCount>16 && System.currentTimeMillis()-sessionUpSince > IDLE_TIMEOUT){
            if(!connectionExpiryDisabled &&!stopped){
                sendShutdown();
                stop();
                logger.info("Session "+session+" expired.");
                return;
            }
        }
        if(!sender.haveLostPackets()){
            sendKeepAlive();
        }

 

On ACK packet处理:

1) Update the largest acknowledged sequence number.
2) Send back an ACK2 with the same ACK sequence number in this ACK.
3) Update RTT and RTTVar.
4) Update both ACK and NAK period to 4 * RTT + RTTVar + SYN.
5) Update flow window size.
6) If this is a Light ACK, stop.
7) Update packet arrival rate: A = (A * 7 + a) / 8, where a is the
   value carried in the ACK.
8) Update estimated link capacity: B = (B * 7 + b) / 8, where b is
   the value carried in the ACK.
9) Update sender's buffer (by releasing the buffer that has been
   acknowledged).
10) Update sender's loss list (by removing all those that has been
      acknowledged).

   

protected void onAcknowledge(Acknowledgement acknowledgement)throws IOException{
        ackLock.lock();
        ackCondition.signal();
        ackLock.unlock();

        CongestionControl cc=session.getCongestionControl();
        long rtt=acknowledgement.getRoundTripTime();
        if(rtt>0){
            long rttVar=acknowledgement.getRoundTripTimeVar();
            // 步骤3） 更新rtt和rttVar
            cc.setRTT(rtt,rttVar);
            statistics.setRTT(rtt, rttVar);
        }
        // 步骤7），步骤8）， 更新包的速率和估计带宽
        long rate=acknowledgement.getPacketReceiveRate();
        if(rate>0){
            long linkCapacity=acknowledgement.getEstimatedLinkCapacity();
            cc.updatePacketArrivalRate(rate, linkCapacity);
            statistics.setPacketArrivalRate(cc.getPacketArrivalRate(), cc.getEstimatedLinkCapacity());
        }

        long ackNumber=acknowledgement.getAckNumber();
        cc.onACK(ackNumber);
        statistics.setCongestionWindowSize((long)cc.getCongestionWindowSize());
        // 步骤9），10）
        //need to remove all sequence numbers up the ack number from the sendBuffer
        boolean removed=false;
        for(long s=lastAckSequenceNumber;s<ackNumber;s++){
            synchronized (sendLock) {
                // 发送缓存里移除
                removed=sendBuffer.remove(s)!=null;
                // 发送丢失列表里移除
                senderLossList.remove(s);
            }
            if(removed){
                unacknowledged.decrementAndGet();
            }
        }
        // 步骤1）
        lastAckSequenceNumber=Math.max(lastAckSequenceNumber, ackNumber);   
        // 步骤2）
        //send ACK2 packet to the receiver
        sendAck2(ackNumber);
        statistics.incNumberOfACKReceived();
        if(storeStatistics)statistics.storeParameters();
    }

 

On NAK packet处理:
1) Add all sequence numbers carried in the NAK into the sender's loss
      list.
2) Update the SND period by rate control
3) Reset the EXP time variable.

On ACK2 packet处理:
1) Locate the related ACK in the ACK History Window according to the
   ACK sequence number in this ACK2.
2) Update the largest ACK number ever been acknowledged.
3) Calculate new rtt according to the ACK2 arrival time and the ACK
   departure time, and update the RTT value as: RTT = (RTT * 7 +
      rtt) / 8.
4) Update RTTVar by: RTTVar = (RTTVar * 3 + abs(RTT - rtt)) / 4.
5) Update both ACK and NAK period to 4 * RTT + RTTVar + SYN.

   

 protected void onAck2PacketReceived(Acknowledgment2 ack2){
        AckHistoryEntry entry=ackHistoryWindow.getEntry(ack2.getAckSequenceNumber());
        if(entry!=null){
            long ackNumber=entry.getAckNumber();
            largestAcknowledgedAckNumber=Math.max(ackNumber, largestAcknowledgedAckNumber);
           
            long rtt=entry.getAge();
            if(roundTripTime>0)roundTripTime = (roundTripTime*7 + rtt)/8;
            else roundTripTime = rtt;
            roundTripTimeVar = (roundTripTimeVar* 3 + Math.abs(roundTripTimeVar- rtt)) / 4;
            ackTimerInterval=4*roundTripTime+roundTripTimeVar+Util.getSYNTime();
            nakTimerInterval=ackTimerInterval;
            statistics.setRTT(roundTripTime, roundTripTimeVar);
        }
    }

 

On message drop request received处理:
   1) Tag all packets belong to the message in the receiver buffer so
      that they will not be read.
   2) Remove all corresponding packets in the receiver's loss list.

On Keep-alive packet received处理:
   Do nothing.

On Handshake/Shutdown packet received处理：后续分析