AVh`dZddlZddlZddlZddlZddlZddlZddlmZddl Z ddl m Z ddl m Z ddl mZddl mZddl mZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddl m!Z!ddl m"Z"ddl#m$Z$GddejJZ&GddejJZ'GddejPZ)GddejPZ*GddejPZ+Gdd ejPZ,Gd!d"e-Z.ej^d#ej`Z1ej^d$d%Z2d&Z3e$d'g(d9d)Z4ejjZ5ejlZ6ej^Z/d*a7e$d+g(d,Z8Gd-d.e-Z9e9a:e$d/g(d0Z;d1Z<ejze<ej|d2d3d4gZ?d5Z@d6ZAd7ZBd8ZCy):zThis module customizes `test_combinations` for `tf.distribute.Strategy`. Additionally it provides `generate()`, `combine()` and `times()` with `tf.distribute.Strategy` customizations as a default. N)app)session)collective_all_reduce_strategy)distribute_lib)multi_process_runner)multi_worker_test_base)context) def_function) combinations)config)opstest_combinations) test_util)flags) tf_logging) tf_decorator) tf_inspect) tf_exportceZdZdZdZy)DistributionParameterzTransforms arguments of type `NamedDistribution`. Convert all arguments of type `NamedDistribution` to the value of their `strategy` property. c|jdd}i}|jD]:\}}t|ts|j}|r||j _|||<<|SNuse_var_policy)getitems isinstanceNamedDistributionstrategyextended_use_var_policy)selfkwargsrequested_parametersrdistribution_argumentskvrs Y/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/combinations.pymodified_argumentsz(DistributionParameter.modified_arguments<slZZ 0$7N -1 A( ):: .<(   +$,q! - "!N__name__ __module__ __qualname____doc__r)r*r(rr5s  "r*rceZdZdZdZy)ClusterParameterszkAdds cluster parameters if a `NamedDistribution` has it. It needs to be before DistributionParameter. cd}|jD]H\}}t|ts|.t|j|j dkDr t d|}J|rj|j}|j }|j}|j} |j} d|vr|d|k7r t dd|vrm|d|k7ret d|jdd}|jdd}|jdd}|jd d } |jd d } i} d|vr|| d<d|vr|| d<d|vr|| d<d |vr| | d <d |vr| | d <| S) Nz9Only support one NamedDistribution for multi workertests. has_chiefzboth num_workers and strategy specified but are not compatibleFrunner share_gpuTnum_psr) rrr_num_total_workersr5r6 ValueErrorr7r8r9r) r"r#r$r_r'r5r6r7r8r9updates r(r)z$ClusterParameters.modified_argumentsQsH 1 A( )  $6q{{78}}%FHI%J$% % $$i((kf$$if  6+#6)#C JL L & VM%:k%I LN N**[%0iJJ}a0kzz(D)f**[$/izz(A&fF**%f[,,)f]''fX**%f[''fX Mr*Nr+r0r*r(r2r2Ks  *r*r2ceZdZdZdZdZy)DistributionCombinationz(Sets up distribution strategy for tests.c|jDcgc]}t|ts|}}tjrt d|Dryycc}w)Nc34K|]}|jywN)no_xla).0ds r( zEDistributionCombination.should_execute_combination..s)Jq!(()Js)Fz@n/a: skipping strategy combination with no_xla=True in XLA testsTN)valuesrrris_xla_enabledany)r"r#r' distributionss r(should_execute_combinationz2DistributionCombination.should_execute_combinationsW==?j4E&FM!c)JM)J&JN s AAc@ttjdgSr)rcombinations_libOptionalParameterr"s r(parameter_modifiersz+DistributionCombination.parameter_modifierss!**+;< r*N)r,r-r.r/rLrQr0r*r(r?r?~s0r*r?ceZdZdZdZy)ClusterCombinationzSets up multi worker tests.ctgSrB)r2rPs r(rQz&ClusterCombination.parameter_modifierss    r*N)r,r-r.r/rQr0r*r(rSrSs #!r*rScxeZdZdZdZej r ejdej dZdZ dZ y)GPUCombinationaEnable tests to request GPU hardware and skip non-GPU combinations. This class expects test_combinations to be generated with `NamedDistribution` wrapping instances of `tf.distribute.Strategy`. Optionally, the `required_gpus` argument is supported. GPU hardware is required, if its value is `True` or > 0. Attributes: GPU_TEST: The environment is considered to have GPU hardware available if the name of the program contains "test_gpu" or "test_xla_gpu". Fz(test_2?gpu|test_xla_2?gpu)$rc8|jDcgc]}t|ts|}}|jdd}|jdd}|r |r t dt |g|gz|Dcgc]}|j xsdc}z|Dcgc]}|jxsdc}z}t |g|Dcgc]}|j xsdc}z}|r |r t d|stjry|dkDr=tj|kr&ddjtj|fS|ttjd kDr'dd jtjd |fSy cc}wcc}wcc}wcc}w) N required_gpusrrequired_physical_gpuszLDo not use `required_gpus` and arguments of type NamedDistribution together.zOnly one of `required_physical_gpus`(number of physical GPUs required) and `required_gpus`(total number of GPUs required) should be set. )FzTest that doesn't require GPUs.Fz*Only {} of {} required GPUs are available.GPUz3Only {} of {} required physical GPUs are available.rG)rHrrrr;maxrYrXrVGPU_TESTr num_gpusformatlenr list_physical_devices) r"r#r'rKrXrYrEnumber_of_required_gpus number_of_required_physical_gpuss r(rLz)GPUCombination.should_execute_combinations==?j4E&FMJJ2M#ZZ(@!D 5 66" 1220=>1 ! ! &Q &> ?'45! A 5 67(+  0=>1 ! ! &Q &> ?(@$ = 8 99 #~'>'> 7 !A % #::BII    5799 )C$$U+-- -DKK..u57MOQQE ?5 ?sFF1F F ;F cVtjdtjdgS)NrXrYrNrOrPs r(rQz"GPUCombination.parameter_modifierss+  . . ?  . ./G H JJr*N) r,r-r.r/r\sysargvresearchrLrQr0r*r(rVrVs< (XXryy8#((1+FH#JJr*rVcZeZdZdZdZej rdej dvZdZdZy)TPUCombinationaAllow to request TPU hardware and skip non-TPU combinations. This class expects test_combinations to be generated with `NamedDistribution` wrapping instances of `tf.distribute.Strategy`. Optionally, the `required_tpus` parameter is supported. TPU hardware is required, if its argument is `True` or > 0. Optionally, the `use_cloud_tpu` parameter is supported. If TPU hardware is required by `required_tpus`, it specifically must be a Cloud TPU (specified with `--tpu`) if `use_cloud_tpu` is `True`. Attributes: TPU_TEST: The environment is considered to have TPU hardware available if the name of the program contains "test_tpu". Ftest_tpurc|jDcgc]}t|ts|}}d|vrd|vr td|j ddxs|j dd}|r |r tdt |xsdg|Dcgc]}|j xsdc}z}t|j dg|Dcgc]}|jc}z}ttjdxrtjjxsd}|stjry |rtjsy |r|sy |s|ry y cc}wcc}wcc}w)N required_tpus required_tpuzSDo not use `required_tpu`. Both `required_tpus` and `required_tpu` were specified.zLDo not use `required_tpus` and arguments of type NamedDistribution together.r use_cloud_tputpu)FzTest that doesn't require TPUs.)Fz,Test requires a TPU, but it's not available.)Fz.Test requires a Cloud TPU, but none specified.)Fz1Test requires local TPU, but Cloud TPU specified.rG)rHrrr;rr[rnrJrohasattrrFLAGSrprjTPU_TEST) r"r#r'rKrmrEnumber_of_required_tpusrorps r(rLz)TPUCombination.should_execute_combinationsU==?j4E&FM & ^v%= 8 99JJ5:M 5 66 "=#5A"6@M"N11>>#6Q#6"N#OPO452?@Q@ABM %++u % 9%++// ?RC "~'>'> 7~'>'> DS F S I ?$#O@sEEE E c~tjdtjdtjdgS)NrmrnrordrPs r(rQz"TPUCombination.parameter_modifiers s6**?;**>:**?; r*N) r,r-r.r/rtrerfrLrQr0r*r(rjrjs2"(XXSXXa[(H Dr*rjcReZdZdZ ddZedZedZdZy)rzAWraps a `tf.distribute.Strategy` and adds a name for test titles.Nc tj|||_||_||_||_||_||_||_||_ | |_ | |_ | |_ | |_ y)aInitialize NamedDistribution. Args: name: Name that will be a part of the name of the test case. distribution_fn: A callable that creates a `tf.distribute.Strategy`. required_gpus: The number of GPUs that the strategy requires. Only one of `required_gpus` and `required_physical_gpus` should be set. required_physical_gpus: Number of physical GPUs required. Only one of `required_gpus` and `required_physical_gpus` should be set. required_tpu: Whether the strategy requires TPU. use_cloud_tpu: Whether the strategy requires cloud TPU. has_chief: Whether the strategy requires a chief worker. num_workers: The number of workers that the strategy requires. num_ps: The number of parameter servers. share_gpu: Whether to share GPUs among workers. pool_runner_fn: An optional callable that returns a MultiProcessPoolRunner to run the test. no_xla: Whether to skip in XLA tests. N)object__init___name_distribution_fnrXrYrnror5r6r9r8_pool_runner_fnrC) r"namedistribution_fnrXrYrnror5r6r9r8pool_runner_fnrCs r(rzzNamedDistribution.__init__sl@ OODDJ+D&D"8D$D&DDN"DDKDN)DDKr*c<|j|jSyrB)r}rPs r(r7zNamedDistribution.runnerEs  '  ! ! ## r*c"|jSrB)r|rPs r(rzNamedDistribution.strategyKs  ""r*c|jSrB)r{rPs r(__repr__zNamedDistribution.__repr__Os ::r*) NrFFFr4rTNF) r,r-r.r/rzpropertyr7rrr0r*r(rrsXI "&'!"",\    # #r*r TfFunction NoTfFunctionc|SrBr0)fs r(rbsr*c"g}|D]}||z } |S)zConcats combinations.r0)combinedresultones r(concatres$ & c cMF -r*z-__internal__.distribute.combinations.generate)v1ctjtjtt t t f}tj|||zfd}|S)aTDistributed adapter of `tf.__internal__.test.combinations.generate`. All tests with distributed strategy should use this one instead of `tf.__internal__.test.combinations.generate`. This function has support of strategy combinations, GPU/TPU and multi worker support. See `tf.__internal__.test.combinations.generate` for usage. rczt|tr|}tj|jj D]`\}}|j tjjs0t|tjsKt||t|b|St|SrB)rtypesix iteritems__dict__copy startswithunittest TestLoadertestMethodPrefixtypes FunctionTypesetattr_multi_worker_test)test_method_or_class class_objectr~ test_methodcombination_decorators r( decoratorzgenerate..decorators&-)l"}}\-B-B-G-G-IJG $ OOH//@@ A {E$6$6 7 ,&8&E FG#< 00 "#56J#K LLr*) framework_combinationsEagerGraphCombinationTFVersionCombinationrSr?rVrjrNgenerate)r rdefault_combinationsrrs @r(rrmsk224113 +33&:=N&NP M r*Fz4__internal__.distribute.combinations.in_main_processct S)zWhether it's in the main test process. This is normally used to prepare the test environment which should only happen in the main process. Returns: A boolean. )_running_in_workerr0r*r(in_main_processrs  r*c(eZdZdZdZfdZxZS)TestEnvironmentzHolds the test environment information. Tests should modify the attributes of the instance returned by `env()` in the main process if needed, and it will be passed to the worker processes each time a test case is run. c d|_d|_yrB)tf_data_service_dispatchertotal_phsyical_gpusrPs r(rzzTestEnvironment.__init__s&*D# $Dr*cPts tdt| ||y)Nzvcombinations.env() should only be modified in the main process. Condition your code on combinations.in_main_process().)rr;super __setattr__)r"r~value __class__s r(rzTestEnvironment.__setattr__s/    C DD Ge$r*)r,r-r.r/rzr __classcell__)rs@r(rrs$ %%r*rz(__internal__.distribute.combinations.envctS)zReturns the object holds the test environment information. Tests should modify this in the main process if needed, and it will be passed to the worker processes each time a test case is run. Returns: a TestEnvironment object. )_envr0r*r(envrs  +r*ctr:ttjjdt _yy)NrZ)rr_r r`rrr0r*r(_set_total_phsyical_gpusrs1 #//6!8CEr* _TestResultstatusmessagecda|atjj |}tj }|j |}|j|jz|jz}|rtd|dd}n7|jrtd|jdd}n tdd}|jrt|j|S) aExecutes the test with the given test_id. This is a simple wrapper around TestRunner to be used with multi_process_runner. Similar to test.main(), but it executes only one test specified by test_id and returns whether the test succeeds. If the test fails, the function prints failures and errors to stdout. Args: test_id: TestCase.id() test_env: a TestEnvironment object. Returns: A boolean indicates whether the test succeeds. Tfailurerr4)rrskippedokN)rrrdefaultTestLoaderloadTestsFromNameTextTestRunnerrunfailuresexpectedFailureserrorsrrrprint)test_idtest_envtestr7rrrets r( _test_runnerrs$ $  # # 5 5g >$  " " $& ::d &__v66 6 F( Y A ?C ~~ Yq0A!0D EC T4 0C [[ #++ *r*cfd}tj}|jxsggdz}|j|}t j ||S)aDecorate test_method so that it runs in each worker. We use `multi_process_runner` to simulate multiple workers. Since we run the this function in the main process and all worker processes, this decoration behaves differently in the main process and worker procssses. In the main process, it spawns subprocesses and runs the test on each of them; in a worker process, it executes test in the same way as a normal test, e.g. setUp()/tearDown() are called before/after the test. Args: test_method: a function which must be a test method. Returns: Decorated `test_method`. Note that the decorated function has additional arguments. ct||dk(ststjr#|dkDrt |5 |fi|dddy|j }|r|j t|tf}ngtj|||d} tjt| ||tf|} | j| jj}d} |D]J} | j dk(r|j#| j$n| j dk(s?| j$} L| |j'| yy#1swYyxYw) Nr4rargsF)r5r6r9has_eval)r8rdependence_on_chiefrr)r:rrrI_multi_worker_sessionidrrrrcreate_cluster_specrMultiProcessRunnerstartjoin return_valuerfailrskipTest)r"r5r6r9r8r7r#rresults cluster_specephemeral_runner skip_reasonrrs r(rz%_multi_worker_test..decoratorsL)%'*+,/A(668VaZ ! ($D#F#$ $ggiG  g+??! l .@@  ' )  %%'44gK % ) # &..! ==I %nn  % mmK c$ s E  E)r5r6r9r8r7r)decorator_argspec)rgetfullargspecr_replacermake_decorator)rrargspecdecorator_argsrs` r(rr sg$B!H  % %k 2'LL&B+.&&N&;  $ $90A CCr*c|r|dzS|S)z2Returns the number of workers including the chief.r4r0)r5r6s r(r:r:ks ? r*cPd}|jD]O\}}t|tjs!|+t j dt jcS|}Qtjst|tjst jStjtjj}|j|}|j j#}t%j&||j)S)aReturns a context manager that enters a session that is configured for the MultiWorkerMirroredStrategy. Args: kwargs: a dict. Keyword arguments passed to the test. Returns: A context manager. If MultiWorkerMirroredStrategy is the one and only one strategy in kwargs and it's in graph mode, it's the seesion that is configured for that strategy. Otherwise, it's a no-op context manager. NzcThe test uses multiple strategies. Skipping entering a session that is configured for the strategy.)r target)rrr StrategyBaseloggingwarningr NullContextmanagerr executing_eagerlyrCollectiveAllReduceStrategyrdeepcopyr update_config_protocluster_resolvermasterrSession as_default)r#rr<r' sess_configrs r(rrrs(llnda!^001   F G%%''h  .JJ)L  ! ! ## goo/667+,,[9+  $ $ + + -&  F ; F F HHr*)r0)Dr/ collectionsrrgrerrabslrrtensorflow.python.clientrtensorflow.python.distributerrrrtensorflow.python.eagerr r tensorflow.python.frameworkr rr r rrNrtensorflow.python.platformrrrtensorflow.python.utilrr tensorflow.python.util.tf_exportrParameterModifierrr2TestCombinationr?rSrVrjryr NamedObjectfunction tf_functionno_tf_functionrrcombinetimesrrrrrrcall_after_init namedtuplerrrr:rr0r*r(r s   -G7=?+0N.+M1,</-6 ",>>",0(::0f.>>(!)99!8J%558Jv=%55=@<<Z+**<9N9NO -!--nkJ :rB&C&R  " "**  AbI  J  %f%, 5"= > 8,-%k$$]Xy4IJ ' T\C~Ir*