DN功能实现（七）---修改OVS源码实现ACK机制(控制器-交换机用户态-交换机内核态通信)

(一)系统部分架构说明

如图所示,其中s1\s4是边缘交换机,s1直接与发送方相连,作为近接受方交换机存在,s4直接与接受方相连,作为近接受方交换机存在.

为了使目的终端获取得到唯一的\可靠\实时传输的UDP数据包,源终端主机发送的数据经由近接收方交换机s1,若流表中没有匹配项,则发送Pakcet-in消息给控制器,控制器获取rudp数据后,通过拓扑发现和时延探测模块获取网络全局拓扑结构,预选取两条节点不相关的主\备份路径{s1-s2-s3-s4,s1-s5-s6-s4}和关键节点{s1,s2,s4,s6},下发对应流表项.数据包在时延最优路径{s1-s2-s3-s4}中进行传输,经过关键节点{s1,s2,s4}时,需要触发ACK报文处理机制,各个关键节点通过控制器进行ACK报文的回送和确认,以保障数据传输的可靠性,减少由于路径故障导致的丢包问题.

其中若是关键节点之间的链路故障\拥塞导致数据包无法实时到达下一个关键节点,则通过内部定时机制或下一个关键节点回送的缺失类型ACK决定对丢失数据包的重传,重传包括原关键节点同时向主路径端口和备份路径端口进行转发,以提高数据传输的实时性和可靠性.如图所示,对于关键节点s2转发数据包{packet1,packet2,packet3}在链路{s2-s3-s4},其中packet2由于链路{s2-s3}故障导致丢失,因此在下一个关键节点s4中无法接收到packet2,通过发送ack报文,请求前一个关键节点进行数据的重传,在关键节点s2接受到重传ack后,从缓存队列中获取对应的数据包,通过端口分别向主路径进行重传和备份路径进行转发,同样关键节点s2内部的定时机制也可以触发重传机制.

(二)ACK机制

补充:RUDP数据报文字段如下图,通过对FrontDPID和PacketNumber字段数据的获取,用于支持ACK机制的实现.

其中各个字段属性如下：

在系统实现中,关键节点交换机(如S4)使用了数据缓存队列缓存所有需要可靠传输服务的RUDP数据包,同时通过开启内核线程实现定时检测队列,如果有新的数据达到,则对最新的数据的包序号进行ACK转发,从内核态通过Generic Netlink机制上传至用户态,用户态通过openflow协议传输至控制器,控制器通过获取的包序号和前一个关键节点交换机(S2)信息,将ACK报文通过OpenFlow协议转发至S2交换机用户空间,S2同样通过Generic Netlink转发至内核态,通过对ACK数据解析,S2获取得到S4中包接收的顺序,因此可以释放数据队列中已经确认的ACK序号前面的所有数据空间,对于丢失的数据,同样通过ACK可以实现重传和转发.

本文主要介绍ACK报文在两个交换机的内核态-用户态-控制器-用户态-内核态之间的转发过程.

其中Generic Netlink学习,可以看前面转载的几篇博客!!!

二:扩展OpenFlow协议,实现控制器和交换机用户态通信

(一)Ryu源码修改

1.修改ofproto/ofproto_v1_3.py,声明新的OpenFlow消息类型

OFPT_GET_ASYNC_REQUEST = 26 # Controller/switch message
OFPT_GET_ASYNC_REPLY = 27 # Controller/switch message
OFPT_SET_ASYNC = 28 # Controller/switch messageOFPT_METER_MOD = 29 # Controller/switch message#与前面的消息类型编码不一致（唯一即可）
OFPT_ACK_REQUEST = 32               # Symmetric message
OFPT_ACK_REPLY = 33                 # Symmetric message

2.修改ofproto/ofproto_v1_3_parser.py,实现对新的ACK消息支持

