Statistics
| Branch: | Revision:

wcn_emulator / wcn_simulator.py @ 8f01e99f

History | View | Annotate | Download (10.4 KB)

1
#!/usr/bin/env python
2
import sys
3
sys.path.append('community_networks_analysis')
4

    
5
from mininet.net import Mininet
6
from mininet.log import setLogLevel
7
from mininet.node import OVSController
8
from mininet.cli import CLI
9
from mininet.log import info, error, debug, output
10
from mininet.node import CPULimitedHost
11
from mininet.link import TCLink
12

    
13
import networkx as nx
14
from time import sleep, time
15
from os import kill, path, makedirs
16
from psutil import Process
17
import signal
18
from matplotlib.pyplot import ion
19
from random import sample, randint
20

    
21
from gengraphs import loadGraph
22
from misclibs import showGraph
23

    
24
from parameters_parser import parameters
25

    
26
sys.path.append("community_networks_analysis/")
27

    
28
class PowerNet(Mininet):
29
    def __init__(self,**params):
30
        if 'controller' not in params.keys():
31
            params['controller'] = OVSController
32
        if 'host' not in params.keys():
33
            params['host'] = CPULimitedHost
34
        if 'link' not in params.keys():
35
            params['link'] = TCLink
36
        super(PowerNet,self).__init__(**params)
37

    
38
    def enableForwarding(self):
39
        for node in self.values():
40
            node.cmd("echo 1 > /proc/sys/net/ipv4/ip_forward")
41

    
42
    def setNeighboursRoutes(self):
43
        for node in self.values():
44
            for intf in node.intfList():
45
                if intf.link:
46
                    rintf = self.remoteIntf(intf)
47
                    raddrs = self.getNodeAddrs(rintf.node)
48
                    for addr in raddrs:
49
                        node.setHostRoute(addr,intf.name)
50

    
51
    def getNodeAddrs(self,node):
52
        r = []
53
        for intf in node.intfList():
54
            if intf.link:
55
                r.append(intf.ip)
56
        return r
57

    
58
    def remoteIntf(self,intf):
59
        if intf.link:
60
            intfs = [ intf.link.intf1, intf.link.intf2 ]
61
            intfs.remove(intf)
62
            return intfs[0]
63
        return None
64

    
65

    
66
class GraphNet(PowerNet):
67
    def __init__(self,edges_file,draw=True,**params):
68
        super(GraphNet,self).__init__(**params)
69
        info("\nReading "+edges_file+"\n")
70

    
71
        g = loadGraph(edges_file, connected=True)
72

    
73
        nodeCounter = 0
74
        nodeMap = {}
75
        for name in g.nodes():
76
            nodeMap[name] = "h"+str(name)+"_"+str(nodeCounter)
77
            nodeCounter += 1
78

    
79
        self.gg = nx.relabel_nodes(g,nodeMap)
80

    
81
        self.hosts_port = {}
82

    
83
        # add nodes
84
        for n in self.gg.nodes():
85
            self.addHost(n)
86
            self.hosts_port[n] = 1 
87

    
88
        # add edges
89
        for e in self.gg.edges(data=True):
90
            # 10 Mbps, 5ms delay, 10% loss, 1000 packet queue
91
            # htp: Hierarchical Token Bucket rate limiter
92
#            quality_params = {"bw":10,"delay":'5ms', "loss":100-100.0/e[2]['weight'], "use_htb":True}
93
            quality_params = {}
94
            quality_params["bw"] = 10
95
#            quality_params["delay"] = '5ms'
96
#            quality_params["loss"] = 100-100.0/e[2]['weight']
97
            quality_params["use_htb"] = True
98
            self.insertLink(self.get(e[0]),self.get(e[1]),quality_params)
99

    
100
        if draw:
101
            showGraph(self.gg)
102

    
103
    def pickHostAddrPort(self, node):
104
        port = self.hosts_port[node.name]
105
        addr = "10.0."+node.name.split('_')[-1]+"."+str(port)+"/8"
106
        self.hosts_port[node.name] += 1
107
        return addr,port
108

    
109
    def insertLink(self, n1, n2, quality_params={}):
