Use bigger buffer size when recv()

Author: francisyyan

env/environment.py

@@ -1,10 +1,10 @@
 import os
+from os import path
 import sys
 import signal
-import project_root
-from os import path
-from sender import Sender
 from subprocess import Popen
+from sender import Sender
+import project_root
 from helpers.helpers import get_open_udp_port
 
 
@@ -44,6 +44,7 @@ def setup(self):
     def set_sample_action(self, sample_action):
         """Set the sender's policy. Must be called before run()."""
 
+        self.sample_action = sample_action
         self.sender.set_sample_action(sample_action)
 
     def rollout(self):
@@ -60,11 +61,19 @@ def rollout(self):
     def cleanup(self):
         if self.sender:
             self.sender.cleanup()
+            self.sender = None
 
         if self.receiver:
             try:
                 os.killpg(os.getpgid(self.receiver.pid), signal.SIGTERM)
             except OSError as e:
                 sys.stderr.write('%s\n' % e)
+            finally:
+                self.receiver = None
 
-        sys.stderr.write('\nEnvironment cleaned up.\n')
+    def reset(self):
+        self.cleanup()
+        self.setup()
+
+        if self.sender:
+            self.sender.set_sample_action(self.sample_action)

env/receiver.py

@@ -68,7 +68,7 @@ def handshake(self):
                     sys.exit('Channel closed or error occurred')
 
                 if flag & READ_FLAGS:
-                    msg, addr = self.sock.recvfrom(1500)
+                    msg, addr = self.sock.recvfrom(1600)
 
                     if addr == self.peer_addr:
                         if msg != 'Hello from sender':
@@ -83,7 +83,7 @@ def run(self):
         self.sock.setblocking(1)  # blocking UDP socket
 
         while True:
-            serialized_data, addr = self.sock.recvfrom(1500)
+            serialized_data, addr = self.sock.recvfrom(1600)
 
             if addr == self.peer_addr:
                 ack = self.construct_ack_from_data(serialized_data)

env/sender.py

@@ -27,7 +27,7 @@ def __init__(self, port=0, train=False, debug=False):
 
         # UDP datagram template
         self.data = {}
-        self.data['payload'] = 'x' * 1400
+        self.data['payload'] = 'x' * 1350
 
         # dimension of state space and action space
         self.state_dim = 4
@@ -45,8 +45,7 @@ def __init__(self, port=0, train=False, debug=False):
         self.prev_recv_ts = None
 
         if self.train:
-            self.max_step_cnt = 2000
-            self.max_runtime = 5000
+            self.max_runtime = 10000
 
             # statistics variables to compute rewards
             self.sent_bytes = 0
@@ -62,7 +61,7 @@ def handshake(self):
         """Handshake with peer receiver. Must be called before run()."""
 
         while True:
-            msg, addr = self.sock.recvfrom(1500)
+            msg, addr = self.sock.recvfrom(1600)
 
             if msg == 'Hello from receiver' and self.peer_addr is None:
                 self.peer_addr = addr
@@ -79,7 +78,7 @@ def set_sample_action(self, sample_action):
         self.sample_action = sample_action
 
     def reset(self):
-        """Reset the sender. Must be called in every training iteration."""
+        """Reset the sender. Must be called after every training iteration."""
 
         self.seq_num = 0
         self.next_ack = 0
@@ -89,13 +88,14 @@ def reset(self):
         self.prev_send_ts = None
         self.prev_recv_ts = None
 
-        self.sent_bytes = 0
-        self.acked_bytes = 0
-        self.first_recv_ts = float('inf')
-        self.last_recv_ts = 0
-        self.total_delays = []
+        if self.train:
+            self.sent_bytes = 0
+            self.acked_bytes = 0
+            self.first_recv_ts = float('inf')
+            self.last_recv_ts = 0
+            self.total_delays = []
 
-        self.drain_packets()
+            self.drain_packets()
 
     def drain_packets(self):
         """Drain all the packets left in the channel."""
@@ -116,7 +116,7 @@ def drain_packets(self):
                     sys.exit('Channel closed or error occurred')
 
                 if flag & READ_FLAGS:
-                    self.sock.recvfrom(1500)
+                    self.sock.recvfrom(1600)
 
     def update_state(self, ack):
         send_ts = ack['ack_send_ts']
@@ -145,10 +145,6 @@ def update_state(self, ack):
             self.first_recv_ts = min(recv_ts, self.first_recv_ts)
             self.last_recv_ts = max(recv_ts, self.last_recv_ts)
 
-            self.step_cnt += 1
-            if self.step_cnt >= self.max_step_cnt:
-                self.running = False
-
             if curr_ts_ms() - self.runtime_start > self.max_runtime:
                 self.running = False
 
@@ -157,9 +153,7 @@ def update_state(self, ack):
 
     def take_action(self, action):
         self.cwnd += self.action_mapping[action]
-
-        if self.cwnd < 5.0:
-            self.cwnd = 5.0
+        self.cwnd = max(5.0, self.cwnd)
 
         if self.debug:
             sys.stderr.write('cwnd %.2f\n' % self.cwnd)
@@ -175,8 +169,9 @@ def compute_reward(self):
         delay_percentile = float(np.percentile(self.total_delays, 95))
         loss_rate = 1.0 - float(self.acked_bytes) / self.sent_bytes
 
-        reward = np.log(max(1e-4, avg_throughput))
-        reward -= np.log(max(1.0, delay_percentile / 10.0))
+        reward = 2 * np.log(max(1e-3, avg_throughput))
+        reward -= np.log(max(1.0, delay_percentile))
+        reward += np.log(1.0 - loss_rate)
 
         sys.stderr.write('Average throughput: %.2f Mbps\n' % avg_throughput)
         sys.stderr.write('95th percentile one-way delay: %d ms\n' %
@@ -204,7 +199,7 @@ def send(self):
             sys.stderr.write('Sent seq_num %d\n' % int(self.data['seq_num']))
 
     def recv(self):
-        serialized_ack, addr = self.sock.recvfrom(1500)
+        serialized_ack, addr = self.sock.recvfrom(1600)
 
         if addr != self.peer_addr:
             return
@@ -231,7 +226,6 @@ def run(self):
         curr_flags = ALL_FLAGS
 
         self.running = True
-        self.step_cnt = 0
         self.runtime_start = curr_ts_ms()
 
         while not self.train or self.running:

env/run_sender.py renamed to reinforce/run_sender.py

@@ -1,10 +1,9 @@
 #!/usr/bin/env python
 
+from os import path
 import sys
-import project_root
 import numpy as np
-import tensorflow as tf
-from os import path
+import project_root
 from env.environment import Environment
 
 
@@ -39,6 +38,7 @@ def cleanup(self):
     def run(self):
         for episode_i in xrange(1, 3):
             sys.stderr.write('\nEpisode %d\n' % episode_i)
+            self.env.reset()
 
             # get an episode of experience
             final_reward = self.env.rollout()

env/train_sender.py renamed to reinforce/train_sender.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 
-import project_root
 import numpy as np
+import project_root
 from helpers.helpers import RingBuffer, MeanVarHistory
 
 

tests/test_environment.py

@@ -2,8 +2,8 @@
 
 import sys
 import argparse
-import project_root
 import numpy as np
+import project_root
 from env.sender import Sender