(name, group, final_outputs, reader, num_threads,
output, capacity)
| 247 | |
| 248 | |
| 249 | def _static_threads_task(name, group, final_outputs, reader, num_threads, |
| 250 | output, capacity): |
| 251 | node_name = str(Node.current()) |
| 252 | profiler_name = "{0}/{1}/{2}/{3}/{4}".format( |
| 253 | node_name, |
| 254 | "pipe", |
| 255 | name, |
| 256 | processor_name(input) if input else "NoInput", |
| 257 | processor_name(output) if output else "NoOutput") |
| 258 | |
| 259 | with Task(name=name, group=group, outputs=final_outputs) as task: |
| 260 | global_exit_net = core.Net('exit') |
| 261 | global_init_net = core.Net('init') |
| 262 | reader.setup_ex(global_init_net, global_exit_net) |
| 263 | |
| 264 | out_queue = None |
| 265 | writer = None |
| 266 | |
| 267 | steps = [] |
| 268 | for thread_id in range(num_threads): |
| 269 | with NetBuilder(name='t:%d' % thread_id) as nb: |
| 270 | init_net = core.Net('init') |
| 271 | exit_net = core.Net('exit') |
| 272 | read_nets, status, rec = reader.read_record_ex( |
| 273 | init_net, exit_net) |
| 274 | init_net.ConstantFill( |
| 275 | [], [status], |
| 276 | shape=[], |
| 277 | value=False, |
| 278 | dtype=core.DataType.BOOL |
| 279 | ) |
| 280 | |
| 281 | if rec is not None: |
| 282 | if writer is None: |
| 283 | # hack so that the out queue gets the right name prefix |
| 284 | # (otherwise they would be prefixed with the thread id) |
| 285 | with NetBuilder(_fullname=task.name): |
| 286 | out_queue, writer = _init_output( |
| 287 | output, capacity, global_init_net, |
| 288 | global_exit_net) |
| 289 | write_nets, _ = writer.write_record_ex( |
| 290 | rec, init_net, exit_net, status) |
| 291 | else: |
| 292 | write_nets = [] |
| 293 | |
| 294 | timer_start_net = core.Net('timer_start') |
| 295 | timer = timer_start_net.TimerBegin([], counter_name=profiler_name) |
| 296 | timer_end_net = core.Net('timer_end') |
| 297 | timer_end_net.TimerEnd(timer, []) |
| 298 | |
| 299 | ops.net(init_net) |
| 300 | ops.net(core.execution_step( |
| 301 | 'body', |
| 302 | [timer_start_net] + list(read_nets) + list(write_nets) + |
| 303 | [timer_end_net], |
| 304 | should_stop_blob=status)) |
| 305 | ops.net(timer_end_net) |
| 306 | ops.net(exit_net) |
no test coverage detected
searching dependent graphs…