110
        addr1, port1 = self.pickHostAddrPort(n1)
111
        addr2, port2 = self.pickHostAddrPort(n2)
112

    
113
        self.addLink(n1, n2,  \
114
                port1 = port1, \
115
                port2 = port2, \
116
                params1=dict([('ip',addr1)] + quality_params.items()), \
117
                params2=dict([('ip',addr2)] + quality_params.items()) \
118
                )
119

    
120
    def setShortestRoutes(self):
121
        paths = nx.shortest_path(self.gg,weight='weight')
122
        for node1 in paths.keys():
123
            host1 = self.get(node1)
124
            debug("Starting node: "+node1+'\n')
125
            debug("\tpaths: "+str(paths[node1])+'\n')
126
            for node2 in paths[node1].keys():
127
                if node2 != node1 :
128
                    if len(paths[node1][node2])>2:
129
                        debug("\tDestination node: "+node2+'\n')
130
                        nextHop = self.get(paths[node1][node2][1])
131
                        debug("\tNextHop node: "+nextHop.name+'\n')
132
                        dsts = self.getNodeAddrs(self.get(node2))
133
                        intfs = host1.connectionsTo(nextHop)
134
                        nextAddrs = [ couple[1].ip for couple in intfs ]
135
                        rintf = intfs[0][0] # WARNING we just consider one link
136
                        for dst in dsts:
137
                            for addr in nextAddrs:
138
                                host1.cmd("ip route add "+dst+" via "+addr+" dev "+rintf.name)
139
                                debug("\tip route add "+dst+" via "+addr+'\n')
140
                    else :
141
                        host2 = self.get(node2)
142
                        intfs = [ couple[0] for couple in host1.connectionsTo(host2) ]
143
                        rintf = intfs[0] # WARNING we just consider one link
144
                        raddrs = self.getNodeAddrs(host2)
145
                        for addr in raddrs:
146
                            host1.setHostRoute(addr,rintf.name)
147

    
148

    
149
class MininetTest(object):
150
    def __init__(self,mininet):
151
        self.net = mininet
152
        self.pendingProc = {} 
153
    
154
    def getHostSample(self,num):
155
        hosts = sample(self.net.values(),num)
156
        return hosts[:num]
157

    
158
    def bgCmd(self,host,force_multiple_processes,*args):
159
        # here it's a little workaround for tracing the resulting pid
160
        # it launch the new process using the mininet interface
161
        # but it check the newly created process id using psutil
162
        host_proc = Process(host.pid)
163
        host_ps = set(host_proc.get_children())
164
        debug("Sending cmd: \n\t"+str(args)+"\n")
165
        if force_multiple_processes:
166
            host.waiting = False
167
        host.sendCmd(*(args+("&",)))
168
        sleep(0.5)
169
        try :
170
            pid = (set(host_proc.get_children()).difference(host_ps)).pop().pid
171
            info("BGProcess: "+str(pid)+"; ")
172
            self.pendingProc[pid] = host
173
        except:
174
            info("*** Unable to launch command:\n\t "+str(args))
175
            return None
176
        return pid
177

    
178
    def sendSig(self,pid,sig=signal.SIGTERM):
179
        try:
180
            info("Killing BGProcess: "+str(pid)+"; ")
181
            kill( pid, sig )
182
        except OSError:
183
            error("Error while killing process "+str(pid))
184
            pass
185

    
186
    def killAll(self):
187
        for pid in self.pendingProc.keys():
188
            self.sendSig(pid,signal.SIGKILL)
189
            self.pendingProc[pid].monitor() # wait exiting
190
        self.pendingProc.clear()
191

    
192
class PSTest(MininetTest):
193
    def __init__(self,mininet,duration=300):
194
        super(PSTest,self).__init__(mininet)
195
        self.source = None
196
        self.hosts = []
197
        self.duration = duration
198
        self.prefix = ''
199

    
200
    def setPrefix(self,name):
201
        self.prefix = str(name)+'_'+str(self.duration)+'_'+str(len(self.hosts)+1)+'hosts/' 
202
        if not path.exists(self.prefix):
203
                makedirs(self.prefix)
