3 from __future__
import absolute_import
4 from __future__
import division
5 from __future__
import print_function
6 from __future__
import unicode_literals
17 Represents the result of a processor function. A processor can either 18 return an Output, or it can return a record, in which case an Output will be 19 created for it afterwards. 21 def __init__(self, nets=None, record=None, should_stop=None):
22 builder_children = NetBuilder.current().get()
23 assert nets
is None or len(builder_children) == 0, (
24 'Cannot both use `ops` syntax and return a list of nets.')
26 nets = builder_children
29 self.
nets = []
if nets
is None else list(nets)
30 self.
record =
None if record
is None else as_record(record)
34 DEFAULT_QUEUE_CAPACITY = 100
37 def _init_output(output, capacity, global_init_net, global_exit_net):
41 capacity
if capacity
is not None 42 else DEFAULT_QUEUE_CAPACITY))
43 writer = out_queue.writer()
44 elif isinstance(output, Writer):
45 assert capacity
is None,
'capacity would not be used.' 48 elif hasattr(output,
'writer'):
49 assert capacity
is None,
'capacity would not be used.' 51 writer = output.writer()
53 raise ValueError(
'output must be a reader, queue or stream.')
54 writer.setup_ex(global_init_net, global_exit_net)
55 return out_queue, writer
58 def make_processor(processor):
60 return lambda rec: rec
61 elif isinstance(processor,
core.Net):
67 def normalize_processor_output(output):
69 Allow for processors to return results in several formats. 70 TODO(azzolini): simplify once all processors use NetBuilder API. 72 if isinstance(output, Output):
73 """ Processor returned an Output. """ 75 elif isinstance(output, Field):
76 """ Processor returned a record. """ 77 return Output(record=output)
78 elif isinstance(output, tuple):
79 is_record_and_blob = (
81 isinstance(output[0], Field)
and 83 if is_record_and_blob:
84 """ Processor returned (record, stop_blob) """ 85 return Output(
None, *output)
87 """ Processor returned (nets, record, stop_blob) """ 90 """ Processor returned nets, no output """ 95 input, output=
None, num_threads=1, processor=
None, name=
None,
96 capacity=
None, group=
None, num_runtime_threads=1):
98 Given a Reader, Queue or DataStream in `input`, and optionally, a Writer, 99 Queue or DataStream in `output`, creates a Task that, when run, will 100 pipe the input into the output, using multiple parallel threads. 101 Additionally, if a processor is given, it will be called between reading 102 and writing steps, allowing it to transform the record. 105 input: either a Reader, Queue or DataStream that will be read 106 until a stop is signaled either by the reader or the 108 output: either a Writer, a Queue or a DataStream that will be 109 writen to as long as neither reader nor writer signal 110 a stop condition. If output is not provided or is None, 111 a Queue is created with given `capacity` and writen to. 112 num_threads: number of concurrent threads used for processing and 113 piping. If set to 0, no Task is created, and a 114 reader is returned instead -- the reader returned will 115 read from the reader passed in and process it. 116 ** DEPRECATED **. Use `num_runtime_threads` instead. 117 This option will be removed once all readers/processors 118 support `num_runtime_threads`. 119 processor: (optional) function that takes an input record and 120 optionally returns a record; this will be called 121 between read and write steps. If the processor does 122 not return a record, a writer will not be instantiated. 123 Processor can also be a core.Net with input and output 124 records properly set. In that case, a NetProcessor is 125 instantiated, cloning the net for each of the threads. 126 name: (optional) name of the task to be created. 127 capacity: when output is not passed, a queue of given `capacity` 128 is created and written to. 129 group: (optional) explicitly add the created Task to this 130 TaskGroup, instead of using the currently active one. 131 num_runtime_threads: Similar to `num_threads`, but instead of expanding 132 the tasks with a `for` loop in python, does that at 133 runtime. This is preferable to `num_threads`, but some 134 processors/readers still require to be called multiple 138 Output Queue, DataStream, Reader, or None, depending on the parameters 141 result, _ = _pipe_step(
142 input, output, num_threads, processor, name, capacity, group,
148 input, output=
None, num_threads=1, processor=
None, name=
None,
149 capacity=
None, group=
None, num_runtime_threads=1, final_outputs=
None):
151 Similar to `pipe`, with the additional ability for the pipe Task to 152 return output values to the `Session` once done. 155 Tuple (out_queue, *task_outputs) 156 out_queue: same as return value of `pipe`. 157 task_outputs: TaskOutput object, fetchable from the client after 158 session.run() returns. 160 assert num_threads > 0
161 result, task = _pipe_step(
162 input, output, num_threads, processor, name, capacity, group,
163 num_runtime_threads, final_outputs)
165 if final_outputs
is not None:
166 output = task.outputs()
167 if type(final_outputs)
not in (list, tuple):
169 return result, output
172 def processor_name(processor):
173 if hasattr(processor,
'name'):
174 return processor.name
175 if hasattr(processor,
'func_name'):
176 if processor.func_name ==
'<lambda>':
177 return processor.__module__
178 if hasattr(processor,
'im_class'):
179 return '%s.%s' % (processor.im_class.__name__, processor.func_name)
180 return processor.func_name
181 return processor.__class__.__name__
184 def _runtime_threads_task(name, group, final_outputs, reader, num_threads,
186 node_name = str(Node.current())
187 profiler_name =
"{0}/{1}/{2}/{3}/{4}".format(
191 processor_name(input)
if input
else "NoInput",
192 processor_name(output)
if output
else "NoOutput")
194 with
Task(name=name, group=group, outputs=final_outputs,
195 num_instances=num_threads)
as task:
196 global_exit_net =
core.Net(
'pipe:exit')
197 global_init_net =
core.Net(
'pipe:init')
198 reader.setup_ex(global_init_net, global_exit_net)
200 init_net =
core.Net(
'pipe:instance:init')
201 exit_net =
core.Net(
'pipe:instance:exit')
202 read_nets, status, rec = reader.read_record_ex(init_net, exit_net)
203 init_net.ConstantFill(
207 dtype=core.DataType.BOOL
211 out_queue, writer = _init_output(
212 output, capacity, global_init_net, global_exit_net)
213 write_nets, _ = writer.write_record_ex(
214 rec, init_net, exit_net, status)
219 with ops.task_init():
220 ops.net(global_init_net)
221 with ops.task_instance_init():
224 timer_start_net =
core.Net(
'timer_start')
225 timer = timer_start_net.TimerBegin([], counter_name=profiler_name)
226 timer_end_net =
core.Net(
'timer_end')
227 timer_end_net.TimerEnd(timer, [])
229 ops.net(core.execution_step(
231 [timer_start_net] + list(read_nets) + list(write_nets) +
233 should_stop_blob=status))
234 ops.net(timer_end_net)
236 with ops.task_instance_exit():
238 with ops.task_exit():
239 ops.net(global_exit_net)
241 return out_queue, task
244 def _static_threads_task(name, group, final_outputs, reader, num_threads,
246 node_name = str(Node.current())
247 profiler_name =
"{0}/{1}/{2}/{3}/{4}".format(
251 processor_name(input)
if input
else "NoInput",
252 processor_name(output)
if output
else "NoOutput")
254 with
Task(name=name, group=group, outputs=final_outputs)
as task:
257 reader.setup_ex(global_init_net, global_exit_net)
263 for thread_id
in range(num_threads):
264 with
NetBuilder(name=
't:%d' % thread_id)
as nb:
267 read_nets, status, rec = reader.read_record_ex(
269 init_net.ConstantFill(
273 dtype=core.DataType.BOOL
281 out_queue, writer = _init_output(
282 output, capacity, global_init_net,
284 write_nets, _ = writer.write_record_ex(
285 rec, init_net, exit_net, status)
289 timer_start_net =
core.Net(
'timer_start')
290 timer = timer_start_net.TimerBegin([], counter_name=profiler_name)
291 timer_end_net =
core.Net(
'timer_end')
292 timer_end_net.TimerEnd(timer, [])
295 ops.net(core.execution_step(
297 [timer_start_net] + list(read_nets) + list(write_nets) +
299 should_stop_blob=status))
300 ops.net(timer_end_net)
302 steps.append(core.to_execution_step(nb))
303 ops.net(global_init_net)
304 ops.net(core.execution_step(
'body', steps, concurrent_substeps=
True))
305 ops.net(global_exit_net)
306 return out_queue, task
310 input, output=
None, num_threads=1, processor=
None, name=
None,
311 capacity=
None, group=
None, num_runtime_threads=
None, final_outputs=
None):
314 assert num_threads <= 1
or num_runtime_threads <= 1, (
315 'Only one of num_threads or num_runtime_threads must be set.')
317 if isinstance(input, Reader):
319 elif hasattr(input,
'reader'):
320 reader = input.reader()
322 raise ValueError(
'in must be a reader, queue or stream.')
324 if processor
is not None:
327 if num_threads == 0
or num_runtime_threads == 0:
328 assert output
is None 331 if name
is None and processor
is not None:
332 name = processor_name(processor)
333 if name
is None and output
is not None:
334 name =
'pipe_into:%s' % processor_name(output)
336 name =
'pipe_from:%s' % processor_name(input)
339 return _static_threads_task(
340 name, group, final_outputs, reader, num_threads, output, capacity)
342 return _runtime_threads_task(
343 name, group, final_outputs, reader, num_runtime_threads, output,
349 Reader that reads from an upstream reader, calls the processor, and returns 350 the processed record. 352 def __init__(self, reader, processor):
353 Reader.__init__(self)
355 self.
processor = make_processor(processor)
357 def setup_ex(self, init_net, finish_net):
358 self.reader.setup_ex(init_net, finish_net)
360 def read_ex(self, init_net, exit_net):
361 read_nets, status, rec = self.reader.read_record_ex(init_net, exit_net)
372 result = normalize_processor_output(self.
processor(rec))
373 read_nets += result.nets
374 if result.should_stop
or nb._stop_blob:
376 if result.should_stop:
377 stop_net.Or([status, result.should_stop], [status])
379 stop_net.Or([status, nb._stop_blob], [status])
380 read_nets.append(stop_net)
382 init_net.add_attribute(TaskGroup.LOCAL_SETUP, self.
processor)
384 fields = result.record.field_blobs()
if result.record
else None 385 return read_nets, status, fields
390 Processor that clones a core.Net each time it's called, executing 391 the cloned net as the processor. It requires the Net to have input 392 and (optionally) output records set, with net.set_input_record() and 393 net.set_output_record(). 395 def __init__(self, net, stop_signal=None, thread_init_nets=None, name=None):
397 assert stop_signal
is None or isinstance(
399 self.
name = name
or str(net)
407 def setup(self, init_net):
411 return cloned_init_nets
413 def __call__(self, rec):
415 prefix = NetBuilder.current().name +
'/' 418 new_net, _ = core.clone_and_bind_net(
419 net, str(net) + prefix, prefix, blob_remap)
420 self._cloned_init_nets.append(new_net)
422 new_net, remappings = core.clone_and_bind_net(
423 self.
net, str(self.
net) + prefix, prefix, blob_remap, rec)
434 self._blob_maps.append(remappings)
435 return Output([new_net], new_net.output_record(), stop_signal)
def _set_schema(self, schema)