seda_config.cpp 13.1 KB
Newer Older
羽飞's avatar
羽飞 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Longda on 2010
//
#include "common/seda/seda_config.h"

#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <typeinfo>
#include <vector>

#include "common/lang/string.h"
#include "common/log/log.h"
#include "common/os/os.h"
#include "common/seda/init.h"
#include "common/seda/stage_factory.h"
namespace common {

SedaConfig *SedaConfig::instance_ = NULL;

31 32
SedaConfig *&SedaConfig::get_instance()
{
羽飞's avatar
羽飞 已提交
33 34 35 36 37 38 39 40
  if (instance_ == NULL) {
    instance_ = new SedaConfig();
    ASSERT((instance_ != NULL), "failed to allocate SedaConfig");
  }
  return instance_;
}

// Constructor
41 42
SedaConfig::SedaConfig() : cfg_file_(), cfg_str_(), thread_pools_(), stages_()
{
羽飞's avatar
羽飞 已提交
43 44 45 46
  return;
}

// Destructor
47 48
SedaConfig::~SedaConfig()
{
羽飞's avatar
羽飞 已提交
49 50 51 52 53 54 55 56 57 58
  ASSERT(instance_, "Instance should not be null");
  // check to see if clean-up is necessary
  if ((!thread_pools_.empty()) || (!stages_.empty())) {
    cleanup();
  }

  instance_ = NULL;
}

// Set the file holding the configuration
59 60
void SedaConfig::set_cfg_filename(const char *filename)
{
羽飞's avatar
羽飞 已提交
61 62 63 64 65 66 67 68 69
  cfg_str_.clear();
  cfg_file_.clear();
  if (filename != NULL) {
    cfg_file_.assign(filename);
  }
  return;
}

// Set the string holding the configuration
70 71
void SedaConfig::set_cfg_string(const char *config_str)
{
羽飞's avatar
羽飞 已提交
72 73 74 75 76 77 78 79 80
  cfg_str_.clear();
  cfg_file_.clear();
  if (config_str != NULL) {
    cfg_str_.assign(config_str);
  }
  return;
}

// Parse config file or string
81 82
SedaConfig::status_t SedaConfig::parse()
{
羽飞's avatar
羽飞 已提交
83 84 85 86 87 88 89 90 91 92 93 94 95 96 97
  // first parse the config
  try {
    // skip parse in this implementation
    // all configuration will be put into one file
  } catch (const std::exception &e) {
    std::string err_msg(e.what());
    LOG_ERROR("Seda config parse failed w/ error %s", err_msg.c_str());
    return PARSEFAIL;
  }
  LOG_DEBUG("Seda config parse success\n");

  return SUCCESS;
}

// instantiate the parsed SEDA configuration
98 99
SedaConfig::status_t SedaConfig::instantiate_cfg()
{
羽飞's avatar
羽飞 已提交
100 101 102 103 104 105 106 107 108
  status_t stat = SUCCESS;

  // instantiate the configuration
  stat = instantiate();

  return stat;
}

// start the configuration - puts the stages_ into action
109 110
SedaConfig::status_t SedaConfig::start()
{
羽飞's avatar
羽飞 已提交
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135
  status_t stat = SUCCESS;

  ASSERT(thread_pools_.size(), "Configuration not yet instantiated");

  // start the stages_ one by one.  connect() calls
  std::map<std::string, Stage *>::iterator iter = stages_.begin();
  std::map<std::string, Stage *>::iterator end = stages_.end();

  while (iter != end) {
    if (iter->second != NULL) {
      Stage *stg = iter->second;
      bool ret = stg->connect();
      if (!ret) {
        cleanup();
        stat = INITFAIL;
        break;
      }
    }
    iter++;
  }

  return stat;
}

// Initialize the thread_pools_ and stages_
136 137
SedaConfig::status_t SedaConfig::init()
{
羽飞's avatar
羽飞 已提交
138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159
  status_t stat = SUCCESS;

  // check the preconditions
  ASSERT(stages_.empty(), "Attempt to initialize sedaconfig twice");
  ASSERT(thread_pools_.empty(), "Attempt to initialize sedaconfig twice");

  // instantiate the parsed config
  stat = instantiate();
  if (stat) {
    return stat;
  }

  // start it running
  stat = start();
  if (stat) {
    return stat;
  }

  return SUCCESS;
}

// Clean-up the threadpool and stages_
160 161
void SedaConfig::cleanup()
{
羽飞's avatar
羽飞 已提交
162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181
  // first disconnect all stages_
  if (stages_.empty() == false) {
    std::map<std::string, Stage *>::iterator iter = stages_.begin();
    std::map<std::string, Stage *>::iterator end = stages_.end();
    while (iter != end) {
      if (iter->second != NULL) {
        Stage *stg = iter->second;
        if (stg->is_connected()) {
          stg->disconnect();
        }
      }
      iter++;
    }
  }
  LOG_TRACE("stages_ disconnected\n");

  // now delete all stages_ and thread_pools_
  clear_config();
}

182 183 184
void SedaConfig::init_event_history()
{
  std::map<std::string, std::string> base_section = get_properties()->get(SEDA_BASE_NAME);
羽飞's avatar
羽飞 已提交
185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208
  std::map<std::string, std::string>::iterator it;
  std::string key;

  // check whether event histories are enabled
  bool ev_hist = false;
  key = EVENT_HISTORY;
  it = base_section.find(key);
  if (it != base_section.end()) {
    if (it->second.compare("true") == 0) {
      ev_hist = true;
    }
  }

  get_event_history_flag() = ev_hist;

  // set max event hops
  u32_t max_event_hops = 100;
  key = MAX_EVENT_HISTORY_NUM;
  it = base_section.find(key);
  if (it != base_section.end()) {
    str_to_val(it->second, max_event_hops);
  }
  get_max_event_hops() = max_event_hops;

209
  LOG_INFO("Successfully init_event_history, EventHistory:%d, MaxEventHops:%u", (int)ev_hist, max_event_hops);
羽飞's avatar
羽飞 已提交
210 211 212
  return;
}

213 214
SedaConfig::status_t SedaConfig::init_thread_pool()
{
羽飞's avatar
羽飞 已提交
215 216
  try {

217
    std::map<std::string, std::string> base_section = get_properties()->get(SEDA_BASE_NAME);
羽飞's avatar
羽飞 已提交
218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248
    std::map<std::string, std::string>::iterator it;
    std::string key;

    // get thread pool names
    key = THREAD_POOLS_NAME;
    it = base_section.find(key);
    if (it == base_section.end()) {
      LOG_ERROR("Configuration hasn't set %s", key.c_str());
      return INITFAIL;
    }

    std::string pool_names = it->second;
    std::vector<std::string> name_list;
    std::string split_tag;
    split_tag.assign(1, Ini::CFG_DELIMIT_TAG);
    split_string(pool_names, split_tag, name_list);

    int cpu_num = getCpuNum();
    std::string default_cpu_num_str;
    val_to_str(cpu_num, default_cpu_num_str);

    for (size_t pos = 0; pos != name_list.size(); pos++) {
      std::string &thread_name = name_list[pos];

      // get count number
      key = COUNT;
      std::string count_str = get_properties()->get(key, default_cpu_num_str, thread_name);

      int thread_count = 1;
      str_to_val(count_str, thread_count);
      if (thread_count < 1) {
249
        LOG_INFO("Thread number of  %s is %d, it is same as cpu's cores.", thread_name.c_str(), cpu_num);
羽飞's avatar
羽飞 已提交
250 251 252 253 254 255 256 257
        thread_count = cpu_num;
      }
      const int max_thread_count = 1000000;
      if (thread_count >= max_thread_count) {
        LOG_ERROR("Thread number is too big: %d(max:%d)", thread_count, max_thread_count);
        return INITFAIL;
      }

258
      Threadpool *thread_pool = new Threadpool(thread_count, thread_name);
羽飞's avatar
羽飞 已提交
259 260 261 262 263 264 265
      if (thread_pool == NULL) {
        LOG_ERROR("Failed to new %s threadpool\n", thread_name.c_str());
        return INITFAIL;
      }
      thread_pools_[thread_name] = thread_pool;
    }

266 267
    if (thread_pools_.find(DEFAULT_THREAD_POOL) == thread_pools_.end()) {
      LOG_ERROR("There is no default thread pool %s, please add it.", DEFAULT_THREAD_POOL);
羽飞's avatar
羽飞 已提交
268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286
      return INITFAIL;
    }

  } catch (std::exception &e) {
    LOG_ERROR("Failed to init thread_pools_:%s\n", e.what());
    clear_config();
    return INITFAIL;
  }

  int pools = thread_pools_.size();
  if (pools < 1) {
    LOG_ERROR("Invalid number of thread_pools_:%d", pools);
    clear_config();
    return INITFAIL;
  }

  return SUCCESS;
}

287 288
std::string SedaConfig::get_thread_pool(std::string &stage_name)
{
羽飞's avatar
羽飞 已提交
289 290
  std::string ret = DEFAULT_THREAD_POOL;
  // Get thread pool
291
  std::map<std::string, std::string> stage_section = get_properties()->get(stage_name);
羽飞's avatar
羽飞 已提交
292 293 294 295
  std::map<std::string, std::string>::iterator itt;
  std::string thread_pool_id = THREAD_POOL_ID;
  itt = stage_section.find(thread_pool_id);
  if (itt == stage_section.end()) {
296
    LOG_INFO("Not set thread_pool_id for %s, use default threadpool %s", stage_name.c_str(), DEFAULT_THREAD_POOL);
羽飞's avatar
羽飞 已提交
297 298 299 300 301 302
    return ret;
  }

  std::string thread_name = itt->second;
  if (thread_name.empty()) {
    LOG_ERROR("Failed to set %s of the %s, use the default threadpool %s",
303 304 305
        thread_pool_id.c_str(),
        stage_name.c_str(),
        DEFAULT_THREAD_POOL);
羽飞's avatar
羽飞 已提交
306 307 308 309 310 311
    return ret;
  }

  if (thread_pools_.find(thread_name) == thread_pools_.end()) {
    LOG_ERROR("The stage %s's threadpool %s is invalid, use the default "
              "threadpool %s",
312 313 314
        stage_name.c_str(),
        thread_name.c_str(),
        DEFAULT_THREAD_POOL);
羽飞's avatar
羽飞 已提交
315 316 317 318
  }
  return ret;
}

319 320
SedaConfig::status_t SedaConfig::init_stages()
{
羽飞's avatar
羽飞 已提交
321
  try {
322
    std::map<std::string, std::string> base_section = get_properties()->get(SEDA_BASE_NAME);
羽飞's avatar
羽飞 已提交
323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338
    std::map<std::string, std::string>::iterator it;
    std::string key;

    // get stage names
    key = STAGES;
    it = base_section.find(key);
    if (it == base_section.end()) {
      LOG_ERROR("Hasn't set stages name in %s", key.c_str());
      clear_config();
      return INITFAIL;
    }

    std::string split_tag;
    split_tag.assign(1, Ini::CFG_DELIMIT_TAG);
    split_string(it->second, split_tag, stage_names_);

339
    for (std::vector<std::string>::iterator it = stage_names_.begin(); it != stage_names_.end(); it++) {
羽飞's avatar
羽飞 已提交
340 341 342 343 344 345 346 347 348 349 350 351 352 353
      std::string stage_name(*it);

      std::string thread_name = get_thread_pool(stage_name);
      Threadpool *t = thread_pools_[thread_name];

      Stage *stage = StageFactory::make_instance(stage_name);
      if (stage == NULL) {
        LOG_ERROR("Failed to make instance of stage %s", stage_name.c_str());
        clear_config();
        return INITFAIL;
      }
      stages_[stage_name] = stage;
      stage->set_pool(t);

354 355
      LOG_INFO("Stage %s use threadpool %s.", stage_name.c_str(), thread_name.c_str());
    }  // end for stage
羽飞's avatar
羽飞 已提交
356 357

  } catch (std::exception &e) {
358
    LOG_ERROR("Failed to parse stages information, please check, err:%s", e.what());
羽飞's avatar
羽飞 已提交
359 360 361 362 363 364 365 366 367 368 369 370 371
    clear_config();
    return INITFAIL;
  }

  if (stages_.size() < 1) {
    LOG_ERROR("Invalid number of stages_: %lu\n", stages_.size());
    clear_config();
    return INITFAIL;
  }

  return SUCCESS;
}

372 373
SedaConfig::status_t SedaConfig::gen_next_stages()
{
羽飞's avatar
羽飞 已提交
374
  try {
375
    for (std::vector<std::string>::iterator it_name = stage_names_.begin(); it_name != stage_names_.end(); it_name++) {
羽飞's avatar
羽飞 已提交
376 377 378 379

      std::string stage_name(*it_name);
      Stage *stage = stages_[stage_name];

380
      std::map<std::string, std::string> stage_section = get_properties()->get(stage_name);
羽飞's avatar
羽飞 已提交
381 382 383 384 385 386 387 388 389 390 391 392 393 394
      std::map<std::string, std::string>::iterator it;
      std::string next_stage_id = NEXT_STAGES;
      it = stage_section.find(next_stage_id);
      if (it == stage_section.end()) {
        continue;
      }

      std::string next_stage_names = it->second;

      std::vector<std::string> next_stage_name_list;
      std::string split_tag;
      split_tag.assign(1, Ini::CFG_DELIMIT_TAG);
      split_string(next_stage_names, split_tag, next_stage_name_list);

395 396 397
      for (std::vector<std::string>::iterator next_it = next_stage_name_list.begin();
           next_it != next_stage_name_list.end();
           next_it++) {
羽飞's avatar
羽飞 已提交
398 399 400 401 402
        std::string &next_stage_name = *next_it;
        Stage *next_stage = stages_[next_stage_name];
        stage->push_stage(next_stage);
      }

403
    }  // end for stage
羽飞's avatar
羽飞 已提交
404 405 406 407 408 409 410 411 412
  } catch (std::exception &e) {
    LOG_ERROR("Failed to get next stages");
    clear_config();
    return INITFAIL;
  }
  return SUCCESS;
}

// instantiate the thread_pools_ and stages_
413 414
SedaConfig::status_t SedaConfig::instantiate()
{
羽飞's avatar
羽飞 已提交
415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439

  init_event_history();

  SedaConfig::status_t status = init_thread_pool();
  if (status) {
    LOG_ERROR("Failed to init thread pool\n");
    return status;
  }

  status = init_stages();
  if (status) {
    LOG_ERROR("Failed init stages_\n");
    return status;
  }

  status = gen_next_stages();
  if (status) {
    LOG_ERROR("Failed to generate next stage list\n");
    return status;
  }

  return SUCCESS;
}

// delete all thread_pools_ and stages_
440 441
void SedaConfig::clear_config()
{
羽飞's avatar
羽飞 已提交
442 443 444 445 446 447 448 449
  // delete stages_
  std::map<std::string, Stage *>::iterator s_iter = stages_.begin();
  std::map<std::string, Stage *>::iterator s_end = stages_.end();
  while (s_iter != s_end) {
    if (s_iter->second != NULL) {

      Stage *stg = s_iter->second;
      LOG_INFO("Stage %s deleted.", stg->get_name());
450
      ASSERT((!stg->is_connected()), "%s%s", "Stage connected in clear_config ", stg->get_name());
羽飞's avatar
羽飞 已提交
451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473
      delete stg;
      s_iter->second = NULL;
    }
    s_iter++;
  }
  stages_.clear();
  LOG_INFO("Seda Stage released");

  // delete thread_pools_
  std::map<std::string, Threadpool *>::iterator t_iter = thread_pools_.begin();
  std::map<std::string, Threadpool *>::iterator t_end = thread_pools_.end();
  while (t_iter != t_end) {
    if (t_iter->second != NULL) {
      LOG_INFO("ThreadPool %s deleted.", t_iter->first.c_str());
      delete t_iter->second;
      t_iter->second = NULL;
    }
    t_iter++;
  }
  thread_pools_.clear();
  LOG_INFO("Seda thread pools released");
}

474 475
void SedaConfig::get_stage_names(std::vector<std::string> &names) const
{
羽飞's avatar
羽飞 已提交
476 477 478
  names = stage_names_;
}

479 480 481
void SedaConfig::get_stage_queue_status(std::vector<int> &stats) const
{
  for (std::map<std::string, Stage *>::const_iterator i = stages_.begin(); i != stages_.end(); ++i) {
羽飞's avatar
羽飞 已提交
482 483 484 485 486 487
    Stage *stg = (*i).second;
    stats.push_back(stg->qlen());
  }
}

// Global seda config object
488 489
SedaConfig *&get_seda_config()
{
羽飞's avatar
羽飞 已提交
490 491 492 493 494
  static SedaConfig *seda_config = NULL;

  return seda_config;
}

495
}  // namespace common