@_register_parser
@_set_msg_type(ofproto.OFPT_ACK_REQUEST)
class OFPACKRequest(MsgBase):def __init__(self, datapath, data=None):super(OFPACKRequest, self).__init__(datapath)self.data = data@classmethoddef parser(cls, datapath, version, msg_type, msg_len, xid, buf):msg = super(OFPACKRequest, cls).parser(datapath, version, msg_type,msg_len, xid, buf)msg.data = msg.buf[ofproto.OFP_HEADER_SIZE:]return msgdef _serialize_body(self):if self.data is not None:self.buf += self.data@_register_parser
@_set_msg_type(ofproto.OFPT_ACK_REPLY)
class OFPACKReply(MsgBase):def __init__(self, datapath, data=None):super(OFPACKReply, self).__init__(datapath)self.data = data@classmethoddef parser(cls, datapath, version, msg_type, msg_len, xid, buf):msg = super(OFPACKReply, cls).parser(datapath, version, msg_type,msg_len, xid, buf)msg.data = msg.buf[ofproto.OFP_HEADER_SIZE:]return msgdef _serialize_body(self):assert self.data is not Noneself.buf += self.data

3.完成1,2步骤后记得进行Ryu源码的重新编译!步骤在前面的SDN实现中

4.实现Ryu App应用对新的ACK消息进行转发

from ryu.base import app_manager
from ryu.controller import ofp_event
from ryu.controller.handler import CONFIG_DISPATCHER, MAIN_DISPATCHER
from ryu.controller.handler import MAIN_DISPATCHER,CONFIG_DISPATCHER,DEAD_DISPATCHER,HANDSHAKE_DISPATCHER #只是表示datapath数据路径的状态
from ryu.controller.handler import set_ev_cls
from ryu.ofproto import ofproto_v1_3
from ryu.lib.packet import packet
from ryu.lib.packet import ethernet
from ryu.lib.packet import ether_types
from ryu.lib.packet import udp
from ryu.lib.packet import ipv4
from struct import packRUDP_MAGIC = 38217class RudpTrans(app_manager.RyuApp):OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]def __init__(self, *args, **kwargs):super(RudpTrans, self).__init__(*args, **kwargs)self.mac_to_port = {}self.datapaths = {}self.initFlag = False@set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)def switch_features_handler(self, ev):datapath = ev.msg.datapathself.datapaths[datapath.id] = datapathofproto = datapath.ofprotoparser = datapath.ofproto_parser# install table-miss flow entry## We specify NO BUFFER to max_len of the output action due to# OVS bug. At this moment, if we specify a lesser number, e.g.,# 128, OVS will send Packet-In with invalid buffer_id and# truncated packet data. In that case, we cannot output packets# correctly.  The bug has been fixed in OVS v2.1.0.match = parser.OFPMatch()actions = [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER,ofproto.OFPCML_NO_BUFFER)]self.add_flow(datapath, 0, match, actions)def add_flow(self, datapath, priority, match, actions, buffer_id=None):ofproto = datapath.ofprotoparser = datapath.ofproto_parserinst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS,actions)]if buffer_id:mod = parser.OFPFlowMod(datapath=datapath, buffer_id=buffer_id,priority=priority, match=match,instructions=inst)else:mod = parser.OFPFlowMod(datapath=datapath, priority=priority,match=match, instructions=inst)datapath.send_msg(mod)def _send_ack_request(self,dp_id,hash_id,packet_number,type_id):"""发送echo_time报文到datapath"""    print("===============_send_ack_request========start============")print("front dpid / hash id / packet number / type id : ",dp_id, hash_id, packet_number, type_id)datapath = self.datapaths.get(dp_id,None)if datapath == None:returnparser = datapath.ofproto_parserprint("========%d"%packet_number);ack_req = parser.OFPACKRequest(datapath,data=pack("BBBI",dp_id,hash_id,type_id,packet_number)) datapath.send_msg(ack_req)print("==========_send_ack_request=========end=====")@set_ev_cls(ofp_event.EventOFPACKReply,[MAIN_DISPATCHER,CONFIG_DISPATCHER,HANDSHAKE_DISPATCHER])def ack_reply_handler(self,ev):"""处理ACK响应报文，获取控制器到交换机的链路往返时延"""print("==================OFPT_ACK_REPLY============start===========")print(ev.msg.data)info = ev.msg.datafront_dpid = int.from_bytes(info[0:1],'little')hash_id = int.from_bytes(info[1:2],'little')packet_id = int.from_bytes(info[2:6],'little')type_id = int.from_bytes(info[6:7],'little')self._send_ack_request(front_dpid,hash_id,packet_id,type_id)  print("==================OFPT_ACK_REPLY===========end============") 

