zhangxuetong
4 years ago
parent
3952a57d85
commit
62c0577d80
@ -1,49 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 MINDSPORE_CORE_MINDRT_INCLUDE_ASYNC_ASYNCAFTER_H
|
||||
#define MINDSPORE_CORE_MINDRT_INCLUDE_ASYNC_ASYNCAFTER_H
|
||||
|
||||
#include "async/async.h"
|
||||
|
||||
#include "timer/timertools.h"
|
||||
|
||||
constexpr mindspore::Duration MILLISECONDS = 1;
|
||||
constexpr mindspore::Duration SECONDS = 1000;
|
||||
|
||||
namespace mindspore {
|
||||
template <typename T>
|
||||
Timer AsyncAfter(const Duration &duration, const AID &aid, void (T::*method)()) {
|
||||
return TimerTools::AddTimer(duration, aid, [=]() { Async(aid, method); });
|
||||
}
|
||||
|
||||
template <typename T, typename Arg0, typename Arg1>
|
||||
Timer AsyncAfter(const Duration &duration, const AID &aid, void (T::*method)(Arg0), Arg1 &&arg) {
|
||||
return TimerTools::AddTimer(duration, aid, [=]() { Async(aid, method, arg); });
|
||||
}
|
||||
|
||||
template <typename T, typename... Args0, typename... Args1>
|
||||
Timer AsyncAfter(const Duration &duration, const AID &aid, void (T::*method)(Args0...), Args1 &&... args) {
|
||||
std::function<void(Args0...)> f([=](Args0... args0) { Async(aid, method, args0...); });
|
||||
|
||||
auto handler = std::bind(f, args...);
|
||||
|
||||
return TimerTools::AddTimer(duration, aid, [=]() { Async(aid, std::move(handler)); });
|
||||
}
|
||||
|
||||
}; // namespace mindspore
|
||||
|
||||
#endif
|
||||
@ -1,32 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 __LITEBUS_DURATION_HPP__
|
||||
#define __LITEBUS_DURATION_HPP__
|
||||
|
||||
#include <deque>
|
||||
#include <functional>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
|
||||
#include "async/spinlock.h"
|
||||
|
||||
namespace mindspore {
|
||||
using Duration = uint64_t;
|
||||
}
|
||||
#endif
|
||||
@ -1,48 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 __LITEBUS_TIMER_HPP__
|
||||
#define __LITEBUS_TIMER_HPP__
|
||||
|
||||
#include "actor/aid.h"
|
||||
#include "timer/timewatch.h"
|
||||
|
||||
namespace mindspore {
|
||||
class Timer {
|
||||
public:
|
||||
Timer();
|
||||
~Timer();
|
||||
bool operator==(const Timer &that) const;
|
||||
// run this timer's thunk.
|
||||
void operator()() const;
|
||||
TimeWatch GetTimeWatch() const;
|
||||
AID GetTimerAID() const;
|
||||
uint64_t GetTimerID() const;
|
||||
|
||||
private:
|
||||
friend class TimerTools;
|
||||
|
||||
Timer(uint64_t timerId, const TimeWatch &timeWatch, const AID &timeAid, const std::function<void()> &handler);
|
||||
|
||||
uint64_t id;
|
||||
TimeWatch t;
|
||||
AID aid;
|
||||
std::function<void()> thunk;
|
||||
};
|
||||
|
||||
} // namespace mindspore
|
||||
|
||||
#endif
|
||||
@ -1,39 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 __LITEBUS_TIMETOOLS_HPP__
|
||||
#define __LITEBUS_TIMETOOLS_HPP__
|
||||
|
||||
#include <atomic>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <set>
|
||||
|
||||
#include "timer/duration.h"
|
||||
#include "timer/timer.h"
|
||||
|
||||
namespace mindspore {
|
||||
class TimerTools {
|
||||
public:
|
||||
static bool Initialize();
|
||||
static void Finalize();
|
||||
static Timer AddTimer(const Duration &duration, const AID &aid, const std::function<void()> &thunk);
|
||||
static bool Cancel(const Timer &timer);
|
||||
static std::atomic_bool g_initStatus;
|
||||
};
|
||||
} // namespace mindspore
|
||||
|
||||
#endif
|
||||
@ -1,65 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 __LITEBUS_TIMEWATCH_HPP__
|
||||
#define __LITEBUS_TIMEWATCH_HPP__
|
||||
|
||||
#include "timer/duration.h"
|
||||
namespace mindspore {
|
||||
constexpr Duration MICRTONANO = 1000;
|
||||
constexpr Duration MILLITOMICR = 1000;
|
||||
constexpr Duration SECTOMILLI = 1000;
|
||||
|
||||
class TimeWatch {
|
||||
public:
|
||||
TimeWatch();
|
||||
|
||||
TimeWatch(const Duration &duration);
|
||||
|
||||
TimeWatch(const TimeWatch &that);
|
||||
~TimeWatch();
|
||||
|
||||
// Constructs a Time instance that is the 'duration' from now.
|
||||
static TimeWatch In(const Duration &duration);
|
||||
|
||||
static Duration Now();
|
||||
|
||||
TimeWatch &operator=(const TimeWatch &that);
|
||||
|
||||
TimeWatch &operator=(const Duration &duration);
|
||||
|
||||
bool operator==(const TimeWatch &that) const;
|
||||
|
||||
bool operator<(const TimeWatch &that) const;
|
||||
|
||||
bool operator<=(const TimeWatch &that) const;
|
||||
|
||||
// Returns the value of the timewatch as a Duration object.
|
||||
Duration Time() const;
|
||||
|
||||
// Returns the amount of time remaining.
|
||||
Duration Remaining() const;
|
||||
|
||||
// return true if the time expired.
|
||||
bool Expired() const;
|
||||
|
||||
private:
|
||||
Duration duration;
|
||||
};
|
||||
|
||||
} // namespace mindspore
|
||||
|
||||
#endif
|
||||
@ -1,85 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 <actor/sysmgr_actor.h>
|
||||
#include "actor/actormgr.h"
|
||||
#include "actor/iomgr.h"
|
||||
#include "utils/log_adapter.h"
|
||||
|
||||
namespace mindspore {
|
||||
|
||||
Duration SysMgrActor::linkRecycleDuration = 10000;
|
||||
|
||||
void MetricsMessage::PrintMetrics() {
|
||||
// print sendMetrics by default, in the future we can add more metrics format
|
||||
std::ostringstream out;
|
||||
|
||||
while (!intTypeMetrics.empty()) {
|
||||
out << intTypeMetrics.front() << "-";
|
||||
intTypeMetrics.pop();
|
||||
}
|
||||
|
||||
out << "|";
|
||||
|
||||
while (!stringTypeMetrics.empty()) {
|
||||
std::string stringMetric = stringTypeMetrics.front();
|
||||
if (stringMetric.empty()) {
|
||||
out << "null"
|
||||
<< "-";
|
||||
} else {
|
||||
out << stringMetric << "-";
|
||||
}
|
||||
|
||||
stringTypeMetrics.pop();
|
||||
}
|
||||
|
||||
MS_LOG(INFO) << "[format:fd-err-sum-size|to-okmsg-failmsg], value:" << out.str().c_str();
|
||||
}
|
||||
|
||||
void SysMgrActor::SendMetricsDurationCallback() {
|
||||
std::string protocol = "tcp";
|
||||
std::shared_ptr<mindspore::IOMgr> ioMgrRef = ActorMgr::GetIOMgrRef(protocol);
|
||||
if (ioMgrRef == nullptr) {
|
||||
MS_LOG(INFO) << "tcp protocol is not exist.";
|
||||
} else {
|
||||
ioMgrRef->CollectMetrics();
|
||||
}
|
||||
|
||||
(void)AsyncAfter(printSendMetricsDuration, GetAID(), &SysMgrActor::SendMetricsDurationCallback);
|
||||
}
|
||||
|
||||
void SysMgrActor::HandleSendMetricsCallback(const AID &from, std::unique_ptr<MetricsMessage> message) {
|
||||
if (message == nullptr) {
|
||||
MS_LOG(WARNING) << "Can't transform to MetricsMessage.";
|
||||
return;
|
||||
}
|
||||
message->PrintMetrics();
|
||||
return;
|
||||
}
|
||||
|
||||
void SysMgrActor::LinkRecycleDurationCallback() {
|
||||
std::string protocol = "tcp";
|
||||
std::shared_ptr<mindspore::IOMgr> ioMgrRef = ActorMgr::GetIOMgrRef(protocol);
|
||||
if (ioMgrRef == nullptr) {
|
||||
MS_LOG(INFO) << "tcp protocol is not exist.";
|
||||
} else {
|
||||
ioMgrRef->LinkRecycleCheck(linkRecyclePeriod);
|
||||
}
|
||||
|
||||
(void)AsyncAfter(linkRecycleDuration, GetAID(), &SysMgrActor::LinkRecycleDurationCallback);
|
||||
}
|
||||
|
||||
} // namespace mindspore
|
||||
@ -1,93 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 MINDSPORE_CORE_MINDRT_SRC_ACTOR_SYSMGR_ACTOR_H
|
||||
#define MINDSPORE_CORE_MINDRT_SRC_ACTOR_SYSMGR_ACTOR_H
|
||||
|
||||
#include <queue>
|
||||
|
||||
#include "async/async.h"
|
||||
#include "async/asyncafter.h"
|
||||
#include "actor/actorapp.h"
|
||||
#include "utils/log_adapter.h"
|
||||
|
||||
namespace mindspore {
|
||||
|
||||
const std::string SYSMGR_ACTOR_NAME = "SysMgrActor";
|
||||
const std::string METRICS_SEND_MSGNAME = "SendMetrics";
|
||||
const int LINK_RECYCLE_PERIOD_MIN = 20;
|
||||
const int LINK_RECYCLE_PERIOD_MAX = 360;
|
||||
|
||||
using IntTypeMetrics = std::queue<int>;
|
||||
using StringTypeMetrics = std::queue<std::string>;
|
||||
|
||||
class MetricsMessage : public MessageBase {
|
||||
public:
|
||||
explicit MetricsMessage(const std::string &tfrom, const std::string &tTo, const std::string &tName,
|
||||
const IntTypeMetrics &tInts = IntTypeMetrics(),
|
||||
const StringTypeMetrics &tStrings = StringTypeMetrics())
|
||||
: MessageBase(tfrom, tTo, tName), intTypeMetrics(tInts), stringTypeMetrics(tStrings) {}
|
||||
|
||||
~MetricsMessage() override {}
|
||||
|
||||
void PrintMetrics();
|
||||
|
||||
private:
|
||||
IntTypeMetrics intTypeMetrics;
|
||||
StringTypeMetrics stringTypeMetrics;
|
||||
};
|
||||
|
||||
class SysMgrActor : public mindspore::AppActor {
|
||||
public:
|
||||
explicit SysMgrActor(const std::string &name, const Duration &duration)
|
||||
: mindspore::AppActor(name), printSendMetricsDuration(duration) {
|
||||
linkRecyclePeriod = 0;
|
||||
}
|
||||
|
||||
~SysMgrActor() override {}
|
||||
|
||||
protected:
|
||||
virtual void Init() override {
|
||||
MS_LOG(INFO) << "Initiaize SysMgrActor";
|
||||
// register receive handle
|
||||
Receive("SendMetrics", &SysMgrActor::HandleSendMetricsCallback);
|
||||
|
||||
// start sys manager timers
|
||||
(void)AsyncAfter(printSendMetricsDuration, GetAID(), &SysMgrActor::SendMetricsDurationCallback);
|
||||
|
||||
char *linkRecycleEnv = getenv("LITEBUS_LINK_RECYCLE_PERIOD");
|
||||
if (linkRecycleEnv != nullptr) {
|
||||
int period = 0;
|
||||
period = std::stoi(linkRecycleEnv);
|
||||
if (period >= LINK_RECYCLE_PERIOD_MIN && period <= LINK_RECYCLE_PERIOD_MAX) {
|
||||
MS_LOG(INFO) << "link recycle set:" << period;
|
||||
linkRecyclePeriod = period;
|
||||
(void)AsyncAfter(linkRecycleDuration, GetAID(), &SysMgrActor::LinkRecycleDurationCallback);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
void SendMetricsDurationCallback();
|
||||
void HandleSendMetricsCallback(const AID &from, std::unique_ptr<MetricsMessage> message);
|
||||
void LinkRecycleDurationCallback();
|
||||
Duration printSendMetricsDuration;
|
||||
static Duration linkRecycleDuration;
|
||||
int linkRecyclePeriod;
|
||||
};
|
||||
|
||||
} // namespace mindspore
|
||||
#endif
|
||||
@ -1,3 +0,0 @@
|
||||
target_sources(litebus_obj PRIVATE
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/evloop.cc
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,109 +0,0 @@
|
||||
/**
|
||||
* Copyright 2021 Huawei Technologies Co., Ltd
|
||||
*
|
||||
* 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 __LITEBUS_EVLOOP_H__
|
||||
#define __LITEBUS_EVLOOP_H__
|
||||
|
||||
#include <sys/epoll.h>
|
||||
#include <sys/ioctl.h>
|
||||
#include <functional>
|
||||
#include <list>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <sys/eventfd.h>
|
||||
#include <semaphore.h>
|
||||
#include "timer/duration.h"
|
||||
|
||||
namespace mindspore {
|
||||
|
||||
/*
|
||||
* max epoll set size
|
||||
*/
|
||||
constexpr auto EPOLL_SIZE = 4096;
|
||||
|
||||
/*
|
||||
* epoll event max size
|
||||
*/
|
||||
constexpr auto EPOLL_EVENTS_SIZE = 64;
|
||||
|
||||
typedef void (*EventHandler)(int fd, uint32_t events, void *data);
|
||||
|
||||
typedef struct EventData {
|
||||
EventHandler handler;
|
||||
void *data;
|
||||
int fd;
|
||||
|
||||
} EventData;
|
||||
|
||||
class EvLoop {
|
||||
public:
|
||||
EvLoop() {
|
||||
efd = -1;
|
||||
stopLoop = 0;
|
||||
queueEventfd = -1;
|
||||
loopThread = 0;
|
||||
};
|
||||
EvLoop(const EvLoop &) = delete;
|
||||
EvLoop &operator=(const EvLoop &) = delete;
|
||||
|
||||
bool Init(const std::string &threadName);
|
||||
|
||||
int AddFuncToEvLoop(std::function<void()> &&func);
|
||||
|
||||
int AddFdEvent(int fd, uint32_t events, EventHandler handler, void *data);
|
||||
int ModifyFdEvent(int fd, uint32_t events);
|
||||
int DelFdEvent(int fd);
|
||||
void Finish();
|
||||
|
||||
~EvLoop();
|
||||
|
||||
int EventLoopCreate(void);
|
||||
void StopEventLoop();
|
||||
|
||||
void EventLoopDestroy();
|
||||
|
||||
void EventFreeDelEvents();
|
||||
void AddDeletedEvents(EventData *eventData);
|
||||
int FindDeletedEvent(const EventData *tev);
|
||||
|
||||
void HandleEvent(const struct epoll_event *events, int nevent);
|
||||
|
||||
void DeleteEvent(int fd);
|
||||
EventData *FindEvent(int fd);
|
||||
void AddEvent(EventData *eventData);
|
||||
void CleanUp();
|
||||
|
||||
int efd;
|
||||
int stopLoop;
|
||||
std::mutex loopMutex;
|
||||
sem_t semId;
|
||||
pthread_t loopThread;
|
||||
int queueEventfd;
|
||||
std::mutex queueMutex;
|
||||
std::queue<std::function<void()>> queue;
|
||||
|
||||
std::mutex eventsLock;
|
||||
// fd,EventData
|
||||
std::map<int, EventData *> events;
|
||||
|
||||
// Just to be safe, let's use a list to preserve deleted events rather than a map. Because the caller may
|
||||
// delete events on the same fd twice in once epoll_wait
|
||||
std::map<int, std::list<EventData *>> deletedEvents;
|
||||
};
|
||||
|
||||
} // namespace mindspore
|
||||
|
||||
#endif
|
||||
@ -1,5 +0,0 @@
|
||||
target_sources(litebus_obj PRIVATE
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/timer.cc
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/timertools.cc
|
||||
${CMAKE_CURRENT_SOURCE_DIR}/timewatch.cc
|
||||
)
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue