enh(broker/disk_accessor): queue size limited (#342)

* enh(persistent_file): acceses to inner splitter simplified
* cleanup(broker/persistent_file): little simplification
* enh(bench): new bench on file size
* fix(broker/applier/state): connected pollers need a mutex
* fix(broker/bam): unit tests use make_unique
* fix(broker/engine): unused code removed
* fix(broker/splitter): segfault fixed

REFS: MON-14375

CMakeLists.txt

@@ -1,35 +1,36 @@
-##
-## Copyright 2009-2021 Centreon
-##
-## Licensed under the Apache License, Version 2.0 (the "License");
-## you may not use this file except in compliance with the License.
-## You may obtain a copy of the License at
-##
-##     http://www.apache.org/licenses/LICENSE-2.0
-##
-## Unless required by applicable law or agreed to in writing, software
-## distributed under the License is distributed on an "AS IS" BASIS,
-## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-## See the License for the specific language governing permissions and
-## limitations under the License.
-##
-## For more information : contact@centreon.com
-##
+#
+# Copyright 2009-2021 Centreon
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+#
+# For more information : contact@centreon.com
+#
 
 #
 # Global settings.
 #
 
-
 # Set necessary settings.
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 cmake_minimum_required(VERSION 3.16)
 project("Centreon Collect" C CXX)
-if (NOT CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND NOT CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
-  message(FATAL_ERROR "You can build broker with g++ or clang++. CMake will exit.")
-endif ()
-#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++14 -stdlib=libc++")
-#set(CMAKE_CXX_COMPILER "clang++")
+if(NOT CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND NOT CMAKE_CXX_COMPILER_ID
+                                                STREQUAL "Clang")
+  message(
+    FATAL_ERROR "You can build broker with g++ or clang++. CMake will exit.")
+endif()
+# set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++14 -stdlib=libc++")
+# set(CMAKE_CXX_COMPILER "clang++")
 add_definitions("-D_GLIBCXX_USE_CXX11_ABI=1")
 set(CMAKE_CXX_STANDARD 14)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
@@ -42,39 +43,41 @@ set(BUILD_ARGS "-w" "dupbuild=warn")
 #
 # Get distributions name
 #
-if (CMAKE_SYSTEM_NAME STREQUAL "Linux")
-  file(STRINGS "/etc/os-release" release
-       REGEX "^ID")
-  foreach (l ${release})
-    if (${l} MATCHES "ID_LIKE=.*")
-      string(REGEX REPLACE "ID_LIKE=\"(.*)\"" "\\1" like ${l})
-    endif ()
-
-    if (${l} MATCHES "ID=.*")
-      string(REGEX REPLACE "ID=\"(.*)\"" "\\1" id ${l})
-    endif ()
-  endforeach ()
+if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
+  file(STRINGS "/etc/os-release" release REGEX "^ID")
+  foreach(line ${release})
+    if(${line} MATCHES "ID_LIKE=.*")
+      string(REGEX REPLACE "ID_LIKE=\"(.*)\"" "\\1" like ${line})
+    endif()
+
+    if(${line} MATCHES "ID=.*")
+      string(REGEX REPLACE "ID=\"(.*)\"" "\\1" id ${line})
+    endif()
+  endforeach()
   string(TOLOWER "${like}" like)
   string(TOLOWER "${id}" id)
 
-  if (("${id}" MATCHES "debian") OR ("${like}" MATCHES "debian") OR ("${id}" MATCHES "ubuntu") OR ("${like}" MATCHES "ubuntu"))
+  if(("${id}" MATCHES "debian")
+     OR ("${like}" MATCHES "debian")
+     OR ("${id}" MATCHES "ubuntu")
+     OR ("${like}" MATCHES "ubuntu"))
     set(OS_DISTRIBUTOR "Debian")
-  elseif (("${id}" MATCHES "centos") OR ("${like}" MATCHES "centos"))
+  elseif(("${id}" MATCHES "centos") OR ("${like}" MATCHES "centos"))
     set(OS_DISTRIBUTOR "CentOS")
-  else ()
+  else()
     message(WARNING "lsb_release in not installed")
     set(OS_DISTRIBUTOR "${CMAKE_SYSTEM_NAME}")
-  endif ()
-else ()
+  endif()
+else()
   set(OS_DISTRIBUTOR "${CMAKE_SYSTEM_NAME}")
-endif ()
+endif()
 
 message(STATUS "${id} detected (compatible with ${OS_DISTRIBUTOR})")
 
 # set -latomic if OS is Raspbian.
-if (CMAKE_SYSTEM_PROCESSOR MATCHES "arm")
+if(CMAKE_SYSTEM_PROCESSOR MATCHES "arm")
   set(CMAKE_CXX_LINK_FLAGS "${CMAKE_CXX_LINK_FLAGS} -latomic")
-endif ()
+endif()
 
 # Version.
 set(COLLECT_MAJOR 22)
@@ -83,25 +86,27 @@ set(COLLECT_PATCH 0)
 set(COLLECT_VERSION "${COLLECT_MAJOR}.${COLLECT_MINOR}.${COLLECT_PATCH}")
 add_definitions(-DCENTREON_CONNECTOR_VERSION=\"${COLLECT_VERSION}\")
 
-
-############ CONSTANTS ###########
+# ########### CONSTANTS ###########
 set(USER_BROKER centreon-broker)
 set(USER_ENGINE centreon-engine)
-##################################
-
-
+# ##############################################################################
 
 include(${CMAKE_BINARY_DIR}/conanbuildinfo.cmake)
 conan_basic_setup(TARGETS)
 
 include(GNUInstallDirs)
 
 # var directories.
-set(BROKER_VAR_LOG_DIR "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-broker")
-set(BROKER_VAR_LIB_DIR "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-broker")
-set(ENGINE_VAR_LOG_DIR "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine")
-set(ENGINE_VAR_LOG_ARCHIVE_DIR "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine/archives")
-set(ENGINE_VAR_LIB_DIR "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-engine")
+set(BROKER_VAR_LOG_DIR
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-broker")
+set(BROKER_VAR_LIB_DIR
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-broker")
+set(ENGINE_VAR_LOG_DIR
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine")
+set(ENGINE_VAR_LOG_ARCHIVE_DIR
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/log/centreon-engine/archives")
+set(ENGINE_VAR_LIB_DIR
+    "${CMAKE_INSTALL_FULL_LOCALSTATEDIR}/lib/centreon-engine")
 
 set(CMAKE_INSTALL_PREFIX "/usr")
 option(WITH_TESTING "Build unit tests." OFF)
@@ -110,18 +115,17 @@ option(WITH_CONF "Install configuration files." ON)
 
 # Code coverage on unit tests
 option(WITH_COVERAGE "Add code coverage on unit tests." OFF)
-if (WITH_TESTING AND WITH_COVERAGE)
+if(WITH_TESTING AND WITH_COVERAGE)
   set(CMAKE_BUILD_TYPE "Debug")
   include(cmake/CodeCoverage.cmake)
-  APPEND_COVERAGE_COMPILER_FLAGS()
-endif ()
+  append_coverage_compiler_flags()
+endif()
 
 set(protobuf_MODULE_COMPATIBLE True)
 
 add_definitions(${spdlog_DEFINITIONS})
 
-include_directories(${CMAKE_SOURCE_DIR}
-                    ${CONAN_INCLUDE_DIRS}
+include_directories(${CMAKE_SOURCE_DIR} ${CONAN_INCLUDE_DIRS}
                     ${CMAKE_SOURCE_DIR}/clib/inc)
 
 add_subdirectory(bbdo)
@@ -131,23 +135,12 @@ add_subdirectory(engine)
 add_subdirectory(connectors)
 add_subdirectory(ccc)
 
-add_custom_target(test-broker
-    COMMAND tests/ut_broker
-    )
-add_custom_target(test-engine
-    COMMAND tests/ut_engine
-    )
-add_custom_target(test-clib
-    COMMAND tests/ut_clib
-    )
-add_custom_target(test-connector
-    COMMAND tests/ut_connector
-    )
-
-add_custom_target(test
-    DEPENDS test-broker test-engine test-clib test-connector
-    )
-
-add_custom_target(test-coverage
-    DEPENDS broker-test-coverage engine-test-coverage clib-test-coverage
-    )
+add_custom_target(test-broker COMMAND tests/ut_broker)
+add_custom_target(test-engine COMMAND tests/ut_engine)
+add_custom_target(test-clib COMMAND tests/ut_clib)
+add_custom_target(test-connector COMMAND tests/ut_connector)
+
+add_custom_target(test DEPENDS test-broker test-engine test-clib test-connector)
+
+add_custom_target(test-coverage DEPENDS broker-test-coverage
+                                        engine-test-coverage clib-test-coverage)

bbdo/CMakeLists.txt

@@ -1,20 +1,20 @@
-# #
-# # Copyright 2021 Centreon
-# #
-# # Licensed under the Apache License, Version 2.0 (the "License");
-# # you may not use this file except in compliance with the License.
-# # You may obtain a copy of the License at
-# #
-# #     http://www.apache.org/licenses/LICENSE-2.0
-# #
-# # Unless required by applicable law or agreed to in writing, software
-# # distributed under the License is distributed on an "AS IS" BASIS,
-# # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# # See the License for the specific language governing permissions and
-# # limitations under the License.
-# #
-# # For more information : contact@centreon.com
-# #
+#
+# Copyright 2021-2022 Centreon
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# For more information : contact@centreon.com
+#
 
 set(protobuf_files
   rebuild_message

broker/CMakeLists.txt

@@ -327,6 +327,7 @@ set(LIBROKER_SOURCES
   ${SRC_DIR}/config/state.cc
   ${SRC_DIR}/database_config.cc
   ${SRC_DIR}/file/cfile.cc
+  ${SRC_DIR}/file/disk_accessor.cc
   ${SRC_DIR}/file/directory_event.cc
   ${SRC_DIR}/file/directory_watcher.cc
   ${SRC_DIR}/file/factory.cc

broker/bam/inc/com/centreon/broker/bam/configuration/applier/kpi.hh

@@ -67,7 +67,8 @@ class kpi {
   void _invalidate_ba(configuration::kpi const& cfg);
   std::map<uint32_t, applied>::iterator _remove_kpi(
       std::map<uint32_t, applied>::iterator& kpi_it);
-  void _resolve_kpi(configuration::kpi const& cfg, std::shared_ptr<bam::kpi>);
+  void _resolve_kpi(configuration::kpi const& cfg,
+                    const std::shared_ptr<bam::kpi>&);
 
   std::map<uint32_t, applied> _applied;
   ba* _bas;

broker/bam/inc/com/centreon/broker/bam/kpi_ba.hh

@@ -54,7 +54,7 @@ class kpi_ba : public kpi {
   void _open_new_event(io::stream* visitor,
                        int impact,
                        state ba_state,
-                       timestamp event_start_time);
+                       const timestamp& event_start_time);
 
  public:
   kpi_ba(uint32_t kpi_id, uint32_t ba_id);

broker/bam/src/ba.cc

@@ -674,8 +674,7 @@ void ba::service_update(const std::shared_ptr<neb::downtime>& dt,
  */
 void ba::save_inherited_downtime(persistent_cache& cache) const {
   if (_inherited_downtime)
-    cache.add(std::shared_ptr<inherited_downtime>(
-        new inherited_downtime(*_inherited_downtime)));
+    cache.add(std::make_shared<inherited_downtime>(*_inherited_downtime));
 }
 
 /**

broker/bam/src/bool_call.cc

@@ -104,7 +104,7 @@ std::string const& bool_call::get_name() const {
  *  @param[in] expression  The expression.
  */
 void bool_call::set_expression(std::shared_ptr<bool_value> expression) {
-  _expression = expression;
+  _expression = std::move(expression);
 }
 
 /**
@@ -118,5 +118,5 @@ void bool_call::set_expression(std::shared_ptr<bool_value> expression) {
 bool bool_call::child_has_update(computable* child, io::stream* visitor) {
   (void)child;
   (void)visitor;
-  return (true);
+  return true;
 }

broker/bam/src/bool_not.cc

@@ -28,7 +28,7 @@ constexpr double eps = 0.000001;
  *
  *  @param[in] val Value that will be negated.
  */
-bool_not::bool_not(bool_value::ptr val) : _value(val) {}
+bool_not::bool_not(bool_value::ptr val) : _value(std::move(val)) {}
 
 /**
  *  Copy constructor.

broker/bam/src/configuration/applier/kpi.cc

@@ -267,9 +267,9 @@ std::shared_ptr<bam::kpi> applier::kpi::_new_kpi(
     log_v2::bam()->info(
         "BAM: creating new KPI {} of service ({}, {}) impacting BA {}",
         cfg.get_id(), cfg.get_host_id(), cfg.get_service_id(), cfg.get_ba_id());
-    auto obj{std::make_shared<bam::kpi_service>(
-        cfg.get_id(), cfg.get_ba_id(), cfg.get_host_id(),
-        cfg.get_service_id())};
+    auto obj{std::make_shared<bam::kpi_service>(cfg.get_id(), cfg.get_ba_id(),
+                                                cfg.get_host_id(),
+                                                cfg.get_service_id())};
     obj->set_acknowledged(cfg.is_acknowledged());
     obj->set_downtimed(cfg.is_downtimed());
     obj->set_impact_critical(cfg.get_impact_critical());
@@ -312,7 +312,7 @@ std::shared_ptr<bam::kpi> applier::kpi::_new_kpi(
  * @param kpi The Kpi itself.
  */
 void applier::kpi::_resolve_kpi(configuration::kpi const& cfg,
-                                std::shared_ptr<bam::kpi> kpi) {
+                                const std::shared_ptr<bam::kpi>& kpi) {
   // Find target BA.
   uint32_t ba_id = cfg.get_ba_id();
   std::shared_ptr<bam::ba> my_ba(_bas->find_ba(ba_id));

broker/bam/src/connector.cc

@@ -45,7 +45,7 @@ void connector::connect_monitoring(std::string const& ext_cmd_file,
   _type = bam_monitoring_type;
   _ext_cmd_file = ext_cmd_file;
   _db_cfg = db_cfg;
-  _cache = cache;
+  _cache = std::move(cache);
   if (storage_db_name.empty())
     _storage_db_name = db_cfg.get_name();
   else

broker/bam/src/kpi_ba.cc

@@ -257,7 +257,7 @@ void kpi_ba::_fill_impact(impact_values& impact,
 void kpi_ba::_open_new_event(io::stream* visitor,
                              int impact,
                              state ba_state,
-                             timestamp event_start_time) {
+                             const timestamp& event_start_time) {
   _event = std::make_shared<kpi_event>(_id, _ba_id, event_start_time);
   _event->impact_level = impact;
   _event->in_downtime = _ba->get_in_downtime();

broker/bam/src/monitoring_stream.cc

@@ -61,7 +61,7 @@ monitoring_stream::monitoring_stream(std::string const& ext_cmd_file,
       _mysql(db_cfg),
       _pending_events(0),
       _storage_db_cfg(storage_db_cfg),
-      _cache(cache) {
+      _cache(std::move(cache)) {
   log_v2::bam()->trace("BAM: monitoring_stream constructor");
   // Prepare queries.
   _prepare();

broker/bam/src/timeperiod_map.cc

@@ -92,7 +92,7 @@ com::centreon::broker::time::timeperiod::ptr timeperiod_map::get_timeperiod(
  *  @param[in] ptr  A ptr to the timeperiod to add.
  */
 void timeperiod_map::add_timeperiod(uint32_t id, time::timeperiod::ptr ptr) {
-  _map[id] = ptr;
+  _map[id] = std::move(ptr);
 }
 
 /**

broker/bam/test/ba/kpi_ba.cc

@@ -39,11 +39,11 @@ class KpiBA : public ::testing::Test {
  public:
   void SetUp() override {
     // Initialization.
-    config::applier::init(0, "test_broker");
+    config::applier::init(0, "test_broker", 0);
 
-    _aply_state.reset(new bam::configuration::applier::state);
-    _state.reset(new bam::configuration::state);
-    _visitor.reset(new test_visitor("test-visitor"));
+    _aply_state = std::make_unique<bam::configuration::applier::state>();
+    _state = std::make_unique<bam::configuration::state>();
+    _visitor = std::make_unique<test_visitor>("test-visitor");
   }
 
   void TearDown() override {