Indigo拥塞控制算法发送端代码解读
本文章将不定时进行更新，欢迎收藏。
相关源文件
Indigo拥塞控制算法集成于Pantheon拥塞控制算法测试平台中，该测试平台由斯坦福大学系统与网络研究组（Stanford Systems and Networking Research）设计与实现，其开源代码在Github上的地址为Pantheon.
Indigo拥塞控制算法单独的GIthub地址为Indigo.
发送端的主要相关文件为以下：
- pantheon/third_party/indigo/env/sender.py
- pantheon/third_party/indigo/dagger/run_sender.py
引入的包
run_sender.py
该文件中引入了这些包：
import argparse
import project_root
import numpy as np
import tensorflow as tf
from os import path
from env.sender import Sender
from models import DaggerLSTM
from helpers.helpers import normalize, one_hot, softmax
这些包中，argparse,numpy,tensorflow,os.path等包都来自于内置或第三方包文件，project_root,env.sender.Sender,DaggerLSTM,helpers源于作者自己实现。
sender.py
该文件中引入了这些包：
import sys
import json
import socket
import select
from os import path
import numpy as np
import datagram_pb2
import project_root
from helpers.helpers import (
curr_ts_ms, apply_op,
READ_FLAGS, ERR_FLAGS, READ_ERR_FLAGS, WRITE_FLAGS, ALL_FLAGS)
这些包中，sys,json,socket,select,os.path,numpy等包都来自于内置或第三方包文件，datagram_pb2,project_root,helpers源于作者自己实现。
datagram_pb2
datagram_pb2是一个基于Google Protocol Buffer的包，实现于文件datagram_pb2.py中，该包由Google Protocol Buffer Compiler生成。
Google Protocol Buffer( 简称 Protobuf) 是 Google 公司内部的混合语言数据标准，目前已经正在使用的有超过 48,162 种报文格式定义和超过 12,183 个 .proto 文件。他们用于 RPC 系统和持续数据存储系统。
简单来说，可以预先在.proto文件中定义数据的格式，然后通过谷歌提供的编译器进行编译即可。目前提供了 C++、Java、Python 三种语言的 API，可以通过编译生成这三种语言的库\包文件。
socket
在类Sender的构造函数中，使用到了self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
代码，该代码中socket.AF_INET代表使用IPv4协议，socket.SOCK_DGRAM代表使用UPD协议（SOCK_STREAM代表使用TCP协议，SOCK_RAW 可以处理ICMP、IGMP等网络报文、特殊的IPv4报文、可以通过IP_HDRINCL套接字选项由用户构造IP头。）
主控制流程
在源文件run_sender.py中，定义了一个基于LSTM的学习器，以类Learner的形式出现，其实现原理不再讨论，具体设计细节移步Pantheon的论文。
发送端的主控制流程位于run_sender.py中，其预处理为如下：
def main():
# 创建一个ArgumentParser对象
parser = argparse.ArgumentParser()
# 加入一个名为port的变量，其类型为整形，从命令行中得到
parser.add_argument('port', type=int)
parser.add_argument('--debug', action='store_true')
# 进行命令行参数解读
args = parser.parse_args()
# 构建一个Sender对象，该对象在sender.py中有定义
sender = Sender(args.port, debug=args.debug)
model_path = path.join(project_root.DIR, 'dagger', 'model', 'model')
# 学习
learner = Learner(
state_dim=Sender.state_dim,
action_cnt=Sender.action_cnt,
restore_vars=model_path)
sender.set_sample_action(learner.sample_action)
预处理完后进行连接的建立、数据的发送以及连接的清理，其具体代码如下：
try:
sender.handshake()
sender.run()
except KeyboardInterrupt:
# 占位语句
pass
finally:
sender.cleanup()
这三个函数都实现于类Sender中。
连接的建立
handshake函数
发送端的连接函数实现如下：
def handshake(self):
"""Handshake with peer receiver. Must be called before run()."""
while True:
# sock.recvfrom(1600)表示该函数接受来自socket的数据，缓冲区大小为1600
msg, addr = self.sock.recvfrom(1600)
# self.peer_address代表之前还没有建立连接
if msg == 'Hello from receiver' and self.peer_addr is None:
# 如果发现建立了连接，那么将该连接对方的地址设置成为接收到的地址
self.peer_addr = addr
# 向接收到的地址发送“已接受到”信息
self.sock.sendto('Hello from sender', self.peer_addr)
sys.stderr.write('[sender] Handshake success! '
'Receiver\'s address is %s:%s\n' % addr)
break
self.sock.setblocking(0) # non-blocking UDP socket
数据的发送
run函数
数据发送函数部分实现如下：
def run(self):
TIMEOUT = 1000 # ms
self.poller.modify(self.sock, ALL_FLAGS)
curr_flags = ALL_FLAGS
在该类进行初始化的时候，执行了以下两行代码：
self.poller = select.poll()
self.poller.register(self.sock, ALL_FLAGS)
select.poll()函数官方的定义如下：
(Not supported by all operating systems.) Returns a polling object, which supports registering and unregistering file descriptors, and then polling them for I/O events; see section Polling Objects below for the methods supported by polling objects.
网友评论