1 /**
2 * Copyright 2019-2021 Huawei Technologies Co., Ltd
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16 #include "minddata/dataset/core/config_manager.h"
17
18 #include <fstream>
19 #include <iostream>
20 #include <limits>
21 #include <string>
22 #include <thread>
23 #include <utility>
24
25 #ifndef ENABLE_ANDROID
26 #include "utils/log_adapter.h"
27 #else
28 #include "mindspore/lite/src/common/log_adapter.h"
29 #endif
30 #include "minddata/dataset/util/system_pool.h"
31 #include "utils/ms_utils.h"
32
33 namespace mindspore {
34 namespace dataset {
ConfigManager()35 ConfigManager::ConfigManager()
36 : num_parallel_workers_(kCfgParallelWorkers),
37 worker_connector_size_(kCfgWorkerConnectorSize),
38 op_connector_size_(kCfgOpConnectorSize),
39 sending_batches_(kCfgSendingBatch),
40 rank_id_(kCfgDefaultRankId),
41 seed_(kCfgDefaultSeed),
42 numa_enable_(false),
43 monitor_sampling_interval_(kCfgMonitorSamplingInterval),
44 stop_profiler_(false),
45 file_ready_(true),
46 callback_timout_(kCfgCallbackTimeout),
47 cache_host_(kCfgDefaultCacheHost),
48 cache_port_(kCfgDefaultCachePort),
49 num_connections_(kDftNumConnections),
50 prefetch_size_(kDftPrefetchSize),
51 auto_num_workers_(kDftAutoNumWorkers),
52 num_cpu_threads_(std::thread::hardware_concurrency()),
53 auto_num_workers_num_shards_(1),
54 auto_worker_config_(0),
55 enable_shared_mem_(true) {
56 num_cpu_threads_ = num_cpu_threads_ > 0 ? num_cpu_threads_ : std::numeric_limits<uint16_t>::max();
57 num_parallel_workers_ = num_parallel_workers_ < num_cpu_threads_ ? num_parallel_workers_ : num_cpu_threads_;
58 std::string env_cache_host = common::GetEnv("MS_CACHE_HOST");
59 std::string env_cache_port = common::GetEnv("MS_CACHE_PORT");
60 if (!env_cache_host.empty()) {
61 cache_host_ = env_cache_host;
62 }
63 if (!env_cache_port.empty()) {
64 char *end = nullptr;
65 cache_port_ = static_cast<int32_t>(strtol(env_cache_port.c_str(), &end, kDecimal));
66 if (*end != '\0') {
67 MS_LOG(WARNING) << "Cache port from env variable MS_CACHE_PORT is invalid\n";
68 cache_port_ = 0; // cause the port range validation to generate an error during the validation checks
69 }
70 }
71 }
72
73 // A print method typically used for debugging
Print(std::ostream & out) const74 void ConfigManager::Print(std::ostream &out) const {
75 // Don't show the test/internal ones. Only display the main ones here.
76 // fyi, boolalpha tells the output stream to write "true" and "false" for bools
77 out << "\nClient config settings :"
78 << "\nParallelOp workers : " << num_parallel_workers_
79 << "\nParallelOp worker connector size : " << worker_connector_size_
80 << "\nSize of each Connector : " << op_connector_size_ << std::endl;
81 }
82
83 // Private helper function that takes a nlohmann json format and populates the settings
FromJson(const nlohmann::json & j)84 Status ConfigManager::FromJson(const nlohmann::json &j) {
85 RETURN_IF_NOT_OK(set_num_parallel_workers(j.value("numParallelWorkers", num_parallel_workers_)));
86 set_worker_connector_size(j.value("workerConnectorSize", worker_connector_size_));
87 set_op_connector_size(j.value("opConnectorSize", op_connector_size_));
88 set_seed(j.value("seed", seed_));
89 set_monitor_sampling_interval(j.value("monitorSamplingInterval", monitor_sampling_interval_));
90 set_cache_host(j.value("cacheHost", cache_host_));
91 set_cache_port(j.value("cachePort", cache_port_));
92 set_num_connections(j.value("numConnections", num_connections_));
93 set_prefetch_size(j.value("prefetchSize", prefetch_size_));
94 return Status::OK();
95 }
96
97 // Loads a json file with the default settings and populates all the settings
LoadFile(const std::string & settingsFile)98 Status ConfigManager::LoadFile(const std::string &settingsFile) {
99 Status rc;
100 if (!Path(settingsFile).Exists()) {
101 RETURN_STATUS_UNEXPECTED("File is not found.");
102 }
103 // Some settings are mandatory, others are not (with default). If a setting
104 // is optional it will set a default value if the config is missing from the file.
105 try {
106 std::ifstream in(settingsFile);
107 nlohmann::json js;
108 in >> js;
109 rc = FromJson(js);
110 in.close();
111 } catch (const nlohmann::json::type_error &e) {
112 std::ostringstream ss;
113 ss << "Client file failed to load:\n" << e.what();
114 std::string err_msg = ss.str();
115 RETURN_STATUS_UNEXPECTED(err_msg);
116 } catch (const std::exception &err) {
117 RETURN_STATUS_UNEXPECTED("Client file failed to load.");
118 }
119 return rc;
120 }
121
122 // Setter function
set_num_parallel_workers(int32_t num_parallel_workers)123 Status ConfigManager::set_num_parallel_workers(int32_t num_parallel_workers) {
124 if (num_parallel_workers > num_cpu_threads_ || num_parallel_workers < 1) {
125 std::string err_msg = "Invalid Parameter, num_parallel_workers exceeds the boundary between 1 and " +
126 std::to_string(num_cpu_threads_) + ", as got " + std::to_string(num_parallel_workers) + ".";
127 RETURN_STATUS_UNEXPECTED(err_msg);
128 }
129 num_parallel_workers_ = num_parallel_workers;
130 return Status::OK();
131 }
132
133 // Setter function
set_worker_connector_size(int32_t connector_size)134 void ConfigManager::set_worker_connector_size(int32_t connector_size) { worker_connector_size_ = connector_size; }
135
136 // Setter function
set_op_connector_size(int32_t connector_size)137 void ConfigManager::set_op_connector_size(int32_t connector_size) { op_connector_size_ = connector_size; }
138
set_sending_batches(int64_t sending_batches)139 void ConfigManager::set_sending_batches(int64_t sending_batches) { sending_batches_ = sending_batches; }
140
seed() const141 uint32_t ConfigManager::seed() const { return seed_; }
142
set_rank_id(int32_t rank_id)143 void ConfigManager::set_rank_id(int32_t rank_id) {
144 if (rank_id_ == kCfgDefaultRankId) rank_id_ = rank_id;
145 }
146
set_numa_enable(bool numa_enable)147 void ConfigManager::set_numa_enable(bool numa_enable) { numa_enable_ = numa_enable; }
148
set_seed(uint32_t seed)149 void ConfigManager::set_seed(uint32_t seed) { seed_ = seed; }
150
set_monitor_sampling_interval(uint32_t interval)151 void ConfigManager::set_monitor_sampling_interval(uint32_t interval) { monitor_sampling_interval_ = interval; }
152
stop_dataset_profiler(bool stop_profiler)153 void ConfigManager::stop_dataset_profiler(bool stop_profiler) { stop_profiler_ = stop_profiler; }
154
set_profiler_file_status(bool file_ready)155 void ConfigManager::set_profiler_file_status(bool file_ready) { file_ready_ = file_ready; }
156
set_callback_timeout(uint32_t timeout)157 void ConfigManager::set_callback_timeout(uint32_t timeout) { callback_timout_ = timeout; }
158
set_cache_host(std::string cache_host)159 void ConfigManager::set_cache_host(std::string cache_host) { cache_host_ = std::move(cache_host); }
160
set_cache_port(int32_t cache_port)161 void ConfigManager::set_cache_port(int32_t cache_port) { cache_port_ = cache_port; }
162
set_num_connections(int32_t num_connections)163 void ConfigManager::set_num_connections(int32_t num_connections) { num_connections_ = num_connections; }
164
set_prefetch_size(int32_t prefetch_size)165 void ConfigManager::set_prefetch_size(int32_t prefetch_size) { prefetch_size_ = prefetch_size; }
166
167 } // namespace dataset
168 } // namespace mindspore
169