Application Level Congestion Control and Reliable Delivery

[Specifications]

Client

• Loss detection via timeout and triple duplicate acks
• Retransmission for lost packets
• Slow start
• Loss based congestion control
• (Extra credit for delay based congestion control)

from socket import *
from threading import Thread # 보내는 thread 따로, 받는 thread 따로
import random
import time

serverIP = '127.0.0.1' # special IP for local host
serverPort = 12000
clientPort = 12001

win = 1      # window size -> window size만큼 동시에 전송
no_pkt = 1000 # the total number of packets to send -> 보내고자하는 packet의 전체 개수
send_base = 0 # oldest packet sent/ window에서 가장 처음 pkt
loss_rate = 0.001 # loss rate
seq = 0        # initial sequence number

timeout_flag = 0 # timeout trigger
triple_dup_flag = 0 # 3 dup trigger
same_ack_count = [0 for i in range(no_pkt)] # 가 2 이상이면 retransmission
sent_time = [0 for i in range(2000)] #packet 보낸 시간을 기록

clientSocket = socket(AF_INET, SOCK_DGRAM)
clientSocket.bind(('', clientPort))
clientSocket.setblocking(0)

# thread for receiving and handling acks
def handling_ack(): # ack 받는 thread
    print("thread")
    global clientSocket
    global send_base

    global timeout_flag
    global triple_dup_flag
    global same_ack_count

    global sent_time

    alpha = 0.125
    beta = 0.25
    timeout_interval = 10  # timeout interval

    
    pkt_delay = 0
    dev_rtt = 0
    init_rtt_flag = 1

    while True:
       
        if sent_time[send_base] != 0: # sent_time이 0이 아니면 보낸거
            pkt_delay = time.time() - sent_time[send_base]

        if 2 in same_ack_count and triple_dup_flag == 0: # triple duplicated ack detected
            print("triple duplicated ack detected:", str(send_base), flush=True)
            triple_dup_flag = 1

        elif pkt_delay > timeout_interval and timeout_flag == 0:    # timeout detected
            print("timeout detected:", str(send_base), flush=True)
            print("timeout interval:", str(timeout_interval), flush=True)
            timeout_flag = 1

        try:
            ack, serverAddress = clientSocket.recvfrom(2048)
            ack_n = int(ack.decode())
            same_ack_count[ack_n] += 1
            print(ack_n, flush=True) # 받은 ack 출력
            
            if init_rtt_flag == 1:
                estimated_rtt = pkt_delay
                init_rtt_flag = 0
            else:
                estimated_rtt = (1-alpha)*estimated_rtt + alpha*pkt_delay
                dev_rtt = (1-beta)*dev_rtt + beta*abs(pkt_delay-estimated_rtt)
            timeout_interval = estimated_rtt + 4*dev_rtt
            #print("timeout interval:", str(timeout_interval), flush=True)

            
        except BlockingIOError:
            continue

            
        # window is moved upon receiving a new ack
        # window stays for cumulative ack
        send_base = ack_n + 1
        
        if ack_n == no_pkt - 1:
            break;

# running a thread for receiving and handling acks
th_handling_ack = Thread(target = handling_ack, args = ())
th_handling_ack.start()

########################
start_time = time.time() * 1000
########################
# 보내는 thread = main thread
while seq < no_pkt: 
    while seq < send_base + win: # send packets within window/ window size만큼 동시에 pkt을 보냄
        if random.random() < 1 - loss_rate: # emulate packet loss -> 99%의 확률로 packet을 보냄 & 1%의 확률로 if문을 건너 뛰고 packet을 보내지 않음
            clientSocket.sendto(str(seq).encode(), (serverIP, serverPort))  
        sent_time[seq] = time.time()    
        seq = seq + 1

    if timeout_flag == 1 or triple_dup_flag == 1: # retransmission
        seq = send_base 
        clientSocket.sendto(str(seq).encode(), (serverIP, serverPort))
        sent_time[seq] = time.time()
        print("retransmission:", str(seq), flush=True)
        same_ack_count = [0 for i in range(no_pkt)]
        seq = seq + 1
        timeout_flag = 0
        triple_dup_flag = 0
        win = 1
    else: # pkt loss 안난 경우
        win = win * 2

        
th_handling_ack.join() # terminating thread
######################
end_time = time.time() * 1000
######################
print ("done")
print(f"{end_time - start_time} msec")

clientSocket.close()

Server

• Cumulative Ack
• Emulating queue behavior (packet loss and delay)

from socket import *
from threading import Thread

serverPort = 12000

serverSocket = socket(AF_INET, SOCK_DGRAM)
serverSocket.bind(('', serverPort))

print('The server is ready to receive')

no_pkt = 1000
rcv_base = 0  # next sequence number we wait for

server_queue = []
server_queue_size = 100

#def server_queue():
#    global



#th_queue = Thread(target = server_queue, args = ())
#th_queue.start()

while True: # main thread
    message, clientAddress = serverSocket.recvfrom(2048)
    seq_n = int(message.decode()) # extract sequence number
    print(seq_n)
    if seq_n == rcv_base: # in order delivery
        rcv_base = seq_n + 1 
    serverSocket.sendto(str(rcv_base-1).encode(), clientAddress) # send cumulative ack
    if seq_n == no_pkt - 1:
        break;

#th_queue.join() # terminating thread

serverSocket.close()

[Performance evaluation: packet loss rate를 변경하면서, latency와 throughput을 측정]

client.py 파일에서 time module을 이용해 start time과 end time을 측정하였습니다.

Loss rate: 0.001
-> 총 소요 시간: 61.612060546875 msec
-> latency: 0.0616 msec
-> throughput: 16233.77

Loss rate: 0.01
-> 총 소요 시간: 90.693359375 msec
-> latency: 0.0906 msec
-> throughput: 11037.52

loss rate: 0.1
-> 총 소요 시간: 168.521728515625 msec
-> latency: 0.1685 msec
-> throughput: 5934.71

loss rate 가 증가함에 따라 latency가 길어지고, throughput이 적어진다는 사실을 알 수 있습니다.