From f608ba2a35243b887a3c401f1262ea273354d58e Mon Sep 17 00:00:00 2001
From: amichelo <andrea.michelotti@lnf.infn.it>
Date: Thu, 13 Apr 2017 16:17:14 +0200
Subject: [PATCH] webui multithread++
---
CMakeLists.txt | 2 +-
ChaosMetadataServiceClient/CMakeLists.txt | 4 +-
.../node_controller/CUController.cpp | 25 +++---
.../node_controller/CUController.h | 4 +-
ChaosMetadataServiceClientTest/main.cpp | 4 +-
ChaosWANProxy/CMakeLists.txt | 3 +-
.../wan_interface/http/HTTPUIInterface.cpp | 87 ++++++++++++-------
.../wan_interface/http/HTTPUIInterface.h | 8 +-
.../http/HTTPWANInterfaceRequest.h | 2 +-
9 files changed, 88 insertions(+), 51 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 666da06b1..15eb49f46 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -31,7 +31,7 @@ ENDIF()
###### Required Libraries
CheckConfigureBuild(jsoncpp jsoncpp "-DJSONCPP_WITH_TESTS=OFF;-DJSONCPP_WITH_POST_BUILD_UNITTEST=OFF" https://github.com/bisegni )
-# CheckConfigureBuild(mongoose mongoose-cpp "-DHAS_JSONCPP=ON;-DENABLE_PIC=ON" https://github.com/amichelotti )
+
CheckConfigureBuild(mongoclient mongo-cxx-driver:legacy-1.0.0-rc0 "--extralib=boost_system;--disable-warnings-as-errors;install-mongoclient" https://github.com/mongodb)
#########
diff --git a/ChaosMetadataServiceClient/CMakeLists.txt b/ChaosMetadataServiceClient/CMakeLists.txt
index 49949f8ad..723e675d1 100644
--- a/ChaosMetadataServiceClient/CMakeLists.txt
+++ b/ChaosMetadataServiceClient/CMakeLists.txt
@@ -40,7 +40,9 @@ SET(mcl_src ${mcl_src} api_proxy/agent/ListNodeForAgent.cpp
api_proxy/agent/RemoveNodeAssociation.cpp
api_proxy/agent/LoadAgentDescription.cpp
api_proxy/agent/NodeOperation.cpp
- api_proxy/agent/CheckAgentHostedProcess.cpp)
+ api_proxy/agent/CheckAgentHostedProcess.cpp
+ api_proxy/agent/logging/GetProcessLogEntries.cpp
+ api_proxy/agent/logging/ManageNodeLogging.cpp)
SET(mcl_src ${mcl_src} api_proxy/agent/logging/ManageNodeLogging.cpp)
diff --git a/ChaosMetadataServiceClient/node_controller/CUController.cpp b/ChaosMetadataServiceClient/node_controller/CUController.cpp
index 142d77c75..9d37725ea 100644
--- a/ChaosMetadataServiceClient/node_controller/CUController.cpp
+++ b/ChaosMetadataServiceClient/node_controller/CUController.cpp
@@ -244,7 +244,7 @@ int CUController::getDeviceAttributeType(const string& attr, DataType::DataType&
int CUController::getType(std::string& control_unit_type) {
int err = 0;
if(cu_type.empty()) {
- CDataWrapper*tmp=fetchCurrentDatatasetFromDomain(KeyDataStorageDomainSystem);
+ CDataWrapper*tmp=fetchCurrentDatatasetFromDomain(KeyDataStorageDomainSystem).get();
if(tmp && tmp->hasKey(DataPackSystemKey::DP_SYS_UNIT_TYPE)){
std::string t=tmp->getCStringValue(DataPackSystemKey::DP_SYS_UNIT_TYPE);
cu_type = t;
@@ -302,7 +302,7 @@ int CUController::recoverDeviceFromError() {
//---------------------------------------------------------------------------------------------------
uint64_t CUController::getState(CUStateKey::ControlUnitState& deviceState) {
uint64_t ret=0;
- CDataWrapper*tmp=fetchCurrentDatatasetFromDomain(KeyDataStorageDomainHealth);
+ CDataWrapper*tmp=fetchCurrentDatatasetFromDomain(KeyDataStorageDomainHealth).get();
deviceState=CUStateKey::UNDEFINED;
if(tmp && tmp->hasKey(NodeHealtDefinitionKey::NODE_HEALT_STATUS)){
std::string state=tmp->getCStringValue(NodeHealtDefinitionKey::NODE_HEALT_STATUS);
@@ -732,7 +732,7 @@ void CUController::initializeAttributeIndexMap() {
if(datasetDB.getAttributeRangeValueInfo(*iter, attributerangeInfo)!=0){
LERR_<<"CANNOT RETRIVE attr range info of:"<<*iter;
}
- LDBG_<<"IN attr:"<<attributerangeInfo.name<<" type:"<<attributerangeInfo.valueType<<" bin type:"<<attributerangeInfo.binType;
+ // LDBG_<<"IN attr:"<<attributerangeInfo.name<<" type:"<<attributerangeInfo.valueType<<" bin type:"<<attributerangeInfo.binType;
attributeValueMap.insert(make_pair(*iter, attributerangeInfo));
}
@@ -748,7 +748,7 @@ void CUController::initializeAttributeIndexMap() {
LERR_<<"CANNOT RETRIVE attr range info of:"<<*iter;
}
- LDBG_<<"OUT attr:"<<attributerangeInfo.name<<" type:"<<attributerangeInfo.valueType<<" bin type:"<<attributerangeInfo.binType;
+ // LDBG_<<"OUT attr:"<<attributerangeInfo.name<<" type:"<<attributerangeInfo.valueType<<" bin type:"<<attributerangeInfo.binType;
attributeValueMap.insert(make_pair(*iter, attributerangeInfo));
}
@@ -898,29 +898,30 @@ CDataWrapper * CUController::getLiveCDataWrapperPtr() {
//---------------------------------------------------------------------------------------------------
-CDataWrapper * CUController::getCurrentDatasetForDomain(DatasetDomain domain) {
+boost::shared_ptr<chaos::common::data::CDataWrapper> CUController::getCurrentDatasetForDomain(DatasetDomain domain) {
+ boost::mutex::scoped_lock lock(trackMutext);
if(domain<current_dataset.size()){
- return current_dataset[domain].get();
+ return current_dataset[domain];
}
- return NULL;
+ return current_dataset[0];
}
//---------------------------------------------------------------------------------------------------
-chaos::common::data::CDataWrapper * CUController::fetchCurrentDatatasetFromDomain(DatasetDomain domain) {
+boost::shared_ptr<chaos::common::data::CDataWrapper> CUController::fetchCurrentDatatasetFromDomain(DatasetDomain domain) {
CHAOS_ASSERT(ioLiveDataDriver.get())
char *value = NULL;
unsigned long value_len = 0;
-
+ boost::mutex::scoped_lock lock(trackMutext);
if(domain<current_dataset.size()){
value = ioLiveDataDriver->retriveRawData(channel_keys[domain],(size_t*)&value_len);
if(value){
chaos::common::data::CDataWrapper *tmp = new CDataWrapper(value);
current_dataset[domain].reset(tmp);
free(value);
- return tmp;
+ return current_dataset[domain];
}
}
- return NULL;
+ return current_dataset[0];
}
//---------------------------------------------------------------------------------------------------
@@ -1003,7 +1004,7 @@ CDataWrapper *CUController::getCurrentData(){
//! get profile info
cu_prof_t CUController::getProfileInfo(){
- chaos::common::data::CDataWrapper *prof= fetchCurrentDatatasetFromDomain(KeyDataStorageDomainHealth);
+ chaos::common::data::CDataWrapper *prof= fetchCurrentDatatasetFromDomain(KeyDataStorageDomainHealth).get();
cu_prof_t p;
bzero(&p,sizeof(cu_prof_t));
if(prof){
diff --git a/ChaosMetadataServiceClient/node_controller/CUController.h b/ChaosMetadataServiceClient/node_controller/CUController.h
index 040f7f734..7a2b25596 100644
--- a/ChaosMetadataServiceClient/node_controller/CUController.h
+++ b/ChaosMetadataServiceClient/node_controller/CUController.h
@@ -389,10 +389,10 @@ namespace chaos {
chaos::common::data::CDataWrapper * getLiveCDataWrapperPtr();
//!return the last fetched dataset for the domain
- chaos::common::data::CDataWrapper * getCurrentDatasetForDomain(DatasetDomain domain);
+ boost::shared_ptr<chaos::common::data::CDataWrapper> getCurrentDatasetForDomain(DatasetDomain domain);
//! fetch from the chaso central cache the dataset associated to the domain
- chaos::common::data::CDataWrapper * fetchCurrentDatatasetFromDomain(DatasetDomain domain);
+ boost::shared_ptr<chaos::common::data::CDataWrapper> fetchCurrentDatatasetFromDomain(DatasetDomain domain);
/*!
Fetch the current live value form live storage
diff --git a/ChaosMetadataServiceClientTest/main.cpp b/ChaosMetadataServiceClientTest/main.cpp
index 417c37c74..676248500 100644
--- a/ChaosMetadataServiceClientTest/main.cpp
+++ b/ChaosMetadataServiceClientTest/main.cpp
@@ -171,7 +171,7 @@ int main(int argc, char *argv[]){
uint64_t last_stat_ts = start_stat_ts;
while(work) {
start_stat_ts = TimingUtil::getTimeStamp();
- ds = cu_ctrl->fetchCurrentDatatasetFromDomain(chaos::cu::data_manager::KeyDataStorageDomainOutput);
+ ds = cu_ctrl->fetchCurrentDatatasetFromDomain(chaos::cu::data_manager::KeyDataStorageDomainOutput).get();
if(ds) {
counter++;
}
@@ -193,7 +193,7 @@ int main(int argc, char *argv[]){
chaos::metadata_service_client::node_controller::CUController *cu_ctrl = NULL;
while(work){
ChaosMetadataServiceClient::getInstance()->getNewCUController(device_id,&cu_ctrl);
- ds = cu_ctrl->fetchCurrentDatatasetFromDomain(chaos::cu::data_manager::KeyDataStorageDomainOutput);
+ ds = cu_ctrl->fetchCurrentDatatasetFromDomain(chaos::cu::data_manager::KeyDataStorageDomainOutput).get();
ChaosMetadataServiceClient::getInstance()->deleteCUController(cu_ctrl);
work = ((chaos::common::utility::TimingUtil::getTimeStamp()-start_ts) < (wait_seconds*1000));
}
diff --git a/ChaosWANProxy/CMakeLists.txt b/ChaosWANProxy/CMakeLists.txt
index 97d030f42..ad5311c4d 100644
--- a/ChaosWANProxy/CMakeLists.txt
+++ b/ChaosWANProxy/CMakeLists.txt
@@ -1,3 +1,4 @@
+
cmake_minimum_required(VERSION 2.6)
# IF (DEFINED ENV{CHAOS_BUNDLE} )
# include($ENV{CHAOS_BUNDLE}/tools/project_template/CMakeChaos.txt)
@@ -42,5 +43,5 @@ IF(BUILD_FORCE_STATIC)
SET(CMAKE_EXE_LINKER_FLAGS "-static -Wl,--whole-archive -lchaos_common -Wl,--no-whole-archive")
ENDIF()
-TARGET_LINK_LIBRARIES(ChaosWANProxy chaos_common chaos_metadata_service_client chaos_driver_misc mongoose ${FrameworkLib})
+TARGET_LINK_LIBRARIES(ChaosWANProxy chaos_common chaos_metadata_service_client chaos_driver_misc ${FrameworkLib})
INSTALL_TARGETS(/bin ChaosWANProxy)
diff --git a/ChaosWANProxy/wan_interface/http/HTTPUIInterface.cpp b/ChaosWANProxy/wan_interface/http/HTTPUIInterface.cpp
index eccc517e2..3755c943b 100644
--- a/ChaosWANProxy/wan_interface/http/HTTPUIInterface.cpp
+++ b/ChaosWANProxy/wan_interface/http/HTTPUIInterface.cpp
@@ -92,7 +92,7 @@ DEFINE_CLASS_FACTORY(HTTPUIInterface, AbstractWANInterface);
HTTPUIInterface::HTTPUIInterface(const string& alias):
AbstractWANInterface(alias),
run(false),
- thread_number(0) {
+ thread_number(1),chaos_thread_number(1),sched_alloc(0) {
info = new ::driver::misc::ChaosController();
@@ -125,8 +125,15 @@ void HTTPUIInterface::init(void *init_data) throw(CException) {
thread_number = getParameter()[OPT_HTTP_THREAD_NUMBER].asInt();
}
+ if(getParameter()[OPT_CHAOS_THREAD_NUMBER].isNull() ||
+ !getParameter()[OPT_CHAOS_THREAD_NUMBER].isInt()) {
+ chaos_thread_number = 1;
+ } else {
+ chaos_thread_number = getParameter()[OPT_CHAOS_THREAD_NUMBER].asInt();
+ }
HTTWAN_INTERFACE_APP_ << "HTTP server listen on port: " << service_port;
HTTWAN_INTERFACE_APP_ << "HTTP server thread used: " << thread_number;
+ HTTWAN_INTERFACE_APP_ << "CHAOS client threads used: " << chaos_thread_number;
//allcoate each server for every thread
for(int idx = 1;
@@ -157,6 +164,11 @@ void HTTPUIInterface::init(void *init_data) throw(CException) {
}
}
if(!http_server_list.size()) throw chaos::CException(-1, "No http server has been instantiated", __PRETTY_FUNCTION__);
+ sched_cu_v.resize(chaos_thread_number);
+ for(int cnt=0;cnt<chaos_thread_number;cnt++){
+ sched_cu_v[cnt]=new ::common::misc::scheduler::Scheduler();
+ }
+ sched_alloc=0;
}
@@ -170,13 +182,19 @@ void HTTPUIInterface::start() throw(CException) {
it++) {
http_server_thread.add_thread(new boost::thread(boost::bind(&HTTPUIInterface::pollHttpServer, this, *it)));
}
- sched_cu.start();
+ for(std::vector< ::common::misc::scheduler::Scheduler*>::iterator i=sched_cu_v.begin();i!=sched_cu_v.end();i++){
+ (*i)->start();
+ }
+
}
//inherited method
void HTTPUIInterface::stop() throw(CException) {
run = false;
http_server_thread.join_all();
+ for(std::vector< ::common::misc::scheduler::Scheduler*>::iterator i=sched_cu_v.begin();i!=sched_cu_v.end();i++){
+ (*i)->stop();
+ }
}
//inherited method
@@ -189,6 +207,13 @@ void HTTPUIInterface::deinit() throw(CException) {
http_server_list.clear();
//clear the service url
service_port = 0;
+ for(int cnt=0;cnt<chaos_thread_number;cnt++){
+ if(sched_cu_v[cnt]){
+ delete (sched_cu_v[cnt]);
+ }
+ sched_cu_v[cnt]=NULL;
+
+ }
}
void HTTPUIInterface::pollHttpServer(struct mg_server *http_server) {
@@ -270,7 +295,7 @@ int HTTPUIInterface::process(struct mg_connection *connection) {
return 1;
}
char decoded[connection->content_len +2];
- mg_url_decode(connection->content, connection->content_len,decoded, connection->content_len+2,0);
+ mg_url_decode(connection->content, connection->content_len+2,decoded, connection->content_len+2,0);
std::string content_data(decoded, connection->content_len);
HTTWAN_INTERFACE_DBG_<<"POST:"<<content_data;
request=mappify(content_data);
@@ -288,7 +313,7 @@ int HTTPUIInterface::process(struct mg_connection *connection) {
if(dev_param.size()==0){
std::string ret;
if(info->get(cmd,(char*)parm.c_str(),0,atoi(cmd_prio.c_str()),atoi(cmd_schedule.c_str()),atoi(cmd_mode.c_str()),0,ret)!=::driver::misc::ChaosController::CHAOS_DEV_OK){
- HTTWAN_INTERFACE_ERR_<<"An error occurred during get:"<<info->getJsonState();
+ HTTWAN_INTERFACE_ERR_<<"An error occurred during get without dev:"<<info->getJsonState();
response.setCode(400);
} else {
@@ -301,12 +326,12 @@ int HTTPUIInterface::process(struct mg_connection *connection) {
answer_multi<<"[";
for(std::vector<std::string>::iterator idevname=dev_v.begin();idevname!=dev_v.end();idevname++){
std::string ret;
+ boost::mutex::scoped_lock l(devio_mutex);
if ((*idevname).empty() || cmd.empty()) {
continue;
}
if(devs.count(*idevname)){
- boost::mutex::scoped_lock l(devio_mutex);
controller = devs[*idevname];
@@ -314,43 +339,47 @@ int HTTPUIInterface::process(struct mg_connection *connection) {
controller = new ::driver::misc::ChaosController();
if (controller == NULL) {
- response << "{}";
- response.setCode(400);
- HTTWAN_INTERFACE_ERR_<<"error creating Chaos Controller";
- flush_response(connection, &response);
- return 1;
- }
- if(controller->init(*idevname,DEFAULT_TIMEOUT_FOR_CONTROLLER)!=0){
- response << controller->getJsonState();
- HTTWAN_INTERFACE_ERR_<<"cannot init controller for "<<*idevname;
- response << "{}";
- response.setCode(400);
- delete controller;
- flush_response(connection, &response);
- return 1;
- }
-
- addDevice(*idevname,controller);
- sched_cu.add(*idevname,controller);
-
+ response << "{}";
+ response.setCode(400);
+ HTTWAN_INTERFACE_ERR_<<"error creating Chaos Controller";
+ flush_response(connection, &response);
+ return 1;
+ }
+ if(controller->init(*idevname,DEFAULT_TIMEOUT_FOR_CONTROLLER)!=0){
+ response << controller->getJsonState();
+ HTTWAN_INTERFACE_ERR_<<"cannot init controller for "<<*idevname<<"\"";
+ // response << "{}";
+ //response.setCode(400);
+ delete controller;
+ //flush_response(connection, &response);
+ //return 1;
+ if((idevname+1) == dev_v.end()){
+ answer_multi<<"{}]";
+ }else {
+ answer_multi<<"{},";
+ }
+ continue;
+ } else{
+
+ addDevice(*idevname,controller);
+ sched_cu_v[sched_alloc++%chaos_thread_number]->add(*idevname,controller);
+ }
}
if(controller->get(cmd,(char*)parm.c_str(),0,atoi(cmd_prio.c_str()),atoi(cmd_schedule.c_str()),atoi(cmd_mode.c_str()),0,ret)!=::driver::misc::ChaosController::CHAOS_DEV_OK){
- HTTWAN_INTERFACE_ERR_<<"An error occurred during get:"<<controller->getJsonState();
- }
-
+ HTTWAN_INTERFACE_ERR_<<"An error occurred during get of:\""<<*idevname<<"\"";
+ }
if((idevname+1) == dev_v.end()){
answer_multi<<ret<<"]";
}else {
answer_multi<<ret<<",";
}
-
}
response<<answer_multi.str();
- }
+ }
diff --git a/ChaosWANProxy/wan_interface/http/HTTPUIInterface.h b/ChaosWANProxy/wan_interface/http/HTTPUIInterface.h
index 437adcae2..20e5c8db3 100644
--- a/ChaosWANProxy/wan_interface/http/HTTPUIInterface.h
+++ b/ChaosWANProxy/wan_interface/http/HTTPUIInterface.h
@@ -49,6 +49,8 @@ namespace chaos {
static const char * const OPT_HTTP_PORT = "HTTP_wi_port";
//!
static const char * const OPT_HTTP_THREAD_NUMBER = "HTTP_wi_thread_number";
+ static const char * const OPT_CHAOS_THREAD_NUMBER = "CHAOS_thread_number";
+
/*
Class that implement the Chaos RPC server using HTTP
*/
@@ -61,7 +63,7 @@ namespace chaos {
bool run;
int thread_number;
-
+ int chaos_thread_number;
ServerList http_server_list;
boost::atomic<int> thread_index;
@@ -73,7 +75,9 @@ namespace chaos {
static std::map<std::string,::driver::misc::ChaosController*> devs;
::driver::misc::ChaosController* info;
- ::common::misc::scheduler::Scheduler sched_cu;
+ std::vector< ::common::misc::scheduler::Scheduler* > sched_cu_v;
+ int sched_alloc;
+
static void addDevice(std::string,::driver::misc::ChaosController*);
static void removeDevice(std::string);
bool checkForContentType(struct mg_connection *connection,
diff --git a/ChaosWANProxy/wan_interface/http/HTTPWANInterfaceRequest.h b/ChaosWANProxy/wan_interface/http/HTTPWANInterfaceRequest.h
index efef45104..816e78c91 100644
--- a/ChaosWANProxy/wan_interface/http/HTTPWANInterfaceRequest.h
+++ b/ChaosWANProxy/wan_interface/http/HTTPWANInterfaceRequest.h
@@ -20,7 +20,7 @@
#ifndef __CHAOSFramework__HTTPWANInterfaceRequest__
#define __CHAOSFramework__HTTPWANInterfaceRequest__
-#include <mongoose.h>
+#include "mongoose.h"
namespace chaos{
namespace wan_proxy {
--
GitLab