1 /* Copyright 2022 The TensorFlow Authors. All Rights Reserved.
2
3 Licensed under the Apache License, Version 2.0 (the "License");
4 you may not use this file except in compliance with the License.
5 You may obtain a copy of the License at
6
7 http://www.apache.org/licenses/LICENSE-2.0
8
9 Unless required by applicable law or agreed to in writing, software
10 distributed under the License is distributed on an "AS IS" BASIS,
11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 See the License for the specific language governing permissions and
13 limitations under the License.
14 ==============================================================================*/
15 #include "tensorflow/core/data/service/client/validate_utils.h"
16
17 #include "tensorflow/core/data/service/client/common.h"
18 #include "tensorflow/core/data/service/common.h"
19 #include "tensorflow/core/data/service/common.pb.h"
20 #include "tensorflow/core/data/service/worker_impl.h"
21 #include "tensorflow/core/framework/dataset.h"
22 #include "tensorflow/core/platform/errors.h"
23 #include "tensorflow/core/platform/status.h"
24 #include "tensorflow/core/protobuf/data_service.pb.h"
25
26 namespace tensorflow {
27 namespace data {
28 namespace {
29
30 // Validates local worker related parameters.
ValidateLocalWorkers(const DataServiceParams & data_service_params)31 Status ValidateLocalWorkers(const DataServiceParams& data_service_params) {
32 if (data_service_params.target_workers != TARGET_WORKERS_LOCAL) {
33 return OkStatus();
34 }
35 if (LocalWorkers::Empty()) {
36 if (IsStaticShard(data_service_params.processing_mode)) {
37 return errors::InvalidArgument(
38 "Static sharding policy <",
39 ProcessingModeDef::ShardingPolicy_Name(
40 data_service_params.processing_mode.sharding_policy()),
41 "> requires local tf.data workers, but no local worker is found. "
42 "You need to run local tf.data service workers in your training "
43 "workers. Static sharding also requires a fixed worker pool and "
44 "a list of worker addresses in the DispatcherConfig. See the "
45 "\"Processing Modes\" section in the module doc for details.");
46 }
47 return errors::InvalidArgument(
48 "Local reads require local tf.data workers, but no local worker "
49 "is found. You need to run local tf.data service workers in your "
50 "training workers.");
51 }
52 if (data_service_params.num_consumers.has_value()) {
53 return errors::InvalidArgument(
54 "Coordinated reads require non-local workers, but `target_workers` "
55 "is \"LOCAL\".");
56 }
57 return OkStatus();
58 }
59
60 // Validates cross-trainer cache related parameters.
ValidateCrossTrainerCache(const DataServiceParams & data_service_params)61 Status ValidateCrossTrainerCache(const DataServiceParams& data_service_params) {
62 if (!data_service_params.cross_trainer_cache_options.has_value()) {
63 return OkStatus();
64 }
65 if (data_service_params.job_name.empty()) {
66 return errors::InvalidArgument(
67 "Cross-trainer caching requires named jobs. Got empty `job_name`.");
68 }
69 if (data_service_params.metadata.cardinality() >= 0) {
70 return errors::InvalidArgument(
71 "Cross-trainer caching requires the input dataset to be infinite. "
72 "Got input with cardinality ",
73 data_service_params.metadata.cardinality());
74 }
75 if (data_service_params.repetition > 1) {
76 return errors::InvalidArgument(
77 "Cross-trainer caching requires infinite datasets and disallows "
78 "multiple repetitions of the same dataset. Got repetition ",
79 data_service_params.repetition);
80 }
81 if (data_service_params.num_consumers.has_value()) {
82 return errors::InvalidArgument(
83 "Cross-trainer caching does not support coordinated reads. "
84 "Got number of coordinated consumers: ",
85 data_service_params.num_consumers.value());
86 }
87 return OkStatus();
88 }
89 } // namespace
90
ValidateDataServiceParams(const DataServiceParams & data_service_params)91 Status ValidateDataServiceParams(const DataServiceParams& data_service_params) {
92 TF_RETURN_IF_ERROR(ValidateLocalWorkers(data_service_params));
93 TF_RETURN_IF_ERROR(ValidateCrossTrainerCache(data_service_params));
94 return OkStatus();
95 }
96
97 } // namespace data
98 } // namespace tensorflow
99