BVhE ddZddlZddlmZddlmZddlmZddlm Z ddlm Z ddl m Z d e fd Ze d Gd d ejd gdZe dgGddZe dGddejdgdZe dgGddZy)z0A Python interface for creating dataset servers.N)Iterable)service_config_pb2)pywrap_tensorflow)_pywrap_server_lib)_pywrap_utils_exp) tf_exportreturnc|dk(ry|y|S)zCModifies time-based config values to account for special behaviors.r)values f/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/data/experimental/service/server_lib.py_get_time_or_placeholderrs aZ  ] ,z*data.experimental.service.DispatcherConfigc6eZdZdZ dfd ZxZS)DispatcherConfigaConfiguration class for tf.data service dispatchers. Fields: port: Specifies the port to bind to. A value of 0 indicates that the server may bind to any available port. protocol: The protocol to use for communicating with the tf.data service, e.g. "grpc". work_dir: A directory to store dispatcher state in. This argument is required for the dispatcher to be able to recover from restarts. fault_tolerant_mode: Whether the dispatcher should write its state to a journal so that it can recover from restarts. Dispatcher state, including registered datasets and created jobs, is synchronously written to the journal before responding to RPCs. If `True`, `work_dir` must also be specified. worker_addresses: If the job uses auto-sharding, it needs to specify a fixed list of worker addresses that will register with the dispatcher. The worker addresses should be in the format `"host"` or `"host:port"`, where `"port"` is an integer, named port, or `%port%` to match any port. job_gc_check_interval_ms: How often the dispatcher should scan through to delete old and unused jobs, in milliseconds. If not set, the runtime will select a reasonable default. A higher value will reduce load on the dispatcher, while a lower value will reduce the time it takes for the dispatcher to garbage collect expired jobs. job_gc_timeout_ms: How long a job needs to be unused before it becomes a candidate for garbage collection, in milliseconds. A value of -1 indicates that jobs should never be garbage collected. If not set, the runtime will select a reasonable default. A higher value will cause jobs to stay around longer with no consumers. This is useful if there is a large gap in time between when consumers read from the job. A lower value will reduce the time it takes to reclaim the resources from expired jobs. worker_timeout_ms: How long to wait for a worker to heartbeat before considering it missing. If not set, the runtime will select a reasonable default. worker_max_concurrent_snapshots: The maximum number of snapshots a worker can concurrently process. c |tj}t|}t|}t ||||||||||| SN)rTF_DATA_DefaultProtocolrsuper__new__) clsportprotocolwork_dirfault_tolerant_modeworker_addressesjob_gc_check_interval_msjob_gc_timeout_msworker_timeout_msworker_max_concurrent_snapshots __class__s rrzDispatcherConfig.__new__csb"::>> dispatcher = tf.data.experimental.service.DispatchServer() >>> dispatcher_address = dispatcher.target.split("://")[1] >>> worker = tf.data.experimental.service.WorkerServer( ... tf.data.experimental.service.WorkerConfig( ... dispatcher_address=dispatcher_address)) >>> dataset = tf.data.Dataset.range(10) >>> dataset = dataset.apply(tf.data.experimental.service.distribute( ... processing_mode="parallel_epochs", service=dispatcher.target)) >>> [a.item() for a in dataset.as_numpy_iterator()] [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] When starting a dedicated tf.data dispatch process, use join() to block after starting up the server, until the server terminates. ``` dispatcher = tf.data.experimental.service.DispatchServer( tf.data.experimental.service.DispatcherConfig(port=5050)) dispatcher.join() ``` Call stop() to gracefully terminate the dispatcher. The server automatically stops when all reference to it have been deleted. To start a `DispatchServer` in fault-tolerant mode, set `work_dir` and `fault_tolerant_mode` like below: ``` dispatcher = tf.data.experimental.service.DispatchServer( tf.data.experimental.service.DispatcherConfig( port=5050, work_dir="gs://my-bucket/dispatcher/work_dir", fault_tolerant_mode=True)) ``` Nc (|xs t}|jr|js td||_t |t jr|}nxt j|j|j|j|j|j|j|j|j|j }tj|j!|_|r|j"j%yy)aOCreates a new dispatch server. Args: config: (Optional.) A `tf.data.experimental.service.DispatcherConfig` configration. If `None`, the dispatcher will use default configuration values. start: (Optional.) Boolean, indicating whether to start the server after creating it. Defaults to True. zCannot enable fault tolerant mode without configuring a work dir. Make sure to set `work_dir` in the `config` object passed to `DispatcherServer`.r*N)rrr ValueError_config isinstancerrrrrrr r!rTF_DATA_NewDispatchServerSerializeToString_serverstartselfconfigr5 config_protos r__init__zDispatchServer.__init__s )')F !!&//   !!DL&,==>l'88{{????$88!22#)#B#B"44"44*0*P*P l&??&&(*DL  ll rc8|jjy)zStarts this server. >>> dispatcher = tf.data.experimental.service.DispatchServer(start=False) >>> dispatcher.start() Raises: tf.errors.OpError: Or one of its subclasses if an error occurs while starting the server. Nr4r5r7s rr5zDispatchServer.starts LLrr c8|jjy)aBlocks until the server has shut down. This is useful when starting a dedicated dispatch process. ``` dispatcher = tf.data.experimental.service.DispatchServer( tf.data.experimental.service.DispatcherConfig(port=5050)) dispatcher.join() ``` Raises: tf.errors.OpError: Or one of its subclasses if an error occurs while joining the server. Nr4joinr=s rr@zDispatchServer.joins LLrc$|jyzStops the server. Raises: tf.errors.OpError: Or one of its subclasses if an error occurs while stopping the server. N_stopr=s rstopzDispatchServer.stop  JJLrc~dj|jj|jj S)aReturns a target that can be used to connect to the server. >>> dispatcher = tf.data.experimental.service.DispatchServer() >>> dataset = tf.data.Dataset.range(10) >>> dataset = dataset.apply(tf.data.experimental.service.distribute( ... processing_mode="parallel_epochs", service=dispatcher.target)) The returned string will be in the form protocol://address, e.g. "grpc://localhost:5050". z{0}://localhost:{1})formatr0rr4 bound_portr=s rtargetzDispatchServer.targets5 ! ' ' (=(=(, (?(?(A CCrc8|jjyrBr4rEr=s rrDzDispatchServer._stop LLrc$|jyrrCr=s r__del__zDispatchServer.__del__ JJLrcTdj|jjSzyReturns the address of the server. The returned string will be in the form address:port, e.g. "localhost:1000". z localhost:{0}rHr4rIr=s r_addresszDispatchServer._address"  ! !$,,"9"9"; <>> dispatcher = tf.data.experimental.service.DispatchServer() >>> dispatcher_address = dispatcher.target.split("://")[1] >>> worker = tf.data.experimental.service.WorkerServer( ... tf.data.experimental.service.WorkerConfig( ... dispatcher_address=dispatcher_address)) >>> dataset = tf.data.Dataset.range(10) >>> dataset = dataset.apply(tf.data.experimental.service.distribute( ... processing_mode="parallel_epochs", service=dispatcher.target)) >>> [a.item() for a in dataset.as_numpy_iterator()] [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] When starting a dedicated tf.data worker process, use join() to block after starting up the worker, until the worker terminates. ``` worker = tf.data.experimental.service.WorkerServer( port=5051, dispatcher_address="localhost:5050") worker.join() ``` Call stop() to gracefully terminate the worker. The worker automatically stops when all reference to it have been deleted. c |j tdt|tjr|}nmtj|j|j |j |j|j|j|j|j}tj|j|_|r|jj!yy)zCreates a new worker server. Args: config: A `tf.data.experimental.service.WorkerConfig` configration. start: (Optional.) Boolean, indicating whether to start the server after creating it. Defaults to True. NzMMust specify a `dispatcher_address` in the `config` passed to `WorkerServer`.rk)rer/r1rrcrfrrrgrhrirjrTF_DATA_NewWorkerServerr3r4r5r6s rr:zWorkerServer.__init__}s  (    &,99:l'44#66..{{?? & < < & < !> & < <>l&==&&(*DL  ll rr Nc8|jjy)zStarts this server. Raises: tf.errors.OpError: Or one of its subclasses if an error occurs while starting the server. Nr<r=s rr5zWorkerServer.starts LLrc8|jjy)aBlocks until the server has shut down. This is useful when starting a dedicated worker process. ``` worker_server = tf.data.experimental.service.WorkerServer( port=5051, dispatcher_address="localhost:5050") worker_server.join() ``` This method currently blocks forever. Raises: tf.errors.OpError: Or one of its subclasses if an error occurs while joining the server. Nr?r=s rr@zWorkerServer.joins" LLrc$|jyrBrCr=s rrEzWorkerServer.stoprFrc8|jjyrBrLr=s rrDzWorkerServer._stoprMrc$|jyrrCr=s rrOzWorkerServer.__del__rPrcTdj|jjSrRrSr=s rrTzWorkerServer._addressrUrc6|jjS)zCReturns the number of tasks currently being executed on the worker.)r4 num_tasksr=s r _num_taskszWorkerServer._num_taskss << ! ! ##rc6|jjS)zReturns the progresses of the snapshot tasks currently being executed. Returns: An `Iterable[common_pb2.SnapshotTaskProgress]`. )r4snapshot_task_progressesr=s r_snapshot_task_progressesz&WorkerServer._snapshot_task_progressess << 0 0 22r)Tr])r$r%r&r'r:r5r@rErDrOr^r_rTr`rxrrSnapshotTaskProgressWrapperr{r rrrmrm\se>:& == =$#$3*FFG3rrm)r' collectionstypingrtensorflow.core.protobufrtensorflow.pythonr+tensorflow.python.data.experimental.servicerr tensorflow.python.util.tf_exportrr`r namedtuplerr-rcrmr rrrs78/JI6 s   78RK  R9Rj 5"=^/^/>^/B 3445K>,45545n 3;A3A3<A3r