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