From 3b9a07ce47bc54918f10eea58bfb1cd071ef5cb2 Mon Sep 17 00:00:00 2001 From: evanc Date: Tue, 26 Sep 2017 08:30:19 -0700 Subject: [PATCH] Fix the unit test failure, enabled cancellation token for classes, added version info --- nodemanager/Version.h | 7 +++++ nodemanager/core/HostsManager.cpp | 2 +- nodemanager/core/HostsManager.h | 2 +- nodemanager/core/HttpFetcher.cpp | 2 +- nodemanager/core/HttpFetcher.h | 3 +-- nodemanager/core/HttpReporter.cpp | 3 ++- nodemanager/core/HttpReporter.h | 3 +-- nodemanager/core/MetricCollectorBase.cpp | 4 +-- nodemanager/core/MetricCollectorBase.h | 2 +- nodemanager/core/Monitor.cpp | 9 ++++--- nodemanager/core/Monitor.h | 4 +-- nodemanager/core/NamingClient.cpp | 34 +++++++++++++++--------- nodemanager/core/NamingClient.h | 4 +-- nodemanager/core/NodeManagerConfig.h | 26 +++++++++--------- nodemanager/core/RemoteCommunicator.cpp | 4 +++ nodemanager/core/RemoteExecutor.cpp | 16 +++++------ nodemanager/core/RemoteExecutor.h | 10 ++++++- nodemanager/core/Reporter.h | 19 +++++++------ nodemanager/core/UdpReporter.cpp | 4 +-- nodemanager/core/UdpReporter.h | 2 +- 20 files changed, 94 insertions(+), 66 deletions(-) diff --git a/nodemanager/Version.h b/nodemanager/Version.h index 5abc13d..799770d 100644 --- a/nodemanager/Version.h +++ b/nodemanager/Version.h @@ -567,6 +567,13 @@ namespace hpc "Added task completion uri configuration", } }, + { "2.3.1.0", + { + "Added docker support", + "Added support to peek the task output", + "Fixed the unit test failure", + } + }, }; return versionHistory; diff --git a/nodemanager/core/HostsManager.cpp b/nodemanager/core/HostsManager.cpp index e1de5ff..55e51e0 100644 --- a/nodemanager/core/HostsManager.cpp +++ b/nodemanager/core/HostsManager.cpp @@ -12,7 +12,7 @@ using namespace hpc::data; using namespace web::http; using namespace hpc::core; -HostsManager::HostsManager(std::function getHostsUri, int fetchInterval) +HostsManager::HostsManager(std::function getHostsUri, int fetchInterval) { this->hostsFetcher = std::unique_ptr( diff --git a/nodemanager/core/HostsManager.h b/nodemanager/core/HostsManager.h index fca1316..dee1cca 100644 --- a/nodemanager/core/HostsManager.h +++ b/nodemanager/core/HostsManager.h @@ -18,7 +18,7 @@ namespace hpc const std::string HPCHostEntryPattern = R"delimiter(^([0-9\.]+)\s+([^\s#]+)\s+#HPC\s*$)delimiter"; const std::string UpdateIdHeaderName = "UpdateId"; - HostsManager(std::function getHostsUri, int fetchInterval); + HostsManager(std::function getHostsUri, int fetchInterval); ~HostsManager() { this->Stop(); } void Start() { this->hostsFetcher->Start(); } diff --git a/nodemanager/core/HttpFetcher.cpp b/nodemanager/core/HttpFetcher.cpp index 1bdd4fa..a7f8e96 100644 --- a/nodemanager/core/HttpFetcher.cpp +++ b/nodemanager/core/HttpFetcher.cpp @@ -12,7 +12,7 @@ int HttpFetcher::Report() std::string uri; try { - uri = this->getReportUri(); + uri = this->getReportUri(this->cts.get_token()); auto client = HttpHelper::GetHttpClient(uri); auto request = HttpHelper::GetHttpRequest(methods::GET); diff --git a/nodemanager/core/HttpFetcher.h b/nodemanager/core/HttpFetcher.h index 97124cf..2f74fd3 100644 --- a/nodemanager/core/HttpFetcher.h +++ b/nodemanager/core/HttpFetcher.h @@ -17,7 +17,7 @@ namespace hpc public: HttpFetcher( const std::string& name, - std::function getReportUri, + std::function getReportUri, int hold, int interval, std::function requestHandler, @@ -40,7 +40,6 @@ namespace hpc private: std::function requestHandler; std::function responseHandler; - pplx::cancellation_token_source cts; }; } } diff --git a/nodemanager/core/HttpReporter.cpp b/nodemanager/core/HttpReporter.cpp index c227532..561f74d 100644 --- a/nodemanager/core/HttpReporter.cpp +++ b/nodemanager/core/HttpReporter.cpp @@ -13,8 +13,9 @@ int HttpReporter::Report() try { - uri = this->getReportUri(); + uri = this->getReportUri(this->cts.get_token()); + if (this->cts.get_token().is_canceled()) return -1; auto jsonBody = this->valueFetcher(); if (jsonBody.is_null()) { diff --git a/nodemanager/core/HttpReporter.h b/nodemanager/core/HttpReporter.h index bac7270..9ea1dba 100644 --- a/nodemanager/core/HttpReporter.h +++ b/nodemanager/core/HttpReporter.h @@ -18,7 +18,7 @@ namespace hpc public: HttpReporter( const std::string& reporterName, - std::function getUri, + std::function getUri, int hold, int interval, std::function fetcher, @@ -37,7 +37,6 @@ namespace hpc protected: private: - pplx::cancellation_token_source cts; }; } } diff --git a/nodemanager/core/MetricCollectorBase.cpp b/nodemanager/core/MetricCollectorBase.cpp index 2a540ee..42dfcac 100644 --- a/nodemanager/core/MetricCollectorBase.cpp +++ b/nodemanager/core/MetricCollectorBase.cpp @@ -2,7 +2,7 @@ using namespace hpc::core; -void MetricCollectorBase::ApplyConfig(const MetricCounter& config) +void MetricCollectorBase::ApplyConfig(const MetricCounter& config, pplx::cancellation_token token) { this->metricId = config.MetricId; @@ -15,7 +15,7 @@ void MetricCollectorBase::ApplyConfig(const MetricCounter& config) if (!instanceNames.empty()) { - auto client = HttpHelper::GetHttpClient(NodeManagerConfig::ResolveMetricInstanceIdsUri()); + auto client = HttpHelper::GetHttpClient(NodeManagerConfig::ResolveMetricInstanceIdsUri(token)); json::value jsonBody = JsonHelper>::ToJson(instanceNames); auto request = HttpHelper::GetHttpRequest(web::http::methods::POST, jsonBody); diff --git a/nodemanager/core/MetricCollectorBase.h b/nodemanager/core/MetricCollectorBase.h index 1eb160f..60876c1 100644 --- a/nodemanager/core/MetricCollectorBase.h +++ b/nodemanager/core/MetricCollectorBase.h @@ -27,7 +27,7 @@ namespace hpc MetricCollectorBase() = default; - void ApplyConfig(const MetricCounter& config); + void ApplyConfig(const MetricCounter& config, pplx::cancellation_token token); void Reset() { diff --git a/nodemanager/core/Monitor.cpp b/nodemanager/core/Monitor.cpp index 6190bb8..64f41a7 100644 --- a/nodemanager/core/Monitor.cpp +++ b/nodemanager/core/Monitor.cpp @@ -333,6 +333,7 @@ Monitor::~Monitor() { if (this->threadId != 0) { + // todo: graceful exit the thread. pthread_cancel(this->threadId); pthread_join(this->threadId, nullptr); } @@ -345,7 +346,7 @@ void Monitor::SetNodeUuid(const uuid& id) this->packet.Uuid.AssignFrom(id); } -void Monitor::ApplyMetricConfig(MetricCountersConfig&& config) +void Monitor::ApplyMetricConfig(MetricCountersConfig&& config, pplx::cancellation_token token) { WriterLock writerLock(&this->lock); @@ -353,7 +354,7 @@ void Monitor::ApplyMetricConfig(MetricCountersConfig&& config) for (auto& counter : config.MetricCounters) { - if (!this->EnableMetricCounter(counter)) + if (!this->EnableMetricCounter(counter, token)) { Logger::Debug("Disabled counter MetricId {0}, InstanceId {1}, InstanceName {2} Path {3}", counter.MetricId, counter.InstanceId, counter.InstanceName, counter.Path); } @@ -364,12 +365,12 @@ void Monitor::ApplyMetricConfig(MetricCountersConfig&& config) } } -bool Monitor::EnableMetricCounter(const MetricCounter& counterConfig) +bool Monitor::EnableMetricCounter(const MetricCounter& counterConfig, pplx::cancellation_token token) { auto collector = this->collectors.find(counterConfig.Path); if (collector != this->collectors.end()) { - collector->second->ApplyConfig(counterConfig); + collector->second->ApplyConfig(counterConfig, token); return true; } diff --git a/nodemanager/core/Monitor.h b/nodemanager/core/Monitor.h index 0d91382..27eee63 100644 --- a/nodemanager/core/Monitor.h +++ b/nodemanager/core/Monitor.h @@ -32,11 +32,11 @@ namespace hpc json::value GetRegisterInfo(); void SetNodeUuid(const uuid& id); - void ApplyMetricConfig(hpc::arguments::MetricCountersConfig&& config); + void ApplyMetricConfig(hpc::arguments::MetricCountersConfig&& config, pplx::cancellation_token token); protected: private: - bool EnableMetricCounter(const hpc::arguments::MetricCounter& counterConfig); + bool EnableMetricCounter(const hpc::arguments::MetricCounter& counterConfig, pplx::cancellation_token token); void Run(); static void* MonitoringThread(void* arg); diff --git a/nodemanager/core/NamingClient.cpp b/nodemanager/core/NamingClient.cpp index d887eca..aaa8fc3 100644 --- a/nodemanager/core/NamingClient.cpp +++ b/nodemanager/core/NamingClient.cpp @@ -23,7 +23,7 @@ void NamingClient::InvalidateCache() } } -std::string NamingClient::GetServiceLocation(const std::string& serviceName) +std::string NamingClient::GetServiceLocation(const std::string& serviceName, pplx::cancellation_token token) { std::map::iterator location; @@ -34,20 +34,28 @@ std::string NamingClient::GetServiceLocation(const std::string& serviceName) if (location == this->serviceLocations.end()) { - WriterLock writerLock(&this->lock); - Logger::Debug("ResolveServiceLocation> there is no entry for {0}", serviceName); - for (auto kvp : this->serviceLocations) { - Logger::Debug("ResolveServiceLocation> entry {0} = {1}", kvp.first, kvp.second); + ReaderLock readerLock(&this->lock); + Logger::Debug("ResolveServiceLocation> there is no entry for {0}", serviceName); + for (auto kvp : this->serviceLocations) + { + Logger::Debug("ResolveServiceLocation> entry {0} = {1}", kvp.first, kvp.second); + } + + location = this->serviceLocations.find(serviceName); } - location = this->serviceLocations.find(serviceName); if (location == this->serviceLocations.end()) { std::string temp; - this->RequestForServiceLocation(serviceName, temp); - this->serviceLocations[serviceName] = temp; - location = this->serviceLocations.find(serviceName); + this->RequestForServiceLocation(serviceName, temp, token); + + { + WriterLock writerLock(&this->lock); + + this->serviceLocations[serviceName] = temp; + location = this->serviceLocations.find(serviceName); + } } } @@ -55,13 +63,13 @@ std::string NamingClient::GetServiceLocation(const std::string& serviceName) return location->second; } -void NamingClient::RequestForServiceLocation(const std::string& serviceName, std::string& serviceLocation) +void NamingClient::RequestForServiceLocation(const std::string& serviceName, std::string& serviceLocation, pplx::cancellation_token token) { int selected = rand() % this->namingServicesUri.size(); std::string uri; int interval = this->intervalSeconds; - while (true) + while (!token.is_canceled()) { try { @@ -71,7 +79,7 @@ void NamingClient::RequestForServiceLocation(const std::string& serviceName, std auto client = HttpHelper::GetHttpClient(uri); auto request = HttpHelper::GetHttpRequest(methods::GET); - http_response response = client->request(*request, this->cts.get_token()).get(); + http_response response = client->request(*request, token).get(); if (response.status_code() == http::status_codes::OK) { serviceLocation = JsonHelper::FromJson(response.extract_json().get()); @@ -96,6 +104,8 @@ void NamingClient::RequestForServiceLocation(const std::string& serviceName, std Logger::Error("ResolveServiceLocation> Unknown error occurred when fetching from {0}", uri); } + if (token.is_canceled()) break; + sleep(interval); interval *= 2; if (interval > 300) interval = 300; diff --git a/nodemanager/core/NamingClient.h b/nodemanager/core/NamingClient.h index 8b6e2ba..f70d5a0 100644 --- a/nodemanager/core/NamingClient.h +++ b/nodemanager/core/NamingClient.h @@ -35,12 +35,12 @@ namespace hpc return instance; } - std::string GetServiceLocation(const std::string& serviceName); + std::string GetServiceLocation(const std::string& serviceName, pplx::cancellation_token token); static void InvalidateCache(); private: - void RequestForServiceLocation(const std::string& serviceName, std::string& serviceLocation); + void RequestForServiceLocation(const std::string& serviceName, std::string& serviceLocation, pplx::cancellation_token token); int intervalSeconds; std::map serviceLocations; diff --git a/nodemanager/core/NodeManagerConfig.h b/nodemanager/core/NodeManagerConfig.h index 755e1a5..90501c4 100644 --- a/nodemanager/core/NodeManagerConfig.h +++ b/nodemanager/core/NodeManagerConfig.h @@ -43,38 +43,39 @@ namespace hpc AddConfigurationItem(std::string, MetricUri); AddConfigurationItem(std::string, HeartbeatUri); AddConfigurationItem(std::string, TaskCompletionUri); + AddConfigurationItem(std::string, HostsFileUri); - static std::string ResolveRegisterUri() + static std::string ResolveRegisterUri(pplx::cancellation_token token) { std::string uri = NodeManagerConfig::GetRegisterUri(); - return ResolveUri(uri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName()); }); + return ResolveUri(uri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName(), token); }); } - static std::string ResolveHeartbeatUri() + static std::string ResolveHeartbeatUri(pplx::cancellation_token token) { std::string uri = NodeManagerConfig::GetHeartbeatUri(); - return ResolveUri(uri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName()); }); + return ResolveUri(uri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName(), token); }); } - static std::string ResolveMetricUri() + static std::string ResolveMetricUri(pplx::cancellation_token token) { std::string uri = NodeManagerConfig::GetMetricUri(); - return ResolveUri(uri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetUdpMetricServiceName()); }); + return ResolveUri(uri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetUdpMetricServiceName(), token); }); } - static std::string ResolveHostsFileUri() + static std::string ResolveHostsFileUri(pplx::cancellation_token token) { std::string uri = NodeManagerConfig::GetHostsFileUri(); - return ResolveUri(uri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName()); }); + return ResolveUri(uri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName(), token); }); } - static std::string ResolveMetricInstanceIdsUri() + static std::string ResolveMetricInstanceIdsUri(pplx::cancellation_token token) { std::string uri = NodeManagerConfig::GetMetricInstanceIdsUri(); - return ResolveUri(uri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName()); }); + return ResolveUri(uri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName(), token); }); } - static std::string ResolveTaskCompletedUri(const std::string& uri) + static std::string ResolveTaskCompletedUri(const std::string& uri, pplx::cancellation_token token) { std::string configUri = NodeManagerConfig::GetTaskCompletionUri(); if (configUri.empty()) @@ -82,13 +83,12 @@ namespace hpc configUri = uri; } - return ResolveUri(configUri, [](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName()); }); + return ResolveUri(configUri, [token](std::shared_ptr namingClient) { return namingClient->GetServiceLocation(NodeManagerConfig::GetDefaultServiceName(), token); }); } protected: private: AddConfigurationItem(std::string, RegisterUri); - AddConfigurationItem(std::string, HostsFileUri); AddConfigurationItem(std::string, MetricInstanceIdsUri); static std::string ResolveUri(const std::string& uri, std::function)> resolver) diff --git a/nodemanager/core/RemoteCommunicator.cpp b/nodemanager/core/RemoteCommunicator.cpp index c9c46e9..da4c722 100644 --- a/nodemanager/core/RemoteCommunicator.cpp +++ b/nodemanager/core/RemoteCommunicator.cpp @@ -76,7 +76,10 @@ void RemoteCommunicator::Close() { try { + Logger::Info("Closing the communicator {0}", this->listener.uri().to_string().c_str()); this->listener.close().wait(); + Logger::Info("Closed the communicator {0}", this->listener.uri().to_string().c_str()); + this->isListening = false; } catch (const std::exception& ex) @@ -210,6 +213,7 @@ void RemoteCommunicator::HandlePost(http_request request) { try { + Logger::Info("Replied with content {0}", t.get()); request.reply(status_codes::OK, t.get()).then([](auto t) { IsError(t); }); } catch (const web::http::http_exception& httpEx) diff --git a/nodemanager/core/RemoteExecutor.cpp b/nodemanager/core/RemoteExecutor.cpp index 4975338..a757cfe 100644 --- a/nodemanager/core/RemoteExecutor.cpp +++ b/nodemanager/core/RemoteExecutor.cpp @@ -30,7 +30,7 @@ RemoteExecutor::RemoteExecutor(const std::string& networkName) std::unique_ptr>( new HttpReporter( "RegisterReporter", - []() { return NodeManagerConfig::ResolveRegisterUri(); }, + [](pplx::cancellation_token token) { return NodeManagerConfig::ResolveRegisterUri(token); }, 3, this->RegisterInterval, [this]() { return this->monitor.GetRegisterInfo(); }, @@ -534,7 +534,7 @@ void RemoteExecutor::ReportTaskCompletion( { if (!jsonBody.is_null()) { - std::string uri = NodeManagerConfig::ResolveTaskCompletedUri(callbackUri); + std::string uri = NodeManagerConfig::ResolveTaskCompletedUri(callbackUri, this->cts.get_token()); Logger::Debug(jobId, taskId, taskRequeueCount, "Callback to {0} with {1}", uri, jsonBody); @@ -579,7 +579,7 @@ void RemoteExecutor::StartHeartbeat() std::unique_ptr>( new HttpReporter( "HeartbeatReporter", - []() { return NodeManagerConfig::ResolveHeartbeatUri(); }, + [](pplx::cancellation_token token) { return NodeManagerConfig::ResolveHeartbeatUri(token); }, 0, this->NodeInfoReportInterval, [this]() { return this->jobTaskTable.ToJson(); }, @@ -590,7 +590,7 @@ void RemoteExecutor::StartHeartbeat() void RemoteExecutor::StartHostsManager() { - std::string hostsUri = NodeManagerConfig::ResolveHostsFileUri(); + std::string hostsUri = NodeManagerConfig::GetHostsFileUri(); if (!hostsUri.empty()) { int interval = this->DefaultHostsFetchInterval; @@ -613,7 +613,7 @@ void RemoteExecutor::StartHostsManager() WriterLock writerLock(&this->lock); - this->hostsManager = std::unique_ptr(new HostsManager([]() { return NodeManagerConfig::ResolveHostsFileUri(); }, interval)); + this->hostsManager = std::unique_ptr(new HostsManager([](pplx::cancellation_token token) { return NodeManagerConfig::ResolveHostsFileUri(token); }, interval)); this->hostsManager->Start(); } else @@ -639,7 +639,7 @@ void RemoteExecutor::StartMetric() { WriterLock writerLock(&this->lock); - std::string uri = NodeManagerConfig::ResolveMetricUri(); + std::string uri = NodeManagerConfig::GetMetricUri(); if (!uri.empty()) { auto tokens = String::Split(uri, '/'); @@ -651,7 +651,7 @@ void RemoteExecutor::StartMetric() std::unique_ptr>>( new UdpReporter( "MetricReporter", - []() { return NodeManagerConfig::ResolveMetricUri(); }, + [](pplx::cancellation_token token) { return NodeManagerConfig::ResolveMetricUri(token); }, 0, this->MetricReportInterval, [this]() { return this->monitor.GetMonitorPacketData(); }, @@ -681,7 +681,7 @@ pplx::task RemoteExecutor::MetricConfig( { this->Metric(std::move(callbackUri)); - this->monitor.ApplyMetricConfig(std::move(config)); + this->monitor.ApplyMetricConfig(std::move(config), this->cts.get_token()); return pplx::task_from_result(json::value()); } diff --git a/nodemanager/core/RemoteExecutor.h b/nodemanager/core/RemoteExecutor.h index 29aa286..226b75a 100644 --- a/nodemanager/core/RemoteExecutor.h +++ b/nodemanager/core/RemoteExecutor.h @@ -21,7 +21,13 @@ namespace hpc { public: RemoteExecutor(const std::string& networkName); - ~RemoteExecutor() { pthread_rwlock_destroy(&this->lock); } + ~RemoteExecutor() + { + Logger::Info("Closing the Remote Executor."); + this->cts.cancel(); + pthread_rwlock_destroy(&this->lock); + Logger::Info("Closed the Remote Executor."); + } virtual pplx::task StartJobAndTask(hpc::arguments::StartJobAndTaskArgs&& args, std::string&& callbackUri); virtual pplx::task StartTask(hpc::arguments::StartTaskArgs&& args, std::string&& callbackUri); @@ -67,6 +73,8 @@ namespace hpc std::map> jobUsers; std::map> userJobs; pthread_rwlock_t lock; + + pplx::cancellation_token_source cts; }; } } diff --git a/nodemanager/core/Reporter.h b/nodemanager/core/Reporter.h index 07caa62..a894c64 100644 --- a/nodemanager/core/Reporter.h +++ b/nodemanager/core/Reporter.h @@ -17,7 +17,7 @@ namespace hpc class Reporter { public: - Reporter(std::string reporterName, std::function getUri, int hold, int interval, std::function fetcher, std::function onErrorFunc) + Reporter(std::string reporterName, std::function getUri, int hold, int interval, std::function fetcher, std::function onErrorFunc) : name(reporterName), getReportUri(getUri), valueFetcher(fetcher), onError(onErrorFunc), intervalSeconds(interval), holdSeconds(hold) { } @@ -32,36 +32,35 @@ namespace hpc void Stop() { + Logger::Debug("Stopping the thread of Reporter {0}", this->name); this->isRunning = false; + this->cts.cancel(); if (this->threadId != 0) { while (this->inRequest) usleep(1); - pthread_cancel(this->threadId); pthread_join(this->threadId, nullptr); - Logger::Debug("Destructed Reporter {0}", this->name); + Logger::Debug("Stopped the thread of Reporter {0}", this->name); } } virtual ~Reporter() { - Logger::Debug("Destruct Reporter {0}", this->name); + Logger::Debug("Destructed Reporter {0}", this->name); } virtual int Report() = 0; protected: std::string name; - std::function getReportUri; + std::function getReportUri; std::function valueFetcher; std::function onError; int intervalSeconds; + pplx::cancellation_token_source cts; private: static void* ReportingThread(void* arg) { - pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, nullptr); - pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, nullptr); - Reporter* r = static_cast(arg); sleep(r->holdSeconds); @@ -82,10 +81,10 @@ namespace hpc r->inRequest = false; } - sleep(needRetry ? r->ErrorRetrySeconds : r->intervalSeconds); + if (r->isRunning) sleep(needRetry ? r->ErrorRetrySeconds : r->intervalSeconds); } - pthread_exit(nullptr); + return nullptr; } const int ErrorRetrySeconds = 2; diff --git a/nodemanager/core/UdpReporter.cpp b/nodemanager/core/UdpReporter.cpp index 86b901e..68bbca0 100644 --- a/nodemanager/core/UdpReporter.cpp +++ b/nodemanager/core/UdpReporter.cpp @@ -12,7 +12,7 @@ using namespace hpc::utils; UdpReporter::UdpReporter( const std::string& name, - std::function getReportUri, + std::function getReportUri, int hold, int interval, std::function()> fetcher, @@ -32,7 +32,7 @@ void UdpReporter::ReConnect() try { - uri = this->getReportUri(); + uri = this->getReportUri(this->cts.get_token()); } catch (const http::http_exception& httpEx) { diff --git a/nodemanager/core/UdpReporter.h b/nodemanager/core/UdpReporter.h index bca75d9..58d7798 100644 --- a/nodemanager/core/UdpReporter.h +++ b/nodemanager/core/UdpReporter.h @@ -15,7 +15,7 @@ namespace hpc public: UdpReporter( const std::string& name, - std::function getReportUri, + std::function getReportUri, int hold, int interval, std::function()> fetcher,