BVhAM8dZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ddl m Zdd lmZdd lmZd Zed d gGddZeddgej*dee j,j.fdZeddgej*dedddde j,j.fdZe j4e j,j.ej6ej8ej:y)z+Create threads to run multiple enqueue ops.N)queue_runner_pb2)session)context)errors)ops) tf_logging) deprecation) tf_exportz7To construct input pipelines, use the `tf.data` module.ztrain.queue_runner.QueueRunnerztrain.QueueRunner)v1ceZdZdZej de ddZ ddZddZ e dZ e dZ e dZ e d Ze d Ze d Ze d Zdd ZdZddZddZeddZy) QueueRunneraHolds a list of enqueue operations for a queue, each to be run in a thread. Queues are a convenient TensorFlow mechanism to compute tensors asynchronously using multiple threads. For example in the canonical 'Input Reader' setup one set of threads generates filenames in a queue; a second set of threads read records from the files, processes them, and enqueues tensors on a second queue; a third set of threads dequeues these input records to construct batches and runs them through training operations. There are several delicate issues when running multiple threads that way: closing the queues in sequence as the input is exhausted, correctly catching and reporting exceptions, etc. The `QueueRunner`, combined with the `Coordinator`, helps handle these issues. @compatibility(TF2) QueueRunners are not compatible with eager execution. Instead, please use [tf.data](https://www.tensorflow.org/guide/data) to get data into your model. @end_compatibility Nc*tjr td|r#|s|r td|j ||n|j |||||t j|_tj|_ g|_ y)aCreate a QueueRunner. On construction the `QueueRunner` adds an op to close the queue. That op will be run if the enqueue ops raise exceptions. When you later call the `create_threads()` method, the `QueueRunner` will create one thread for each op in `enqueue_ops`. Each thread will run its enqueue op in parallel with the other threads. The enqueue ops do not have to all be the same op, but it is expected that they all enqueue tensors in `queue`. Args: queue: A `Queue`. enqueue_ops: List of enqueue ops to run in threads later. close_op: Op to close the queue. Pending enqueue ops are preserved. cancel_op: Op to close the queue and cancel pending enqueue ops. queue_closed_exception_types: Optional tuple of Exception types that indicate that the queue has been closed when raised during an enqueue operation. Defaults to `(tf.errors.OutOfRangeError,)`. Another common case includes `(tf.errors.OutOfRangeError, tf.errors.CancelledError)`, when some of the enqueue ops may dequeue from other Queues. queue_runner_def: Optional `QueueRunnerDef` protocol buffer. If specified, recreates the QueueRunner from its contents. `queue_runner_def` and the other arguments are mutually exclusive. import_scope: Optional `string`. Name scope to add. Only used when initializing from protocol buffer. Raises: ValueError: If both `queue_runner_def` and `queue` are both specified. ValueError: If `queue` or `enqueue_ops` are not provided when not restoring from `queue_runner_def`. RuntimeError: If eager execution is enabled. zxQueueRunners are not supported when eager execution is enabled. Instead, please use tf.data to get data into your model.z2queue_runner_def and queue are mutually exclusive.) import_scope)queue enqueue_opsclose_op cancel_opqueue_closed_exception_typesN) rexecuting_eagerly RuntimeError ValueError_init_from_proto_init_from_args threadingLock_lockweakrefWeakKeyDictionary_runs_per_session_exceptions_raised)selfrrrrrqueue_runner_defrs \/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/training/queue_runner_impl.py__init__zQueueRunner.__init__9sJ  "  E FF +MNN ,)57 ;y'CE !DJ%668D Dc|r|s td||_||_||_||_|2t |t r|rtd|Dstd|z||_ |j|jj|_|j!|jjd|_|jstjf|_ yt |j|_ y)aCreate a QueueRunner from arguments. Args: queue: A `Queue`. enqueue_ops: List of enqueue ops to run in threads later. close_op: Op to close the queue. Pending enqueue ops are preserved. cancel_op: Op to close the queue and cancel pending enqueue ops. queue_closed_exception_types: Tuple of exception types, which indicate the queue has been safely closed. Raises: ValueError: If `queue` or `enqueue_ops` are not provided when not restoring from `queue_runner_def`. TypeError: If `queue_closed_exception_types` is provided, but is not a non-empty tuple of error types (subclasses of `tf.errors.OpError`). z#Must provide queue and enqueue_ops.Nc3PK|]}t|tj ywN) issubclassrOpError).0ts r# z.QueueRunner._init_from_args..s#< 6>>2.s%/C  --d3/Cs!N)r3rQueueRunnerDefrget_default_graphas_graph_elementprepend_name_scope queue_namer/enqueue_op_namer0 close_op_namer1cancel_op_namer2r4rr7rr9)r!r"rgops r#rzQueueRunner._init_from_protos4 &(8(G(G HH H A$$ /::LIKDK$4#C#CE++ r<02ED''(>(>&& )67DN(()?)?''*78DO)./C$AA/C*CD&  - -,2,B,B+Dd( .Es3,Ec|jSr()r/r!s r#rzQueueRunner.queues ;;r%c|jSr()r0rIs r#rzQueueRunner.enqueue_opss   r%c|jSr()r1rIs r#rzQueueRunner.close_ops >>r%c|jSr()r2rIs r#rzQueueRunner.cancel_ops ??r%c|jSr()r7rIs r#rz(QueueRunner.queue_closed_exception_typess  - --r%c|jS)aExceptions raised but not handled by the `QueueRunner` threads. Exceptions raised in queue runner threads are handled in one of two ways depending on whether or not a `Coordinator` was passed to `create_threads()`: * With a `Coordinator`, exceptions are reported to the coordinator and forgotten by the `QueueRunner`. * Without a `Coordinator`, exceptions are captured by the `QueueRunner` and made available in this `exceptions_raised` property. Returns: A list of Python `Exception` objects. The list is empty if no exception was captured. (No exceptions are captured when using a Coordinator.) )r rIs r#exceptions_raisedzQueueRunner.exceptions_raiseds"  " ""r%c.|jjS)z(The string name of the underlying Queue.)r/namerIs r#rQzQueueRunner.names ;;  r%c Rd} |j|} |r|jrn | |s-|j5|j|xxdzcc<dddyy#|j$r|j5|j|xxdzcc<d}|j|dk(rS |j |j n6#t$r*}tjddt|Yd}~nd}~wwxYw dddY|s9|j5|j|xxdzcc<dddy#1swYyxYwy#1swYnxYwY!wxYw#t$r{}|r|j|n\tjdt||j5|jj|ddd#1swYxYwYd}~d}~wwxYw#1swYyxYw#|s9|j5|j|xxdzcc<dddw#1swYwxYwwxYw)zExecute the enqueue op in a loop, close the queue in case of error. Args: sess: A Session. enqueue_op: The Operation to run. coord: Optional Coordinator object for reporting errors and checking for stop conditions. FTrIgnored exception: %sNzException in QueueRunner: %s) make_callable should_stopr7rrrunr1 Exceptionloggingvlogstr request_stoperrorr append)r!sess enqueue_opcoord decrementedenqueue_callablees r#_runzQueueRunner._runsK",++J7  U&&(      4 ZZ ,   &! + & , ,/11 zz   " "4 (A - (K%%d+q0A(A Q 7Q@@A   * ZZ ,   &! + & , , ,+       1 4c!f= ZZ ,  ! ! ( ( + ,  ,  , , ZZ ,   &! + & , , ,s%EA#EG)G#E>,E+CE C: C50E5C::E>EED66D?E EEEE G?GG: GG GG)GG)G&)H&8H H&H" H&c|j |j|y#t$r*}tjddt |Yd}~yd}~wwxYw)zClose the queue when the Coordinator requests stop. Args: sess: A Session. cancel_op: The Operation to run. coord: Coordinator. rSrTN) wait_for_stoprWrXrYrZr[)r!r_rrards r#_close_on_stopzQueueRunner._close_on_stopsI 7 hhy 7 ll1-s1v667s$ A AAc |j5 |j|dkDr gcdddS t|j|j|<g|_dddg}|jD]\}dj |j|j}|jtj|j|||f|^|rYdj |j}|jtj|j||j|f||D]1}|r|j||rd|_|s"|j!3|S#t$rY=wxYw#1swYxYw)aCreate threads to run the enqueue ops for the given session. This method requires a session in which the graph was launched. It creates a list of threads, optionally starting them. There is one thread for each op passed in `enqueue_ops`. The `coord` argument is an optional coordinator that the threads will use to terminate together and report exceptions. If a coordinator is given, this method starts an additional thread to close the queue when the coordinator requests a stop. If previously created threads for the given session are still running, no new threads will be created. Args: sess: A `Session`. coord: Optional `Coordinator` object for reporting errors and checking stop conditions. daemon: Boolean. If `True` make the threads daemon threads. start: Boolean. If `True` starts the threads. If `False` the caller must call the `start()` method of the returned threads. Returns: A list of threads. rNzQueueRunnerThread-{}-{})targetargsrQz"QueueRunnerThread-{}-close_on_stopT)rrKeyErrorlenr0r formatrQr^rThreadrerhr2register_threaddaemonstart) r!r_rarqrr ret_threadsrGrQr,s r#create_threadszQueueRunner.create_threads*sr4  #  ! !$ '! + # # + &)):):%;dT" "d #K6 & - -dii Ad))04b%/@/3566  1 8 8 Cd))1D1D04doou/M/356  a     1   # #s-E.E*E. E+'E.*E++E..E8c|&|jjj|r3tj}t j |jj||_|jD];}|jjt j |j|=t j |jj||_ t j |jj||_|jj!|j"Dcgc]}t%j&|c}|Sycc}w)a Converts this `QueueRunner` to a `QueueRunnerDef` protocol buffer. Args: export_scope: Optional `string`. Name scope to remove. Returns: A `QueueRunnerDef` protocol buffer, or `None` if the `Variable` is not in the specified name scope. N)rrQ startswithrr>rstrip_name_scoperBrrCr^rrDrrErextendr7rerror_code_from_exception_type)r! export_scoper"r`clss r#to_protozQueueRunner.to_protocs#  ""<0)88:$'$8$8 **//<%)!((A*((//  , ? AA(+';'; --  l(,$(+(<(< ..  |)-%33::77<9  / / 4<9:  <9s7Ect||S)z?Returns a `QueueRunner` object created from `queue_runner_def`.r"r)r r~s r# from_protozQueueRunner.from_protos (8$0 22r%)NNNNNNN)NNNNNr()NFF)__name__ __module__ __qualname____doc__r deprecated_DEPRECATION_INSTRUCTIONr$rrpropertyrrrrrrOrQrerhrtr| staticmethodrr%r#r r !s,;$ 89<@<@379!:9!vDHCG..`E4         . . # #$   ,,\ 7 7r:22r%r z#train.queue_runner.add_queue_runnerztrain.add_queue_runnerc0tj||y)aAdds a `QueueRunner` to a collection in the graph. When building a complex model that uses many queues it is often difficult to gather all the queue runners that need to be run. This convenience function allows you to add a queue runner to a well known collection in the graph. The companion method `start_queue_runners()` can be used to start threads for all the collected queue runners. @compatibility(TF2) QueueRunners are not compatible with eager execution. Instead, please use [tf.data](https://www.tensorflow.org/guide/data) to get data into your model. @end_compatibility Args: qr: A `QueueRunner`. collection: A `GraphKey` specifying the graph collection to add the queue runner to. Defaults to `GraphKeys.QUEUE_RUNNERS`. N)radd_to_collection)qr collections r#add_queue_runnerrs. B'r%z&train.queue_runner.start_queue_runnersztrain.start_queue_runnersTc ttjr td|!tj}|s t dt |tjs>|jjdvrgStdj|jtj|}|stjd|j j#5g}tj|D]&}|j%|j'||||( ddd|S#1swYSxYw)aStarts all queue runners collected in the graph. This is a companion method to `add_queue_runner()`. It just starts threads for all queue runners collected in the graph. It returns the list of all threads. @compatibility(TF2) QueueRunners are not compatible with eager execution. Instead, please use [tf.data](https://www.tensorflow.org/guide/data) to get data into your model. @end_compatibility Args: sess: `Session` used to run the queue ops. Defaults to the default session. coord: Optional `Coordinator` for coordinating the started threads. daemon: Whether the threads should be marked as `daemons`, meaning they don't block program exit. start: Set to `False` to only create the threads, not start them. collection: A `GraphKey` specifying the graph collection to get the queue runners from. Defaults to `GraphKeys.QUEUE_RUNNERS`. Raises: ValueError: if `sess` is None and there isn't any default session. TypeError: if `sess` is not a `tf.compat.v1.Session` object. Returns: A list of threads. Raises: RuntimeError: If called with eager execution enabled. ValueError: If called without a default `tf.compat.v1.Session` registered. z/Queues are not compatible with eager execution.NzCannot start queue runners: No default session is registered. Use `with sess.as_default()` or pass an explicit session to tf.start_queue_runners(sess=sess))MonitoredSessionSingularMonitoredSessionz3sess must be a `tf.Session` object. Given class: {}z`tf.train.start_queue_runners()` was called when no queue runners were defined. You can safely remove the call to this deprecated function.)rarqrr)rrrrget_default_sessionrr3rSessionInterface __class__rr6rnget_collectionrYwarninggraph as_defaultrxrt)r_rarqrrr queue_runnersthreadsrs r#start_queue_runnersrs2L  H II \  " " $D  O PP D'22 3 ~~#88 i &&,fT^^&< >>$$Z0-  OO   zz5G  ,5 nnR&&t5-2'4555 . 5 .s !AD--D7) proto_typer|r)rrrtensorflow.core.protobufrtensorflow.python.clientrtensorflow.python.eagerrtensorflow.python.frameworkrrtensorflow.python.platformrrYtensorflow.python.utilr tensorflow.python.util.tf_exportr rr r GraphKeys QUEUE_RUNNERSrrregister_proto_functionr>r|rrr%r#rs0 25,+.+<.6> /1DEFb2b2Gb2J  46NOP67$'MM$?$?(8Q(0 7*,-67!d$#&==#>#>@8-@FCMM77'7'F'F%0%9%9'2'='=?r%