204

    
205
    def launchPS(self,host,params,stdout,stderr):
206
        cmd = "./streamer"
207
        params['-c'] = '38'
208
#        params['-M'] = '5'
209
#        params['-O'] = '3'
210
        params['--chunk_log'] = ''
211
        params['>'] = stdout
212
        params['2>'] = stderr
213
        return self.bgCmd(host,True,cmd,*reduce(lambda x, y: x + y, params.items()))
214

    
215
    def launchPeer(self,host,source,source_port=7000):
216
        idps = randint(0,100)
217
        logfile = self.prefix+host.name.split('_')[0]+"-"+str(idps)+"_peerstreamer_normal_$(date +%s).log"
218
        params = {}
219
        params['-i'] = source.defaultIntf().ip
220
        params['-p'] = str(source_port)
221
        params['-P'] = str(randint(4000,8000))
222
        return self.launchPS(host,params,'/dev/null',logfile)
223

    
224
    def launchSource(self,host,chunk_mult=1,source_port=7000):
225
        idps = randint(0,100)
226
        video_file = "bunny.ts,loop=1"
227
        logfile = self.prefix+host.name.split('_')[0]+"-"+str(idps)+"_source_normal_$(date +%s).log"
228
        params = {}
229
        params['-I'] = host.defaultIntf().name
230
        params['-P'] = str(source_port)
231
        params['-f'] = video_file
232
        params['-m'] = str(chunk_mult)
233
        return self.launchPS(host,params,'/dev/null',logfile)
234

    
235
    def runTest(self):
236
        info("*** Launching PeerStreamer test\n")
237
        info("Data folder: "+self.prefix+"\n")
238
        if self.source:
239
            self.launchSource(self.source)
240

    
241
        for h in self.hosts:
242
            self.launchPeer(h,self.source)
243
        info("Waiting completion...\n")
244
        sleep(self.duration)
245

    
246
        self.killAll()
247

    
248
class PSHostsTest(PSTest):
249
    def __init__(self,mininet,source_name,peer_names,duration=300,name=None):
250
        super(PSHostsTest,self).__init__(mininet,duration=duration)
251
        self.source = mininet.get(source_name)
252
        for n in peer_names:
253
            self.hosts.append(mininet.get(n))
254
        self.setPrefix(name)
255

    
256
class PSRandomTest(PSTest):
257
    def __init__(self,mininet,duration=300,num_peers=5,name=None):
258
        super(PSRandomTest,self).__init__(mininet,duration)
259
        self.hosts = self.getHostSample(num_peers)
260
        if len(self.hosts) > 0:
261
            self.source = self.hosts.pop()
262
        self.setPrefix(name)
263

    
264
class conf(parameters):
265
    def checkCorrectness(self):
266
        self.checkNeededParams()
267
        return True
268

    
269
if __name__ == '__main__':
270
    setLogLevel('info')
271
    need = [
272
            ("-f", ["graphDefinition", True, "", "path of the graph definition", str]),
273
            ("-t", ["testName", True, "", "base name for test output", str])
274
           ]
275
    opt = [
276
            ("-d", ["drawGraph", False, False, 
277
                "draw the graph before you run the test", int])
278
          ]
279

    
280
    P = conf(path.basename(__file__),need, opt)
281
    P.parseArgs()
282
    drawGraph = P.getParam("drawGraph")
283
    if P.checkCorrectness() == False:
284
        P.printUsage()
285
        sys.exit(1)
286
    net = GraphNet(P.getParam("graphDefinition"), draw = drawGraph)
287
    net.start()
288
    net.enableForwarding()
289
    net.setShortestRoutes()
290
#    CLI(net)
291
    test_name = P.getParam("testName")+str(int(time()))
292
    for i in range(1):
293
        info("+++++++ Round: "+str(i+1) + '\n')
294
        #test = PSRandomTest(net,duration=6,name=test_name,num_peers=2)
295
        test = PSHostsTest(net, 'h0_0', ['h1_1','h1_1','h2_2'],
296
                duration = 600, name = test_name)
297
        test.runTest()
298
      #  sleep(60)
299
    net.stop()
300
    info("*** Done with experiment: "+test_name+"\n")