（免费订阅,永久学习）学习地址: Dpdk/网络协议栈/vpp/OvS/DDos/NFV/虚拟化/高性能专家-学习视频教程-腾讯课堂

更多DPDK相关学习资料有需要的可以自行报名学习,免费订阅,永久学习,或点击这里加qun免费
领取,关注我持续更新哦! ! 

(二)OVS源码修改,实现对新的OpenFlow消息支持

1.修改include/openvswitch/ofp-msgs.h下的enum ofptype和enum ofpraw枚举类型

enum ofptype {OFPTYPE_HELLO, /* OFPRAW_OFPT_HELLO. */OFPTYPE_ERROR, /* OFPRAW_OFPT_ERROR. ......OFPTYPE_ACK_REQUEST, /* OFPRAW_OFPT_ACK_REQUEST */OFPTYPE_ACK_REPLY, /* OFPRAW_OFPT_ACK_REPLY */......
}

注意：注释对应了下面ofpraw枚举的名称

enum ofpraw {....../* OFPT 1.1-1.3 (32): uint8_t[]. */OFPRAW_OFPT_ACK_REQUEST,/* OFPT 1.1-1.3 (33): uint8_t[]. */OFPRAW_OFPT_ACK_REPLY,......
}

注意:序号需要保持与Ryu中的消息序号一致！！！

2.修改在ofproto/ofproto.c文件中handle_flow函数下的handle_single_part_openflow方法，处理自定义的类型

static enum ofperr
handle_single_part_openflow(struct ofconn *ofconn, const struct ofp_header *oh,enum ofptype type)OVS_EXCLUDED(ofproto_mutex)
{switch (type) {/* OpenFlow requests. */......case OFPTYPE_ACK_REQUEST:// VLOG_INFO("--------------------OFPTYPE_ACK_REQUEST--------------------");return handle_ack_request(ofconn,oh);......case OFPTYPE_ACK_REPLY:return 0;......  
}

同样在该文件中实现handle_ack_request方法：

static enum ofperr
handle_ack_request(struct ofconn *ofconn, const struct ofp_header *oh)
{//不同的是,获取数据,发送消息给内核// VLOG_INFO("----handle_ack_request-------start----");if(!ack_init) return OFPERR_OFPAFC_INIT;if(!ack_oh || ack_oh != oh) ack_oh = oh;struct ofpbuf rq_buf = ofpbuf_const_initializer(oh, ntohs(oh->length));//rq_buf.data是我们的数据,但是前面8个字节是结构体ofp_header,所以我们获取真正的数据需要跳过这8个字节数据char *data = rq_buf.data;// uint32_t packetid = ack_parse_packet_number(data+8,rq_buf.size-8);uint8_t front_dpid = *(uint8_t *)(data + 8);uint8_t hash_id = *(uint8_t *)(data + 9);uint8_t typeid = *(uint8_t *)(data + 10);uint32_t packetid = *(uint32_t *)(data + 12);   //存在对齐问题，所以间隔2//通过以上操作，可以获取并解析出控制器转发的ACK消息数据// VLOG_INFO("--handle_ack_request---front dpid:%u--hash id:%u--packet id:%u- type:%u-------",front_dpid,hash_id,packetid,typeid);dpif_netlink_ack_reply(false,front_dpid,hash_id,packetid,typeid);return 0;
}

其中dpif_netlink_ack_reply方案用于将解析的数据通过Generic Netlink转发至内核态，将在下一章中进行详细介绍！

三:实现交换机内核态与用户态通信

(一)用户态主动与内核态通信

1.在内核态进行新的Generic Netlink套接字注册

（1）在datapath/linux/compat/include/linux/openvswitch.h中进行类型声明

/*     ACK.    */#define OVS_ACK_FAMILY     "ovs_ack"
#define OVS_ACK_VERSION 0x1enum ovs_ack_cmd {OVS_ACK_CMD_UNSPEC,OVS_ACK_CMD_ECHO,                          //用户发给内核，该命令将由内核进行处理OVS_ACK_CMD_SET,                           //用戶发送给内核,进行设置OVS_ACK_CMD_REPLY,                         //由内核发送响应给用户态，该命令将由用户态进行处理__OVS_ACK_CMD_MAX,
};
#define OVS_ACK_CMD_MAX (__OVS_ACK_CMD_MAX - 1)enum ovs_ack_attr {OVS_ACK_ATTR_UNSPEC = 0,OVS_ACK_ATTR_INIT,                        //flag标识初始化OVS_ACK_ATTR_VID,                        //8bits无符号整型OVS_ACK_ATTR_OUTPUT,                    //16bits无符号整型OVS_ACK_ATTR_TURNOUT,                    //16bits无符号整型OVS_ACK_ATTR_QUEUE_SIZE,                //16bits无符号整型OVS_ACK_ATTR_TIME_INTERVAL,                //16bits无符合整型OVS_ACK_ATTR_DEVSTATUS,                    //8bits无符号整型OVS_ACK_ATTR_HASH_ID,                    //8bits无符号整型OVS_ACK_ATTR_TYPE,                        //8bits无符号整型OVS_ACK_ATTR_NUMBER,                    //32bits无符号整型__OVS_ACK_ATTR_MAX,
};
#define OVS_ACK_ATTR_MAX (__OVS_ACK_ATTR_MAX - 1)

