xref: /aosp_15_r20/external/tensorflow/tensorflow/core/data/service/test_cluster.cc (revision b6fb3261f9314811a0f4371741dbb8839866f948)
1 /* Copyright 2020 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 
16 #include "tensorflow/core/data/service/test_cluster.h"
17 
18 #include <memory>
19 #include <string>
20 #include <utility>
21 #include <vector>
22 
23 #include "absl/strings/str_cat.h"
24 #include "absl/strings/str_split.h"
25 #include "absl/strings/string_view.h"
26 #include "tensorflow/core/data/service/export.pb.h"
27 #include "tensorflow/core/data/service/server_lib.h"
28 #include "tensorflow/core/platform/errors.h"
29 #include "tensorflow/core/platform/status.h"
30 #include "tensorflow/core/protobuf/data_service.pb.h"
31 #include "tensorflow/core/protobuf/service_config.pb.h"
32 
33 namespace tensorflow {
34 namespace data {
35 namespace {
36 constexpr const char kProtocol[] = "grpc";
37 }  // namespace
38 
TestCluster(int num_workers)39 TestCluster::TestCluster(int num_workers) : num_workers_(num_workers) {}
40 
TestCluster(const TestCluster::Config & config)41 TestCluster::TestCluster(const TestCluster::Config& config)
42     : num_workers_(config.num_workers), config_(config) {}
43 
Initialize()44 Status TestCluster::Initialize() {
45   if (initialized_) {
46     return errors::FailedPrecondition(
47         "Test cluster has already been initialized.");
48   }
49   initialized_ = true;
50   experimental::DispatcherConfig dispatcher_config;
51   dispatcher_config.set_protocol(kProtocol);
52   for (int i = 0; i < num_workers_; ++i) {
53     dispatcher_config.add_worker_addresses("localhost");
54   }
55   dispatcher_config.set_deployment_mode(DEPLOYMENT_MODE_COLOCATED);
56   dispatcher_config.set_job_gc_check_interval_ms(
57       config_.job_gc_check_interval_ms);
58   dispatcher_config.set_job_gc_timeout_ms(config_.job_gc_timeout_ms);
59   dispatcher_config.set_client_timeout_ms(config_.client_timeout_ms);
60   TF_RETURN_IF_ERROR(NewDispatchServer(dispatcher_config, dispatcher_));
61   TF_RETURN_IF_ERROR(dispatcher_->Start());
62   dispatcher_address_ = absl::StrCat("localhost:", dispatcher_->BoundPort());
63   workers_.reserve(num_workers_);
64   worker_addresses_.reserve(num_workers_);
65   for (int i = 0; i < num_workers_; ++i) {
66     TF_RETURN_IF_ERROR(AddWorker());
67   }
68   return OkStatus();
69 }
70 
AddWorker()71 Status TestCluster::AddWorker() {
72   std::unique_ptr<WorkerGrpcDataServer> worker;
73   experimental::WorkerConfig config;
74   config.set_protocol(kProtocol);
75   config.set_dispatcher_address(dispatcher_address_);
76   config.set_worker_address("localhost:%port%");
77   config.set_heartbeat_interval_ms(config_.worker_heartbeat_interval_ms);
78   TF_RETURN_IF_ERROR(NewWorkerServer(config, worker));
79   TF_RETURN_IF_ERROR(worker->Start());
80   worker_addresses_.push_back(absl::StrCat("localhost:", worker->BoundPort()));
81   workers_.push_back(std::move(worker));
82   return OkStatus();
83 }
84 
DispatcherAddress() const85 std::string TestCluster::DispatcherAddress() const {
86   return dispatcher_address_;
87 }
88 
WorkerAddress(int index) const89 std::string TestCluster::WorkerAddress(int index) const {
90   DCHECK_GE(index, 0);
91   DCHECK_LT(index, worker_addresses_.size());
92   return worker_addresses_[index];
93 }
94 
StopWorker(size_t index)95 void TestCluster::StopWorker(size_t index) {
96   DCHECK_GE(index, 0);
97   DCHECK_LT(index, worker_addresses_.size());
98   workers_[index]->Stop();
99 }
100 
StopWorkers()101 void TestCluster::StopWorkers() {
102   for (std::unique_ptr<WorkerGrpcDataServer>& worker : workers_) {
103     worker->Stop();
104   }
105 }
106 
ExportDispatcherState() const107 ServerStateExport TestCluster::ExportDispatcherState() const {
108   return dispatcher_->ExportState();
109 }
110 
ExportWorkerState(size_t index) const111 ServerStateExport TestCluster::ExportWorkerState(size_t index) const {
112   return workers_[index]->ExportState();
113 }
114 
115 }  // namespace data
116 }  // namespace tensorflow
117