模擬python中慢速網絡的簡單方法

Question

場景。我有一個客戶端與服務器的兩個網絡連接。一個連接使用手機，另一個連接使用wlan連接。

，我已經解決了這一問題的方法是通過讓服務器在兩個端口上偵聽。但是，移動連接應該比wlan連接慢。發送的數據通常只是一行文本。我通過允許移動連接以5秒的間隔發送數據來解決較慢的連接問題，因此wlan連接的間隔爲一秒。

但是有沒有更好的方法來做減速？也許通過發送更小的數據包，意味着我需要發送更多的數據？

任何想法都可以減慢其中一個連接的速度嗎？

Orjanp

一個簡單的客戶端示例只有一個連接。

def client(): 
    import sys, time, socket 

    port = 11111 
    host = '127.0.0.1' 
    buf_size = 1024 

    try: 
     mySocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
     mySocket.connect((host, port)) 
    except socket.error, (value, message): 
     if mySocket: 
      mySocket.close() 
     print 'Could not open socket: ' + message 
     sys.exit(1) 
    mySocket.send('Hello, server') 
    data = mySocket.recv(buf_size) 
    print data 
    time.sleep(5) 

    mySocket.close() 

client() 

一個簡單的服務器監聽一個端口。

def server(): 
    import sys, os, socket 

    port = 11111 
    host = '' 
    backlog = 5 
    buf_size = 1024 

    try: 
     listening_socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM) 
     listening_socket.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) 
     listening_socket.bind((host, port)) 
     listening_socket.listen(backlog) 
    except socket.error, (value, message): 
     if listening_socket: 
      listening_socket.close() 
     print 'Could not open socket: ' + message 
     sys.exit(1) 

    while True: 
     accepted_socket, adress = listening_socket.accept() 
     data = accepted_socket.recv(buf_size) 
     if data: 
      accepted_socket.send('Hello, and goodbye.') 
     accepted_socket.close() 

server() 

Answer 1

除了使用外部工具來模擬一種網絡你感興趣的，一個好的辦法是使用插座的替代實現。

這涉及到使套接字構造成爲您的函數的一個參數，而不是導入套接字模塊並直接使用它。對於正常操作，您將傳入真實的套接字類型，但是當您要測試各種不利的網絡條件時，可以傳入模擬這些條件的實現。例如，您可以創建其參數化延遲和帶寬的插座類型（未經測試的代碼，謹防）：

import time, socket 

class ControllableSocket: 
    def __init__(self, latency, bandwidth): 
     self._latency = latency 
     self._bandwidth = bandwidth 
     self._bytesSent = 0 
     self._timeCreated = time.time() 
     self._socket = socket.socket() 

    def send(self, bytes): 
     now = time.time() 
     connectionDuration = now - self._timeCreated 
     self._bytesSent += len(bytes) 
     # How long should it have taken to send how many bytes we've sent with our 
     # given bandwidth limitation? 
     requiredDuration = self._bytesSent/self._bandwidth 
     time.sleep(max(requiredDuration - connectionDuration, self._latency)) 
     return self._socket.send(bytes) 

如果要實現其他插座方法，連接的recv等，可以替代這個類的一個實例爲真實套接字類型的一個實例。這使得程序的其餘部分完全沒有任何有關仿真的知識，簡化了它，同時還讓您通過實現模擬它們的新套接字類型來嘗試許多不同的網絡配置。

這個想法是Twisted明確地將「協議」的概念 - 知道如何從網絡中解釋字節並生成新字節發送到網絡的對象 - 從「傳輸」 - 知道如何從網絡中獲取字節並將字節放到網絡上。這種分離方式簡化了測試過程，並且允許像這樣的新穎配置，其中一些其他網絡條件（實際上可能難以產生）的模擬由運輸工具提供。

Answer 2

那麼，什麼使網絡連接比其他「慢」，是由於延遲和/或帶寬。因此，如果您想要進行逼真的模擬，您需要找到手機連接的帶寬以及延遲，並在您的客戶端程序中進行模擬。

但你似乎在暗示你發送的數據非常少，因此帶寬可能不是真的會影響到你的連接速度。所以你可以模擬延遲，這就是你要做的事情：發送每個數據包之間的睡眠（延遲）。雖然5秒似乎很多。

但是，如果你認爲帶寬可能是相關的，它實際上很簡單的模擬：帶寬是你可以發送每秒的最大字節數，延遲是它到達目的地所需的持續時間。

如何做到這一點：有一個全局時間戳「blockedUntil」，表示時間，直到你的連接變得免費再次發送數據。在程序開始時初始化爲0，因爲我們假設它尚未使用。然後，每次你有一個包發送，如果「_blockedUntil」小於now（），將其設置爲now（）。然後計算寫入虛擬「wire」所需的時間：packet.size（）/ bandwidth，這會給你一個持續時間，增加延遲時間，並將其添加到「blockedUntil」。

現在計算DT = blockedUntil - 現在（），該數據包添加到隊列中，並在「DT」添加一個計時器射擊，這將彈出在隊列中的第一個數據包，並將其發送。

你走了，你已經模擬了帶寬和延遲。

編輯：有人提到有丟棄的數據包太的問題。您可以通過丟棄數據包的可能性來模擬該數據包。 注意：只有在處理來自未連接協議（例如以太網或UDP）的數據包時，纔可能發生這種情況。以TCP爲例，它將不起作用。

Answer 3

冒着不回答你問的問題，我會尋找軟件，在較低的水平上做到這一點。

Netlimiter這是否適用於Windows。我認爲BWMeter和Bandwidth Controller也可以做到。

pyshaper是一個類似Linux的工具。開源。您可能只能將它導入到您的Python程序中。

（另一件需要考慮的事情是，您可能已經擁有了一臺能夠以您想要的方式改變流量的路由器，不過這對您的軟件增加了相當大的依賴性，並且可能需要更多的配置工作。）