引入的包

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，可以通过编译生成这三种语言的库\包文件。

该数据格式定义于文件pantheon/third_party/indigo/envdatagram.proto中，该文件的内容如下：

syntax = "proto3";

message Data {
  fixed32 seq_num = 1;
  fixed32 send_ts = 2;
  fixed64 sent_bytes = 3;
  fixed32 delivered_time = 4;
  fixed64 delivered = 5;
  string payload = 6;
}

message Ack {
  fixed32 seq_num = 1;
  fixed32 send_ts = 2;
  fixed64 sent_bytes = 3;
  fixed32 delivered_time = 4;
  fixed64 delivered = 5;
  fixed32 ack_bytes = 6;
}

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.

Indigo拥塞控制算法发送端代码解读

相关源文件