From cbcd8022088ce1ffdf48ab2b5e54578c71eb4628 Mon Sep 17 00:00:00 2001 From: "Ruben S. Montero" Date: Sun, 17 Nov 2013 17:41:46 +0100 Subject: [PATCH] Add multithead Information Manager to process monitor messages (cherry picked from commit 042ba56731720c4d7973015b5cc89945bebed8d3) --- include/InformationManagerDriver.h | 19 +- include/MonitorThread.h | 102 +++++++++++ src/im/InformationManager.cc | 2 +- src/im/InformationManagerDriver.cc | 190 ++------------------ src/im/MonitorThread.cc | 273 +++++++++++++++++++++++++++++ src/im/SConstruct | 3 +- 6 files changed, 400 insertions(+), 189 deletions(-) create mode 100644 include/MonitorThread.h create mode 100644 src/im/MonitorThread.cc diff --git a/include/InformationManagerDriver.h b/include/InformationManagerDriver.h index 0c23dac1b4..59556b7339 100644 --- a/include/InformationManagerDriver.h +++ b/include/InformationManagerDriver.h @@ -22,12 +22,11 @@ #include #include "Mad.h" -#include "HostPool.h" -#include "DatastorePool.h" - using namespace std; +class MonitorThreadPool; + /** * InformationManagerDriver provides a base class to implement IM * Drivers. This class implements the protocol and recover functions @@ -41,10 +40,9 @@ public: InformationManagerDriver( int userid, const map& attrs, - bool sudo, - HostPool * pool); + bool sudo); - virtual ~InformationManagerDriver(){}; + virtual ~InformationManagerDriver(); /** * Implements the IM driver protocol. @@ -75,14 +73,9 @@ public: void stop_monitor(int oid, const string& host) const; private: - /** - * Pointer to the Virtual Machine Pool, to access VMs - */ - HostPool * hpool; - - DatastorePool * dspool; - friend class InformationManager; + + MonitorThreadPool * mtpool; }; /* -------------------------------------------------------------------------- */ diff --git a/include/MonitorThread.h b/include/MonitorThread.h new file mode 100644 index 0000000000..765a8084fd --- /dev/null +++ b/include/MonitorThread.h @@ -0,0 +1,102 @@ + +/* -------------------------------------------------------------------------- */ +/* Copyright 2002-2013, OpenNebula Project (OpenNebula.org), C12G Labs */ +/* */ +/* 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. */ +/* -------------------------------------------------------------------------- */ + +#ifndef MONITOR_THREAD_H_ +#define MONITOR_THREAD_H_ + +#include +#include + +class HostPool; +class DatastorePool; +class LifeCycleManager; + +class MonitorThreadPool; + +extern "C" void * do_message_thread(void *arg); + +class MonitorThread +{ +private: + friend class MonitorThreadPool; + + friend void * do_message_thread(void *arg); + + MonitorThread(int hid, std::string res, std::string inf):host_id(hid), + result(res), hinfo64(inf){}; + + ~MonitorThread(){}; + + void do_message(); + + // Message variables + int host_id; + + std::string result; + + std::string hinfo64; + + // Pointers shared by all the MonitorThreads, init by MonitorThreadPool + static HostPool * hpool; + + static DatastorePool * dspool; + + static LifeCycleManager *lcm; + + static MonitorThreadPool * mthpool; +}; + +// ----------------------------------------------------------------------------- +// ----------------------------------------------------------------------------- + +class MonitorThreadPool +{ +public: + MonitorThreadPool(int num_threads); + + ~MonitorThreadPool(){}; + + /** + * Creates a new thread to parse and process a monitor message + * @param hid host id + * @param result of the monitor operation + * @oaram hinfo the information sent by the driver + */ + void do_message(int hid, const std::string& result, const std::string& hinfo); + + /** + * Terminates a running thread, and signal the main control thread. + */ + void exit_monitor_thread(); + +private: + + int concurrent_threads; /**< Max number of concurrent threads*/ + + int running_threads; /**< Number of running threads*/ + + //Concurrency control variables + pthread_mutex_t mutex; + + pthread_cond_t cond; +}; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +#endif /*MONITOR_THREAD_H_*/ + diff --git a/src/im/InformationManager.cc b/src/im/InformationManager.cc index 0a0b0ff0de..9bdc5ceb5c 100644 --- a/src/im/InformationManager.cc +++ b/src/im/InformationManager.cc @@ -71,7 +71,7 @@ int InformationManager::load_mads(int uid) NebulaLog::log("InM",Log::INFO,oss); - im_mad = new InformationManagerDriver(0,vattr->value(),false,hpool); + im_mad = new InformationManagerDriver(0,vattr->value(),false); rc = add(im_mad); diff --git a/src/im/InformationManagerDriver.cc b/src/im/InformationManagerDriver.cc index e48dc2285a..28176d1c74 100644 --- a/src/im/InformationManagerDriver.cc +++ b/src/im/InformationManagerDriver.cc @@ -19,6 +19,8 @@ #include "Nebula.h" #include "NebulaUtil.h" #include "VirtualMachineManagerDriver.h" +#include "MonitorThread.h" + #include /* -------------------------------------------------------------------------- */ @@ -27,13 +29,19 @@ InformationManagerDriver::InformationManagerDriver( int userid, const map& attrs, - bool sudo, - HostPool * pool): - Mad(userid,attrs,sudo),hpool(pool) + bool sudo): + Mad(userid,attrs,sudo) { - dspool = Nebula::instance().get_dspool(); + mtpool = new MonitorThreadPool(100); }; +InformationManagerDriver::~InformationManagerDriver() +{ + if (mtpool != 0) + { + delete mtpool; + } +}; /* ************************************************************************** */ /* Driver ASCII Protocol Implementation */ @@ -75,13 +83,8 @@ void InformationManagerDriver::protocol(const string& message) const ostringstream ess; - Host * host; - set vm_ids; - string hinfo64; - string* hinfo; - // Parse the driver message if ( is.good() ) @@ -111,31 +114,13 @@ void InformationManagerDriver::protocol(const string& message) const goto error_parse; } - // ----------------------- + // ------------------------------------------------------------------------- // Protocol implementation - // ----------------------- + // ------------------------------------------------------------------------- if ( action == "MONITOR" ) { - bool vm_poll; - - set lost; - map found; - set non_shared_ds; - - map datastores; - - Template tmpl; - - Datastore * ds; - - map::iterator itm; - - int rc; - - // --------------------------------------------------------------------- - // Get information from driver and decode from base64 - // --------------------------------------------------------------------- + string hinfo64; getline (is, hinfo64); @@ -144,144 +129,7 @@ void InformationManagerDriver::protocol(const string& message) const return; } - host = hpool->get(id,true); - - if ( host == 0 ) - { - goto error_host; - } - - hinfo = one_util::base64_decode(hinfo64); - - // --------------------------------------------------------------------- - // Monitoring Error - // --------------------------------------------------------------------- - if (result != "SUCCESS") - { - set vm_ids; - - host->error_info(*hinfo, vm_ids); - - Nebula &ne = Nebula::instance(); - LifeCycleManager *lcm = ne.get_lcm(); - - for (set::iterator it = vm_ids.begin(); it != vm_ids.end(); it++) - { - lcm->trigger(LifeCycleManager::MONITOR_DONE, *it); - } - - delete hinfo; - - hpool->update(host); - - host->unlock(); - - return; - } - - // --------------------------------------------------------------------- - // Get DS Information from Moniroting Information - // --------------------------------------------------------------------- - - rc = host->extract_ds_info(*hinfo, tmpl, datastores); - - delete hinfo; - - host->unlock(); - - if (rc != 0) - { - return; - } - - for (itm = datastores.begin(); itm != datastores.end(); itm++) - { - ds = dspool->get(itm->first, true); - - if (ds == 0) - { - continue; - } - - if (ds->get_type() == Datastore::SYSTEM_DS) - { - if (ds->is_shared()) - { - float total = 0, free = 0, used = 0; - ostringstream oss; - - (itm->second)->vector_value("TOTAL_MB", total); - (itm->second)->vector_value("FREE_MB", free); - (itm->second)->vector_value("USED_MB", used); - - ds->update_monitor(total, free, used); - - oss << "Datastore " << ds->get_name() << - " (" << ds->get_oid() << ") successfully monitored."; - - NebulaLog::log("ImM", Log::DEBUG, oss); - - dspool->update(ds); - } - else - { - non_shared_ds.insert(itm->first); - } - } - - ds->unlock(); - } - - // --------------------------------------------------------------------- - // Parse Host information - // --------------------------------------------------------------------- - - host = hpool->get(id,true); - - if ( host == 0 ) - { - delete hinfo; - goto error_host; - } - - rc = host->update_info(tmpl, vm_poll, lost, found, non_shared_ds); - - hpool->update(host); - - if (rc != 0) - { - host->unlock(); - - return; - } - - hpool->update_monitoring(host); - - ess << "Host " << host->get_name() << " (" << host->get_oid() << ")" - << " successfully monitored."; - - NebulaLog::log("InM", Log::DEBUG, ess); - - host->unlock(); - - if (vm_poll) - { - set::iterator its; - map::iterator itm; - - Nebula &ne = Nebula::instance(); - LifeCycleManager *lcm = ne.get_lcm(); - - for (its = lost.begin(); its != lost.end(); its++) - { - lcm->trigger(LifeCycleManager::MONITOR_DONE, *its); - } - - for (itm = found.begin(); itm != found.end(); itm++) - { - VirtualMachineManagerDriver::process_poll(itm->first, itm->second); - } - } + mtpool->do_message(id, result, hinfo64); } else if (action == "LOG") { @@ -308,12 +156,6 @@ void InformationManagerDriver::protocol(const string& message) const return; -error_host: - ess << "Could not get host " << id; - NebulaLog::log("InM",Log::ERROR,ess); - - return; - error_parse: ess << "Error while parsing driver message: " << message; NebulaLog::log("InM",Log::ERROR,ess); diff --git a/src/im/MonitorThread.cc b/src/im/MonitorThread.cc new file mode 100644 index 0000000000..b4e56d695b --- /dev/null +++ b/src/im/MonitorThread.cc @@ -0,0 +1,273 @@ + +/* -------------------------------------------------------------------------- */ +/* Copyright 2002-2013, OpenNebula Project (OpenNebula.org), C12G Labs */ +/* */ +/* 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. */ +/* -------------------------------------------------------------------------- */ + +#include "MonitorThread.h" + +#include +#include + +#include "Nebula.h" +#include "NebulaUtil.h" + +using namespace std; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +HostPool * MonitorThread::hpool; + +DatastorePool * MonitorThread::dspool; + +LifeCycleManager * MonitorThread::lcm; + +MonitorThreadPool * MonitorThread::mthpool; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +extern "C" void * do_message_thread(void *arg) +{ + MonitorThread * mt = static_cast(arg); + + mt->do_message(); + + MonitorThread::mthpool->exit_monitor_thread(); + + delete mt; + + return 0; +}; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +void MonitorThread::do_message() +{ + // ------------------------------------------------------------------------- + // Decode from base64 + // ------------------------------------------------------------------------- + string* hinfo = one_util::base64_decode(hinfo64); + + Host* host = hpool->get(host_id,true); + + if ( host == 0 ) + { + delete hinfo; + return; + } + + // ------------------------------------------------------------------------- + // Monitoring Error + // ------------------------------------------------------------------------- + if (result != "SUCCESS") + { + set vm_ids; + + host->error_info(*hinfo, vm_ids); + + for (set::iterator it = vm_ids.begin(); it != vm_ids.end(); it++) + { + lcm->trigger(LifeCycleManager::MONITOR_DONE, *it); + } + + delete hinfo; + + hpool->update(host); + + host->unlock(); + + return; + } + + // ------------------------------------------------------------------------- + // Get DS Information from Moniroting Information + // ------------------------------------------------------------------------- + map datastores; + map::iterator itm; + + Template tmpl; + Datastore * ds; + + set non_shared_ds; + + int rc = host->extract_ds_info(*hinfo, tmpl, datastores); + + delete hinfo; + + host->unlock(); + + if (rc != 0) + { + return; + } + + for (itm = datastores.begin(); itm != datastores.end(); itm++) + { + ds = dspool->get(itm->first, true); + + if (ds == 0) + { + continue; + } + + if (ds->get_type() == Datastore::SYSTEM_DS) + { + if (ds->is_shared()) + { + float total = 0, free = 0, used = 0; + ostringstream oss; + + (itm->second)->vector_value("TOTAL_MB", total); + (itm->second)->vector_value("FREE_MB", free); + (itm->second)->vector_value("USED_MB", used); + + ds->update_monitor(total, free, used); + + oss << "Datastore " << ds->get_name() << " (" << ds->get_oid() + << ") successfully monitored."; + + NebulaLog::log("ImM", Log::DEBUG, oss); + + dspool->update(ds); + } + else + { + non_shared_ds.insert(itm->first); + } + } + + ds->unlock(); + } + + // ------------------------------------------------------------------------- + // Parse Host information + // ------------------------------------------------------------------------- + bool vm_poll; + + set lost; + map found; + + ostringstream oss; + + host = hpool->get(host_id,true); + + if ( host == 0 ) + { + return; + } + + rc = host->update_info(tmpl, vm_poll, lost, found, non_shared_ds); + + hpool->update(host); + + if (rc != 0) + { + host->unlock(); + + return; + } + + hpool->update_monitoring(host); + + oss << "Host " << host->get_name() << " (" << host->get_oid() << ")" + << " successfully monitored."; + + NebulaLog::log("InM", Log::DEBUG, oss); + + host->unlock(); + + if (vm_poll) + { + set::iterator its; + map::iterator itm; + + for (its = lost.begin(); its != lost.end(); its++) + { + lcm->trigger(LifeCycleManager::MONITOR_DONE, *its); + } + + for (itm = found.begin(); itm != found.end(); itm++) + { + VirtualMachineManagerDriver::process_poll(itm->first, itm->second); + } + } +}; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +MonitorThreadPool::MonitorThreadPool(int max_thr):concurrent_threads(max_thr), + running_threads(0) +{ + //Initialize the MonitorThread constants + MonitorThread::dspool = Nebula::instance().get_dspool(); + + MonitorThread::hpool = Nebula::instance().get_hpool(); + + MonitorThread::lcm = Nebula::instance().get_lcm(); + + MonitorThread::mthpool= this; + + //Initialize concurrency variables + pthread_mutex_init(&mutex,0); + + pthread_cond_init(&cond,0); +}; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +void MonitorThreadPool::do_message(int hid, const string& result, + const string& hinfo) +{ + pthread_attr_t attr; + pthread_t id; + + pthread_mutex_lock(&mutex); + + while (running_threads >= concurrent_threads) + { + pthread_cond_wait(&cond, &mutex); + } + + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); + + MonitorThread * mt = new MonitorThread(hid, result, hinfo); + + running_threads++; + + pthread_create(&id, &attr, do_message_thread, (void *)mt); + + pthread_attr_destroy(&attr); + + pthread_mutex_unlock(&mutex); +}; + +/* -------------------------------------------------------------------------- */ +/* -------------------------------------------------------------------------- */ + +void MonitorThreadPool::exit_monitor_thread() +{ + pthread_mutex_lock(&mutex); + + running_threads--; + + pthread_cond_signal(&cond); + + pthread_mutex_unlock(&mutex); +}; diff --git a/src/im/SConstruct b/src/im/SConstruct index 217720e533..80ed466c63 100755 --- a/src/im/SConstruct +++ b/src/im/SConstruct @@ -23,7 +23,8 @@ lib_name='nebula_im' # Sources to generate the library source_files=[ 'InformationManager.cc', - 'InformationManagerDriver.cc' + 'InformationManagerDriver.cc', + 'MonitorThread.cc' ] # Build library