BVh]8dZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddl m Z ddl mZdd Zd Zed ej"ej$Gd d eZedGddeZedGddeZy)zFCluster Resolvers are used for dynamic cluster IP/hostname resolution.N)session)contextconfig)ops) ClusterSpec) tf_exportc|r|d|S|S)Nz://)master rpc_layers n/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/cluster_resolver/cluster_resolver.pyformat_master_urlrs!6 ** Mc tjrytj}g}|D]\}|jdk(s|jdk(r"|j t j|j|jdd^|Stjj5t j||5}|j}ddddddS#1swYxYw#1swYSxYw)z?Returns accelerator devices given a master and a configuration.CPUXLA_CPUrrN)rexecuting_eagerlyrlist_logical_devices device_typeappendr_DeviceAttributesnamerGraph as_defaultSession list_devices)r config_protological_devicesdevicesdss rget_accelerator_devicesr#&s  113OG M % 1==I#= nnW..qvvq}}aKLM N    !# ??6, 7#1.."## N### Ns$0C7C+C7+C4 0C77Dz+distribute.cluster_resolver.ClusterResolverceZdZdZej dZej d dZ d dZe dZ e dZ e dZ e jd Z e jd Z y) ClusterResolveraAbstract class for all implementations of ClusterResolvers. This defines the skeleton for all implementations of ClusterResolvers. ClusterResolvers are a way for TensorFlow to communicate with various cluster management systems (e.g. GCE, AWS, etc...) and gives TensorFlow necessary information to set up distributed training. By letting TensorFlow communicate with these systems, we will be able to automatically discover and resolve IP addresses for various TensorFlow workers. This will eventually allow us to automatically recover from underlying machine failures and scale TensorFlow worker clusters up and down. Note to Implementors of `tf.distribute.cluster_resolver.ClusterResolver` subclass: In addition to these abstract methods, when task_type, task_id, and rpc_layer attributes are applicable, you should also implement them either as properties with getters or setters, or directly set the attributes `self._task_type`, `self._task_id`, or `self._rpc_layer` so the base class' getters and setters are used. See `tf.distribute.cluster_resolver.SimpleClusterResolver.__init__` for an example. In general, multi-client tf.distribute strategies such as `tf.distribute.experimental.MultiWorkerMirroredStrategy` require task_type and task_id properties to be available in the `ClusterResolver` they are using. On the other hand, these concepts are not applicable in single-client strategies, such as `tf.distribute.experimental.TPUStrategy`, because the program is only expected to be run on one task, so there should not be a need to have code branches according to task type and task id. - task_type is the name of the server's current named job (e.g. 'worker', 'ps' in a distributed parameterized training job). - task_id is the ordinal index of the server within the task type. - rpc_layer is the protocol used by TensorFlow to communicate with other TensorFlow servers in a distributed environment. ct)a@Retrieve the current state of the cluster and return a `tf.train.ClusterSpec`. Returns: A `tf.train.ClusterSpec` representing the state of the cluster at the moment this function is called. Implementors of this function must take care in ensuring that the ClusterSpec returned is up-to-date at the time of calling this function. This usually means retrieving the information from the underlying cluster management system every time this function is invoked and reconstructing a cluster_spec, rather than attempting to cache anything. NotImplementedErrorselfs r cluster_speczClusterResolver.cluster_spec^s   rNct)ahRetrieves the name or URL of the session master. Note: this is only useful for TensorFlow 1.x. Args: task_type: (Optional) The type of the TensorFlow task of the master. task_id: (Optional) The index of the TensorFlow task of the master. rpc_layer: (Optional) The RPC protocol for the given cluster. Returns: The name or URL of the session master. Implementors of this function must take care in ensuring that the master returned is up-to-date at the time to calling this function. This usually means retrieving the master every time this function is invoked. r')r* task_typetask_idr s rr zClusterResolver.masterns $  rc|j||}t||}tjt}|D]D}|)|'d|z}d|z} ||j vs| |j vr.||j xxdz cc<F|S)aKReturns the number of accelerator cores per worker. This returns the number of accelerator cores (such as GPUs and TPUs) available per worker. Optionally, we allow callers to specify the task_type, and task_id, for if they want to target a specific TensorFlow task to query the number of accelerators. This is to support heterogenous environments, where the number of accelerators cores per host is different. Args: task_type: (Optional) The type of the TensorFlow task of the machine we want to query. task_id: (Optional) The index of the TensorFlow task of the machine we want to query. config_proto: (Optional) Configuration for starting a new session to query how many accelerator cores it has. Returns: A map of accelerator types to number of cores. z/job:%sz/task:%s)r r# collections defaultdictintrr) r*r-r.rr r mappingdevicejob_path task_paths rnum_acceleratorsz ClusterResolver.num_acceleratorss2[[G ,F&fl;G%%c*G'  7#6y(( 6;; &)6;;*F  f  !Q&! ' Nrcy)aReturns the current environment which TensorFlow is running in. There are two possible return values, "google" (when TensorFlow is running in a Google-internal environment) or an empty string (when TensorFlow is running elsewhere). If you are implementing a ClusterResolver that works in both the Google environment and the open-source world (for instance, a TPU ClusterResolver or similar), you will have to return the appropriate string depending on the environment, which you will have to detect. Otherwise, if you are implementing a ClusterResolver that will only work in open-source TensorFlow, you do not need to implement this property. r r)s r environmentzClusterResolver.environments rct|ddS)aReturns the task type this `ClusterResolver` indicates. In TensorFlow distributed environment, each job may have an applicable task type. Valid task types in TensorFlow include 'chief': a worker that is designated with more responsibility, 'worker': a regular worker for training/evaluation, 'ps': a parameter server, or 'evaluator': an evaluator that evaluates the checkpoints for metrics. See [Multi-worker configuration]( https://www.tensorflow.org/tutorials/distribute/multi_worker_with_keras#multi-worker_configuration) for more information about 'chief' and 'worker' task type, which are most commonly used. Having access to such information is useful when user needs to run specific code according to task types. For example, ```python cluster_spec = tf.train.ClusterSpec({ "ps": ["localhost:2222", "localhost:2223"], "worker": ["localhost:2224", "localhost:2225", "localhost:2226"] }) # SimpleClusterResolver is used here for illustration; other cluster # resolvers may be used for other source of task type/id. simple_resolver = SimpleClusterResolver(cluster_spec, task_type="worker", task_id=1) ... if cluster_resolver.task_type == 'worker': # Perform something that's only applicable on workers. This block # will run on this particular instance since we've specified this task to # be a worker in above cluster resolver. elif cluster_resolver.task_type == 'ps': # Perform something that's only applicable on parameter servers. This # block will not run on this particular instance. ``` Returns `None` if such information is not available or is not applicable in the current distributed environment, such as training with `tf.distribute.experimental.TPUStrategy`. For more information, please see `tf.distribute.cluster_resolver.ClusterResolver`'s class doc. _task_typeNgetattrr)s rr-zClusterResolver.task_types` 4t ,,rct|ddS)aReturns the task id this `ClusterResolver` indicates. In TensorFlow distributed environment, each job may have an applicable task id, which is the index of the instance within its task type. This is useful when user needs to run specific code according to task index. For example, ```python cluster_spec = tf.train.ClusterSpec({ "ps": ["localhost:2222", "localhost:2223"], "worker": ["localhost:2224", "localhost:2225", "localhost:2226"] }) # SimpleClusterResolver is used here for illustration; other cluster # resolvers may be used for other source of task type/id. simple_resolver = SimpleClusterResolver(cluster_spec, task_type="worker", task_id=0) ... if cluster_resolver.task_type == 'worker' and cluster_resolver.task_id == 0: # Perform something that's only applicable on 'worker' type, id 0. This # block will run on this particular instance since we've specified this # task to be a 'worker', id 0 in above cluster resolver. else: # Perform something that's only applicable on other ids. This block will # not run on this particular instance. ``` Returns `None` if such information is not available or is not applicable in the current distributed environment, such as training with `tf.distribute.cluster_resolver.TPUClusterResolver`. For more information, please see `tf.distribute.cluster_resolver.ClusterResolver`'s class docstring. _task_idNr>r)s rr.zClusterResolver.task_idsL 4T **rc||_y)z=Setter of `task_type` property. See `task_type` property doc.Nr=r*r-s rr-zClusterResolver.task_types  DOrc||_y)z;Setter of `task_id` property. See `task_type` property doc.NrAr*r.s rr.zClusterResolver.task_ids DMrNNN)__name__ __module__ __qualname____doc__abcabstractmethodr+r r8propertyr;r-r.setterr rrr%r%7s"H      ("&#$(%N  " /- /-b %+ %+N    >>rr%z1distribute.cluster_resolver.SimpleClusterResolverceZdZdZ d fd ZdZddZedZedZ ejdZe jdZ ed Z dd Z ed Z e jd Z xZS)SimpleClusterResolveraSimple implementation of ClusterResolver that accepts all attributes. Please see the base class for documentation of arguments of its constructor. It is useful if you want to specify some or all attributes. Usage example with `tf.distribute.Strategy`: ```Python cluster = tf.train.ClusterSpec({"worker": ["worker0.example.com:2222", "worker1.example.com:2222"]}) # On worker 0 cluster_resolver = SimpleClusterResolver(cluster, task_type="worker", task_id=0, num_accelerators={"GPU": 8}, rpc_layer="grpc") strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy( cluster_resolver=cluster_resolver) # On worker 1 cluster_resolver = SimpleClusterResolver(cluster, task_type="worker", task_id=1, num_accelerators={"GPU": 8}, rpc_layer="grpc") strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy( cluster_resolver=cluster_resolver) ``` ctt| ||_||_||_||_||_t|ts td||_ t|ts td||_ y)z3Creates a SimpleClusterResolver from a ClusterSpec.z.cluster_spec must be a `tf.train.ClusterSpec`.zmaster must be a string.N)superrR__init__r=rA _environment_num_accelerators _rpc_layer isinstancer TypeError _cluster_specstr_master) r*r+r r-r.r;r8r __class__s rrUzSimpleClusterResolver.__init__@ss /1DODM#D-DDO lK 0 F GG%D fc " 0 11DLrc|jS)z4Returns the ClusterSpec passed into the constructor.)r[r)s rr+z"SimpleClusterResolver.cluster_specUs   rc|#|!|jj||}n |j}t||xs |jS)aReturns the master address to use when creating a session. Note: this is only useful for TensorFlow 1.x. Args: task_type: (Optional) The type of the TensorFlow task of the master. task_id: (Optional) The index of the TensorFlow task of the master. rpc_layer: (Optional) The RPC used by distributed TensorFlow. Returns: The name or URL of the session master. If a task_type and task_id is given, this will override the `master` string passed into the initialization function. r )r+ task_addressr]rrXr*r-r.r r s rr zSimpleClusterResolver.masterYsI !4  "// 7Cf||f Vy/KDOO LLrc|jSNrCr)s rr-zSimpleClusterResolver.task_typep ??rc|jSrerFr)s rr.zSimpleClusterResolver.task_idts ==rc||_yrerCrDs rr-zSimpleClusterResolver.task_typex DOrc||_yrerFrGs rr.zSimpleClusterResolver.task_id| DMrc|jSre)rVr)s rr;z!SimpleClusterResolver.environments   rc<~~~|jiS|jS)aEReturns the number of accelerator cores per worker. The SimpleClusterResolver does not do automatic detection of accelerators, and thus all arguments are unused and we simply return the value provided in the constructor. Args: task_type: Unused. task_id: Unused. config_proto: Unused. )rWr*r-r.rs rr8z&SimpleClusterResolver.num_acceleratorss( 7L % i  ! !!rc|jSrerXr)s rr zSimpleClusterResolver.rpc_layerrfrc||_yrerpr*r s rr zSimpleClusterResolver.rpc_layerrir)r:NNr:NNrHrIrJrKrLrUr+r rOr-r.rPr;r8r __classcell__r^s@rrRrR s<GK04*M.        >>  "&#$("*     rrRz)distribute.cluster_resolver.UnionResolverceZdZdZfdZdZd dZedZedZ ejdZe jdZ ed Z d d Z ed Z e jd Z xZS)UnionClusterResolveraPerforms a union on underlying ClusterResolvers. This class performs a union given two or more existing ClusterResolvers. It merges the underlying ClusterResolvers, and returns one unified ClusterSpec when cluster_spec is called. The details of the merge function is documented in the cluster_spec function. For additional ClusterResolver properties such as task type, task index, rpc layer, environment, etc..., we will return the value from the first ClusterResolver in the union. An example to combine two cluster resolvers: ```Python cluster_0 = tf.train.ClusterSpec({"worker": ["worker0.example.com:2222", "worker1.example.com:2222"]}) cluster_resolver_0 = SimpleClusterResolver(cluster, task_type="worker", task_id=0, rpc_layer="grpc") cluster_1 = tf.train.ClusterSpec({"ps": ["ps0.example.com:2222", "ps1.example.com:2222"]}) cluster_resolver_1 = SimpleClusterResolver(cluster, task_type="ps", task_id=0, rpc_layer="grpc") # Its task type would be "worker". cluster_resolver = UnionClusterResolver(cluster_resolver_0, cluster_resolver_1) ``` An example to override the number of GPUs in a TFConfigClusterResolver instance: ```Python tf_config = TFConfigClusterResolver() gpu_override = SimpleClusterResolver(tf_config.cluster_spec(), num_accelerators={"GPU": 1}) cluster_resolver = UnionResolver(gpu_override, tf_config) ``` cXtt| |jdd|_|jdd|_|jdd|_|rtdj||s td|D]}t|trtd||_ y)aInitializes a UnionClusterResolver with other ClusterResolvers. Args: *args: `ClusterResolver` objects to be unionized. **kwargs: rpc_layer - (Optional) Override value for the RPC layer used by TensorFlow. task_type - (Optional) Override value for the current task type. task_id - (Optional) Override value for the current task index. Raises: TypeError: If any argument is not a subclass of `ClusterResolvers`. ValueError: If there are no arguments passed. r Nr-r.zUnexpected kwargs provided {!r}z)At least one ClusterResolver is required.z7All arguments must be a sub-class of `ClusterResolver.`) rTrwrUpoprXr=rA ValueErrorformatrYr%rZ_cluster_resolvers)r*argskwargscluster_resolverr^s rrUzUnionClusterResolver.__init__s .0jjd3DOjjd3DOJJy$/DM 8??G HH  B CC . (/ :-. ..#Drc \i}|jD]p}|j}|j}|jD];\}}||vrt |t si||<!t |t rg||<7i||<=r|jD]}|j}|j}|jD]\}}t ||t r||j|.t |t r)t ttdt||}n|j}t|}t||j} |j| } | rtdt!| z||j#|t%|S)aReturns a union of all the ClusterSpecs from the ClusterResolvers. Returns: A ClusterSpec containing host information merged from all the underlying ClusterResolvers. Raises: KeyError: If there are conflicting keys detected when merging two or more dictionaries, this exception is raised. Note: If there are multiple ClusterResolvers exposing ClusterSpecs with the same job name, we will merge the list/dict of workers. If *all* underlying ClusterSpecs expose the set of workers as lists, we will concatenate the lists of workers, starting with the list of workers from the first ClusterResolver passed into the constructor. If *any* of the ClusterSpecs expose the set of workers as a dict, we will treat all the sets of workers as dicts (even if they are returned as lists) and will only merge them into a dict if there is no conflicting keys. If there is a conflicting key, we will raise a `KeyError`. rz9Duplicate keys detected when merging two ClusterSpecs: %s)r|r+as_dictitemsrYdictlistextendziprangelencopysetkeys intersectionKeyErrorreprupdater) r*merged_clusterrr+ cluster_dictjob_nametasks task_dict task_keys merged_keysintersected_keyss rr+z!UnionClusterResolver.cluster_specs0N!33*%224l!))+l)//1 */(E ~ % t $')N8 $t $')N8 $')N8 $ * *"!335%224l!))+l)//15/(E nX. 5  " ) )% 0 t $Sq#e*!5u=>I I)n)N8499;<+&33K@  .045E0FGH H  " ) )) 4+5 56 ~ &&rc|<|:|jj||}t||xs |jS|jdj |S)a'Returns the master address to use when creating a session. This usually returns the master from the first ClusterResolver passed in, but you can override this by specifying the task_type and task_id. Note: this is only useful for TensorFlow 1.x. Args: task_type: (Optional) The type of the TensorFlow task of the master. task_id: (Optional) The index of the TensorFlow task of the master. rpc_layer: (Optional) The RPC protocol for the given cluster. Returns: The name or URL of the session master. rra)r+rbrrXr|r rcs rr zUnionClusterResolver.master9s_ !4  "// 7Cf vy'CDOO DD  " "1 % , ,y , AArcP|jxs|jdjSNr)r=r|r-r)s rr-zUnionClusterResolver.task_typeO" ?? Bd55a8BBBrcP|jxs|jdjSr)rAr|r.r)s rr.zUnionClusterResolver.task_idSs" == >D33A6>>>rc||_yrerCrDs rr-zUnionClusterResolver.task_typeWrirc||_yrerFrGs rr.zUnionClusterResolver.task_id[rkrc4|jdjSr)r|r;r)s rr;z UnionClusterResolver.environment_s  " "1 % 1 11rcB|jdj|||Sr)r|r8rns rr8z%UnionClusterResolver.num_acceleratorscs)  " "1 % 6 67L **rcP|jxs|jdjSr)rXr|r r)s rr zUnionClusterResolver.rpc_layerjrrc||_yrerprrs rr zUnionClusterResolver.rpc_layernrirrHrsrus@rrwrws(T#BH'TB, C C ? ?    >> 2 2"&#$(* C C   rrwre)rLrMr1sixtensorflow.python.clientrtensorflow.python.eagerrtensorflow.python.frameworkrr%tensorflow.python.training.server_libr tensorflow.python.util.tf_exportr rr# add_metaclassABCMetaobjectr%rRrwr rrrsM  ,+.+=6" 893;;dfd :dN >?~ O~ @~ B 67M ?M 8M r