其中红色的字段与本次的ACK机制有关,其他的是关于其他功能的,这里不需要关心.

另外需要补充,datapath/linux/compat/include/linux/openvswitch.h在编译过程中会自动生成一个include/odp-netlink.h文件,我们在用户空间中使用这个头文件即可.

（2）在datapath/datapath.c中进行实现

Ⅰ.为模块添加Generic Netlink family信息

module_init(dp_init);
module_exit(dp_cleanup);MODULE_DESCRIPTION("Open vSwitch switching datapath");
MODULE_LICENSE("GPL");
MODULE_VERSION(VERSION);
MODULE_ALIAS_GENL_FAMILY(OVS_DATAPATH_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_VPORT_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_FLOW_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_PACKET_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_ACK_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_METER_FAMILY);
MODULE_ALIAS_GENL_FAMILY(OVS_CT_LIMIT_FAMILY);

Ⅱ.定义策略、操作集和family

static const struct nla_policy ack_policy[OVS_ACK_ATTR_MAX + 1] = {    [OVS_ACK_ATTR_INIT]            =    {.type = NLA_FLAG},[OVS_ACK_ATTR_VID]            =    {.type = NLA_U8},[OVS_ACK_ATTR_HASH_ID]        =    {.type = NLA_U8},[OVS_ACK_ATTR_NUMBER]        =    {.type = NLA_U32},[OVS_ACK_ATTR_TYPE]            =    {.type = NLA_U8},[OVS_ACK_ATTR_OUTPUT]        =    {.type = NLA_U16},[OVS_ACK_ATTR_TURNOUT]        =    {.type = NLA_U16},[OVS_ACK_ATTR_QUEUE_SIZE]        =    {.type = NLA_U16},[OVS_ACK_ATTR_TIME_INTERVAL]        =    {.type = NLA_U16},[OVS_ACK_ATTR_DEVSTATUS]    =    {.type = NLA_U8},
};                                                                        //其他的为默认，NLA_UNSPEC表示类型和长度未知static const struct genl_ops dp_ack_genl_ops[] = {{.cmd         =    OVS_ACK_CMD_ECHO,.doit         =     ovs_ack_cmd_execute,.policy     =     ack_policy,.flags         =    0,},{.cmd         =    OVS_ACK_CMD_SET,.doit         =     ovs_ack_cmd_set,.policy     =     ack_policy,.flags         =    0,},
};static struct genl_family dp_ack_genl_family __ro_after_init = {.id      =    0,