Runs a pipeline that injects data (input_pkts) at module src_module on input gate igate and collects it at module dst_module on output gates ogates. Runs the pipeline for at most time_out seconds; stops sooner if and when all the input packets are processed.
(self, src_module, dst_module, igate, input_pkts,
ogates, time_out=3, proto=scapy.Ether)
| 237 | self.bess.pause_all() |
| 238 | |
| 239 | def run_pipeline(self, src_module, dst_module, igate, input_pkts, |
| 240 | ogates, time_out=3, proto=scapy.Ether): |
| 241 | """ |
| 242 | Runs a pipeline that injects data (input_pkts) at module |
| 243 | src_module on input gate igate and collects it at module |
| 244 | dst_module on output gates ogates. Runs the pipeline for |
| 245 | at most time_out seconds; stops sooner if and when all the |
| 246 | input packets are processed. |
| 247 | |
| 248 | input_pkts may be a single bytestring, or a list of them. |
| 249 | """ |
| 250 | if not isinstance(input_pkts, list): |
| 251 | input_pkts = [input_pkts] |
| 252 | |
| 253 | self.bess.pause_all() |
| 254 | |
| 255 | # output ports and associate sockets |
| 256 | for ogate in ogates: |
| 257 | if ogate not in self.output_ports: |
| 258 | sock_name = "soc_{}_{}".format(ogate, |
| 259 | SCRIPT_STARTTIME) |
| 260 | |
| 261 | if ogate not in self.sockets: |
| 262 | sock = gen_unix_socket(self.bess, sock_name) |
| 263 | self.sockets[ogate] = sock |
| 264 | |
| 265 | po = self.bess.create_module('PortOut', 'po%d' % ogate, |
| 266 | {'port': sock_name}) |
| 267 | self.output_ports[ogate] = po |
| 268 | |
| 269 | po = self.output_ports[ogate] |
| 270 | self.bess.disconnect_modules(dst_module.name, ogate) |
| 271 | self.bess.connect_modules(dst_module.name, po.name, ogate, 0) |
| 272 | |
| 273 | # input ports and associate sockets |
| 274 | if igate not in self.input_ports: |
| 275 | sock_name = "soc_{}_{}".format(igate, |
| 276 | SCRIPT_STARTTIME) |
| 277 | |
| 278 | if igate not in self.sockets: |
| 279 | sock = gen_unix_socket(self.bess, sock_name) |
| 280 | self.sockets[igate] = sock |
| 281 | |
| 282 | pi = self.bess.create_module('PortInc', 'pi%d' % igate, |
| 283 | {'port': sock_name}) |
| 284 | self.input_ports[igate] = pi |
| 285 | |
| 286 | pi = self.input_ports[igate] |
| 287 | self.bess.disconnect_modules(pi.name) |
| 288 | self.bess.connect_modules(pi.name, src_module.name, 0, igate) |
| 289 | |
| 290 | self.bess.resume_all() # attach all orphan tasks into root |
| 291 | self.bess.pause_all() |
| 292 | |
| 293 | root_tc = get_root_tc(self.bess) |
| 294 | if not root_tc: |
| 295 | raise Exception('Fail to find root tc') |
| 296 |
no test coverage detected