Refactor event poll classes so that we don't have to allocate memory on query

I left Port and libuv classes, since I don't have them compiled.
So they were updated minimally.
pull/287/head
Tatsuhiro Tsujikawa 2014-09-13 15:36:27 +09:00
parent f55c16c7ed
commit 596e5c6162
11 changed files with 271 additions and 259 deletions

View File

@ -132,12 +132,11 @@ void EpollEventPoll::poll(const struct timeval& tv)
// own timeout and ares may create new sockets or closes socket in
// their API. So we call ares_process_fd for all ares_channel and
// re-register their sockets.
for(KAsyncNameResolverEntrySet::iterator i =
nameResolverEntries_.begin(), eoi = nameResolverEntries_.end();
i != eoi; ++i) {
(*i)->processTimeout();
(*i)->removeSocketEvents(this);
(*i)->addSocketEvents(this);
for(auto& i : nameResolverEntries_) {
auto& ent = i.second;
ent.processTimeout();
ent.removeSocketEvents(this);
ent.addSocketEvents(this);
}
#endif // ENABLE_ASYNC_DNS
@ -168,36 +167,37 @@ int translateEvents(EventPoll::EventType events)
bool EpollEventPoll::addEvents(sock_t socket,
const EpollEventPoll::KEvent& event)
{
auto socketEntry = std::make_shared<KSocketEntry>(socket);
KSocketEntrySet::iterator i = socketEntries_.lower_bound(socketEntry);
auto i = socketEntries_.lower_bound(socket);
int r = 0;
int errNum = 0;
if(i != socketEntries_.end() && *(*i) == *socketEntry) {
if(i != std::end(socketEntries_) && (*i).first == socket) {
auto& socketEntry = (*i).second;
event.addSelf(*i);
event.addSelf(&socketEntry);
struct epoll_event epEvent = (*i)->getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_MOD, (*i)->getSocket(), &epEvent);
struct epoll_event epEvent = socketEntry.getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_MOD, socketEntry.getSocket(), &epEvent);
if(r == -1) {
// try EPOLL_CTL_ADD: There is a chance that previously socket X is
// added to epoll, but it is closed and is not yet removed from
// SocketEntries. In this case, EPOLL_CTL_MOD is failed with ENOENT.
r = epoll_ctl(epfd_, EPOLL_CTL_ADD, (*i)->getSocket(),
r = epoll_ctl(epfd_, EPOLL_CTL_ADD, socketEntry.getSocket(),
&epEvent);
errNum = errno;
}
} else {
socketEntries_.insert(i, socketEntry);
i = socketEntries_.insert(i, std::make_pair(socket, KSocketEntry(socket)));
auto& socketEntry = (*i).second;
if(socketEntries_.size() > epEventsSize_) {
epEventsSize_ *= 2;
epEvents_ = make_unique<struct epoll_event[]>(epEventsSize_);
}
event.addSelf(socketEntry);
event.addSelf(&socketEntry);
struct epoll_event epEvent = socketEntry->getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_ADD, socketEntry->getSocket(), &epEvent);
struct epoll_event epEvent = socketEntry.getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_ADD, socketEntry.getSocket(), &epEvent);
errNum = errno;
}
if(r == -1) {
@ -228,41 +228,41 @@ bool EpollEventPoll::addEvents(sock_t socket, Command* command, int events,
bool EpollEventPoll::deleteEvents(sock_t socket,
const EpollEventPoll::KEvent& event)
{
auto socketEntry = std::make_shared<KSocketEntry>(socket);
KSocketEntrySet::iterator i = socketEntries_.find(socketEntry);
if(i == socketEntries_.end()) {
auto i = socketEntries_.find(socket);
if(i == std::end(socketEntries_)) {
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.", socket));
return false;
}
auto& socketEntry = (*i).second;
event.removeSelf(&socketEntry);
int r = 0;
int errNum = 0;
if(socketEntry.eventEmpty()) {
// In kernel before 2.6.9, epoll_ctl with EPOLL_CTL_DEL requires non-null
// pointer of epoll_event.
struct epoll_event ev = {0,{0}};
r = epoll_ctl(epfd_, EPOLL_CTL_DEL, socketEntry.getSocket(), &ev);
errNum = errno;
socketEntries_.erase(i);
} else {
event.removeSelf(*i);
int r = 0;
int errNum = 0;
if((*i)->eventEmpty()) {
// In kernel before 2.6.9, epoll_ctl with EPOLL_CTL_DEL requires non-null
// pointer of epoll_event.
struct epoll_event ev = {0,{0}};
r = epoll_ctl(epfd_, EPOLL_CTL_DEL, (*i)->getSocket(), &ev);
errNum = errno;
socketEntries_.erase(i);
} else {
// If socket is closed, then it seems it is automatically removed from
// epoll, so following EPOLL_CTL_MOD may fail.
struct epoll_event epEvent = (*i)->getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_MOD, (*i)->getSocket(), &epEvent);
errNum = errno;
if(r == -1) {
A2_LOG_DEBUG(fmt("Failed to delete socket event, but may be ignored:%s",
util::safeStrerror(errNum).c_str()));
}
}
// If socket is closed, then it seems it is automatically removed from
// epoll, so following EPOLL_CTL_MOD may fail.
struct epoll_event epEvent = socketEntry.getEvents();
r = epoll_ctl(epfd_, EPOLL_CTL_MOD, socketEntry.getSocket(), &epEvent);
errNum = errno;
if(r == -1) {
A2_LOG_DEBUG(fmt("Failed to delete socket event:%s",
A2_LOG_DEBUG(fmt("Failed to delete socket event, but may be ignored:%s",
util::safeStrerror(errNum).c_str()));
return false;
} else {
return true;
}
}
if(r == -1) {
A2_LOG_DEBUG(fmt("Failed to delete socket event:%s",
util::safeStrerror(errNum).c_str()));
return false;
} else {
return true;
}
}
#ifdef ENABLE_ASYNC_DNS
@ -284,31 +284,31 @@ bool EpollEventPoll::deleteEvents(sock_t socket, Command* command,
bool EpollEventPoll::addNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
auto entry = std::make_shared<KAsyncNameResolverEntry>(resolver, command);
KAsyncNameResolverEntrySet::iterator itr =
nameResolverEntries_.lower_bound(entry);
if(itr != nameResolverEntries_.end() && *(*itr) == *entry) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.lower_bound(key);
if(itr != std::end(nameResolverEntries_) && (*itr).first == key) {
return false;
} else {
nameResolverEntries_.insert(itr, entry);
entry->addSocketEvents(this);
return true;
}
itr = nameResolverEntries_.insert
(itr, std::make_pair(key, KAsyncNameResolverEntry(resolver, command)));
(*itr).second.addSocketEvents(this);
return true;
}
bool EpollEventPoll::deleteNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
auto entry = std::make_shared<KAsyncNameResolverEntry>(resolver, command);
KAsyncNameResolverEntrySet::iterator itr =
nameResolverEntries_.find(entry);
if(itr == nameResolverEntries_.end()) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.find(key);
if(itr == std::end(nameResolverEntries_)) {
return false;
} else {
(*itr)->removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
(*itr).second.removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
#endif // ENABLE_ASYNC_DNS

View File

@ -39,7 +39,7 @@
# include <sys/epoll.h>
#include <set>
#include <map>
#include "Event.h"
#include "a2functional.h"
@ -64,19 +64,20 @@ private:
public:
KSocketEntry(sock_t socket);
KSocketEntry(const KSocketEntry&) = delete;
KSocketEntry(KSocketEntry&&) = default;
struct epoll_event getEvents();
};
friend int accumulateEvent(int events, const KEvent& event);
private:
typedef std::set<std::shared_ptr<KSocketEntry>,
DerefLess<std::shared_ptr<KSocketEntry> > > KSocketEntrySet;
typedef std::map<sock_t, KSocketEntry> KSocketEntrySet;
KSocketEntrySet socketEntries_;
#ifdef ENABLE_ASYNC_DNS
typedef std::set<std::shared_ptr<KAsyncNameResolverEntry>,
DerefLess<std::shared_ptr<KAsyncNameResolverEntry> > >
KAsyncNameResolverEntrySet;
typedef std::map<std::pair<AsyncNameResolver*, Command*>,
KAsyncNameResolverEntry> KAsyncNameResolverEntrySet;
KAsyncNameResolverEntrySet nameResolverEntries_;
#endif // ENABLE_ASYNC_DNS

View File

@ -59,10 +59,9 @@ public:
virtual int getEvents() const = 0;
virtual void addSelf(const std::shared_ptr<SocketEntry>& socketEntry) const = 0;
virtual void addSelf(SocketEntry* socketEntry) const = 0;
virtual void removeSelf
(const std::shared_ptr<SocketEntry>& socketEntry) const = 0;
virtual void removeSelf(SocketEntry* socketEntry) const = 0;
};
template<typename SocketEntry, typename EventPoll>
@ -124,12 +123,12 @@ public:
}
}
virtual void addSelf(const std::shared_ptr<SocketEntry>& socketEntry) const
virtual void addSelf(SocketEntry* socketEntry) const
{
socketEntry->addCommandEvent(*this);
}
virtual void removeSelf(const std::shared_ptr<SocketEntry>& socketEntry) const
virtual void removeSelf(SocketEntry* socketEntry) const
{
socketEntry->removeCommandEvent(*this);
}
@ -179,12 +178,12 @@ public:
command_->setStatusActive();
}
virtual void addSelf(const std::shared_ptr<SocketEntry>& socketEntry) const
virtual void addSelf(SocketEntry* socketEntry) const
{
socketEntry->addADNSEvent(*this);
}
virtual void removeSelf(const std::shared_ptr<SocketEntry>& socketEntry) const
virtual void removeSelf(SocketEntry* socketEntry) const
{
socketEntry->removeADNSEvent(*this);
}
@ -209,7 +208,8 @@ protected:
public:
SocketEntry(sock_t socket):socket_(socket) {}
virtual ~SocketEntry() {}
SocketEntry(const SocketEntry&) = delete;
SocketEntry(SocketEntry&&) = default;
bool operator==(const SocketEntry& entry) const
{
@ -318,6 +318,9 @@ public:
Command* command):
nameResolver_(std::move(nameResolver)), command_(command), socketsSize_(0) {}
AsyncNameResolverEntry(const AsyncNameResolverEntry&) = delete;
AsyncNameResolverEntry(AsyncNameResolverEntry&&) = default;
bool operator==(const AsyncNameResolverEntry& entry)
{
return *nameResolver_ == *entry.nameResolver_ &&

View File

@ -89,7 +89,7 @@ size_t KqueueEventPoll::KSocketEntry::getEvents
KqueueEventPoll::KqueueEventPoll()
: kqEventsSize_(KQUEUE_EVENTS_MAX),
kqEvents_(new struct kevent[kqEventsSize_])
kqEvents_(make_unique<struct kevent[]>(kqEventsSize_))
{
kqfd_ = kqueue();
}
@ -106,7 +106,6 @@ KqueueEventPoll::~KqueueEventPoll()
util::safeStrerror(errNum).c_str()));
}
}
delete [] kqEvents_;
}
bool KqueueEventPoll::good() const
@ -118,7 +117,8 @@ void KqueueEventPoll::poll(const struct timeval& tv)
{
struct timespec timeout = { tv.tv_sec, tv.tv_usec*1000 };
int res;
while((res = kevent(kqfd_, kqEvents_, 0, kqEvents_, kqEventsSize_, &timeout))
while((res = kevent(kqfd_, kqEvents_.get(), 0, kqEvents_.get(),
kqEventsSize_, &timeout))
== -1 && errno == EINTR);
if(res > 0) {
for(int i = 0; i < res; ++i) {
@ -142,9 +142,10 @@ void KqueueEventPoll::poll(const struct timeval& tv)
// their API. So we call ares_process_fd for all ares_channel and
// re-register their sockets.
for(auto & r : nameResolverEntries_) {
r->processTimeout();
r->removeSocketEvents(this);
r->addSocketEvents(this);
auto& ent = r.second;
ent.processTimeout();
ent.removeSocketEvents(this);
ent.addSocketEvents(this);
}
#endif // ENABLE_ASYNC_DNS
@ -169,24 +170,24 @@ int translateEvents(EventPoll::EventType events)
bool KqueueEventPoll::addEvents
(sock_t socket, const KqueueEventPoll::KEvent& event)
{
std::shared_ptr<KSocketEntry> socketEntry(new KSocketEntry(socket));
auto i = socketEntries_.lower_bound(socketEntry);
auto i = socketEntries_.lower_bound(socket);
int r = 0;
struct timespec zeroTimeout = { 0, 0 };
struct kevent changelist[2];
size_t n;
if(i != socketEntries_.end() && *(*i) == *socketEntry) {
event.addSelf(*i);
n = (*i)->getEvents(changelist);
if(i != std::end(socketEntries_) && (*i).first == socket) {
auto& socketEntry = (*i).second;
event.addSelf(&socketEntry);
n = socketEntry.getEvents(changelist);
} else {
socketEntries_.insert(i, socketEntry);
i = socketEntries_.insert(i, std::make_pair(socket, KSocketEntry(socket)));
auto& socketEntry = (*i).second;
if(socketEntries_.size() > kqEventsSize_) {
kqEventsSize_ *= 2;
delete [] kqEvents_;
kqEvents_ = new struct kevent[kqEventsSize_];
kqEvents_ = make_unique<struct kevent[]>(kqEventsSize_);
}
event.addSelf(socketEntry);
n = socketEntry->getEvents(changelist);
event.addSelf(&socketEntry);
n = socketEntry.getEvents(changelist);
}
r = kevent(kqfd_, changelist, n, changelist, 0, &zeroTimeout);
int errNum = errno;
@ -201,7 +202,7 @@ bool KqueueEventPoll::addEvents
}
bool KqueueEventPoll::addEvents(sock_t socket, Command* command,
EventPoll::EventType events)
EventPoll::EventType events)
{
int kqEvents = translateEvents(events);
return addEvents(socket, KCommandEvent(command, kqEvents));
@ -216,44 +217,44 @@ bool KqueueEventPoll::addEvents(sock_t socket, Command* command, int events,
#endif // ENABLE_ASYNC_DNS
bool KqueueEventPoll::deleteEvents(sock_t socket,
const KqueueEventPoll::KEvent& event)
const KqueueEventPoll::KEvent& event)
{
std::shared_ptr<KSocketEntry> socketEntry(new KSocketEntry(socket));
auto i = socketEntries_.find(socketEntry);
if(i == socketEntries_.end()) {
auto i = socketEntries_.find(socket);
if(i == std::end(socketEntries_)) {
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.", socket));
return false;
}
auto& socketEntry = (*i).second;
event.removeSelf(&socketEntry);
int r = 0;
struct timespec zeroTimeout = { 0, 0 };
struct kevent changelist[2];
size_t n = socketEntry.getEvents(changelist);
r = kevent(kqfd_, changelist, n, changelist, 0, &zeroTimeout);
int errNum = errno;
if(socketEntry.eventEmpty()) {
socketEntries_.erase(i);
}
if(r == -1) {
A2_LOG_DEBUG(fmt("Failed to delete socket event:%s",
util::safeStrerror(errNum).c_str()));
return false;
} else {
event.removeSelf(*i);
int r = 0;
struct timespec zeroTimeout = { 0, 0 };
struct kevent changelist[2];
size_t n = (*i)->getEvents(changelist);
r = kevent(kqfd_, changelist, n, changelist, 0, &zeroTimeout);
int errNum = errno;
if((*i)->eventEmpty()) {
socketEntries_.erase(i);
}
if(r == -1) {
A2_LOG_DEBUG(fmt("Failed to delete socket event:%s",
util::safeStrerror(errNum).c_str()));
return false;
} else {
return true;
}
return true;
}
}
#ifdef ENABLE_ASYNC_DNS
bool KqueueEventPoll::deleteEvents(sock_t socket, Command* command,
const std::shared_ptr<AsyncNameResolver>& rs)
const std::shared_ptr<AsyncNameResolver>& rs)
{
return deleteEvents(socket, KADNSEvent(rs, command, socket, 0));
}
#endif // ENABLE_ASYNC_DNS
bool KqueueEventPoll::deleteEvents(sock_t socket, Command* command,
EventPoll::EventType events)
EventPoll::EventType events)
{
int kqEvents = translateEvents(events);
return deleteEvents(socket, KCommandEvent(command, kqEvents));
@ -263,16 +264,17 @@ bool KqueueEventPoll::deleteEvents(sock_t socket, Command* command,
bool KqueueEventPoll::addNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
std::shared_ptr<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
auto itr = nameResolverEntries_.find(entry);
if(itr == nameResolverEntries_.end()) {
nameResolverEntries_.insert(entry);
entry->addSocketEvents(this);
return true;
} else {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.lower_bound(key);
if(itr != std::end(nameResolverEntries_) && (*itr).first == key) {
return false;
}
itr = nameResolverEntries_.insert
(itr, std::make_pair(key, KAsyncNameResolverEntry(resolver, command)));
(*itr).second.addSocketEvents(this);
return true;
}
bool KqueueEventPoll::deleteNameResolver
@ -280,14 +282,15 @@ bool KqueueEventPoll::deleteNameResolver
{
std::shared_ptr<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
auto itr = nameResolverEntries_.find(entry);
if(itr == nameResolverEntries_.end()) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.find(key);
if(itr == std::end(nameResolverEntries_)) {
return false;
} else {
(*itr)->removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
(*itr).second.removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
#endif // ENABLE_ASYNC_DNS

View File

@ -41,7 +41,7 @@
#include <sys/event.h>
#include <sys/time.h>
#include <set>
#include <map>
#include "Event.h"
#include "a2functional.h"
@ -67,6 +67,9 @@ private:
public:
KSocketEntry(sock_t socket);
KSocketEntry(const KSocketEntry&) = delete;
KSocketEntry(KSocketEntry&&) = default;
// eventlist should be at least size 2. This function returns the
// number of filled struct kevent in eventlist.
size_t getEvents(struct kevent* eventlist);
@ -75,13 +78,11 @@ private:
friend int accumulateEvent(int events, const KEvent& event);
private:
typedef std::set<std::shared_ptr<KSocketEntry>,
DerefLess<std::shared_ptr<KSocketEntry> > > KSocketEntrySet;
typedef std::map<sock_t, KSocketEntry> KSocketEntrySet;
KSocketEntrySet socketEntries_;
#ifdef ENABLE_ASYNC_DNS
typedef std::set<std::shared_ptr<KAsyncNameResolverEntry>,
DerefLess<std::shared_ptr<KAsyncNameResolverEntry> > >
KAsyncNameResolverEntrySet;
typedef std::map<std::pair<AsyncNameResolver*, Command*>,
KAsyncNameResolverEntry> KAsyncNameResolverEntrySet;
KAsyncNameResolverEntrySet nameResolverEntries_;
#endif // ENABLE_ASYNC_DNS
@ -89,7 +90,7 @@ private:
size_t kqEventsSize_;
struct kevent* kqEvents_;
std::unique_ptr<struct kevent[]> kqEvents_;
static const size_t KQUEUE_EVENTS_MAX = 1024;

View File

@ -233,7 +233,7 @@ bool LibuvEventPoll::addEvents(sock_t socket,
auto i = socketEntries_.lower_bound(socketEntry);
if (i != socketEntries_.end() && **i == *socketEntry) {
event.addSelf(*i);
event.addSelf((*i).get());
auto poll = polls_.find(socket);
if (poll == polls_.end()) {
throw std::logic_error("Invalid socket");
@ -243,7 +243,7 @@ bool LibuvEventPoll::addEvents(sock_t socket,
}
socketEntries_.insert(i, socketEntry);
event.addSelf(socketEntry);
event.addSelf(socketEntry.get());
auto poll = new KPoll(this, socketEntry.get(), socket);
polls_[socket] = poll;
poll->start();
@ -275,7 +275,7 @@ bool LibuvEventPoll::deleteEvents(sock_t socket,
return false;
}
event.removeSelf(*i);
event.removeSelf((*i).get());
auto poll = polls_.find(socket);
if (poll == polls_.end()) {

View File

@ -78,35 +78,30 @@ struct pollfd PollEventPoll::KSocketEntry::getEvents()
PollEventPoll::PollEventPoll()
: pollfdCapacity_(1024),
pollfdNum_(0)
{
pollfds_ = new struct pollfd[pollfdCapacity_];
}
pollfdNum_(0),
pollfds_(make_unique<struct pollfd[]>(pollfdCapacity_))
{}
PollEventPoll::~PollEventPoll()
{
delete [] pollfds_;
}
{}
void PollEventPoll::poll(const struct timeval& tv)
{
// timeout is millisec
int timeout = tv.tv_sec*1000+tv.tv_usec/1000;
int res;
while((res = ::poll(pollfds_, pollfdNum_, timeout)) == -1 &&
while((res = ::poll(pollfds_.get(), pollfdNum_, timeout)) == -1 &&
errno == EINTR);
if(res > 0) {
std::shared_ptr<KSocketEntry> se(new KSocketEntry(0));
for(struct pollfd* first = pollfds_, *last = pollfds_+pollfdNum_;
for(auto first = pollfds_.get(), last = pollfds_.get() + pollfdNum_;
first != last; ++first) {
if(first->revents) {
se->setSocket(first->fd);
auto itr = socketEntries_.find(se);
if(itr == socketEntries_.end()) {
auto itr = socketEntries_.find(first->fd);
if(itr == std::end(socketEntries_)) {
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.",
first->fd));
} else {
(*itr)->processEvents(first->revents);
(*itr).second.processEvents(first->revents);
}
}
}
@ -120,9 +115,10 @@ void PollEventPoll::poll(const struct timeval& tv)
// their API. So we call ares_process_fd for all ares_channel and
// re-register their sockets.
for(auto& r : nameResolverEntries_) {
r->processTimeout();
r->removeSocketEvents(this);
r->addSocketEvents(this);
auto& ent = r.second;
ent.processTimeout();
ent.removeSocketEvents(this);
ent.addSocketEvents(this);
}
#endif // ENABLE_ASYNC_DNS
@ -151,28 +147,29 @@ int PollEventPoll::translateEvents(EventPoll::EventType events)
bool PollEventPoll::addEvents
(sock_t socket, const PollEventPoll::KEvent& event)
{
std::shared_ptr<KSocketEntry> socketEntry(new KSocketEntry(socket));
auto i = socketEntries_.lower_bound(socketEntry);
if(i != socketEntries_.end() && *(*i) == *socketEntry) {
event.addSelf(*i);
for(struct pollfd* first = pollfds_, *last = pollfds_+pollfdNum_;
auto i = socketEntries_.lower_bound(socket);
if(i != std::end(socketEntries_) && (*i).first == socket) {
auto& socketEntry = (*i).second;
event.addSelf(&socketEntry);
for(auto first = pollfds_.get(), last = pollfds_.get() + pollfdNum_;
first != last; ++first) {
if(first->fd == socket) {
*first = (*i)->getEvents();
*first = socketEntry.getEvents();
break;
}
}
} else {
socketEntries_.insert(i, socketEntry);
event.addSelf(socketEntry);
i = socketEntries_.insert(i, std::make_pair(socket, KSocketEntry(socket)));
auto& socketEntry = (*i).second;
event.addSelf(&socketEntry);
if(pollfdCapacity_ == pollfdNum_) {
pollfdCapacity_ *= 2;
auto newPollfds = new struct pollfd[pollfdCapacity_];
memcpy(newPollfds, pollfds_, pollfdNum_*sizeof(struct pollfd));
delete [] pollfds_;
pollfds_ = newPollfds;
auto newPollfds = make_unique<struct pollfd[]>(pollfdCapacity_);
memcpy(newPollfds.get(), pollfds_.get(),
pollfdNum_ * sizeof(struct pollfd));
pollfds_ = std::move(newPollfds);
}
pollfds_[pollfdNum_] = socketEntry->getEvents();
pollfds_[pollfdNum_] = socketEntry.getEvents();
++pollfdNum_;
}
return true;
@ -197,30 +194,30 @@ bool PollEventPoll::addEvents
bool PollEventPoll::deleteEvents
(sock_t socket, const PollEventPoll::KEvent& event)
{
std::shared_ptr<KSocketEntry> socketEntry(new KSocketEntry(socket));
auto i = socketEntries_.find(socketEntry);
if(i == socketEntries_.end()) {
auto i = socketEntries_.find(socket);
if(i == std::end(socketEntries_)) {
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.", socket));
return false;
} else {
event.removeSelf(*i);
for(struct pollfd* first = pollfds_, *last = pollfds_+pollfdNum_;
first != last; ++first) {
if(first->fd == socket) {
if((*i)->eventEmpty()) {
if(pollfdNum_ >= 2) {
*first = *(last-1);
}
--pollfdNum_;
socketEntries_.erase(i);
} else {
*first = (*i)->getEvents();
}
break;
}
}
return true;
}
auto& socketEntry = (*i).second;
event.removeSelf(&socketEntry);
for(auto first = pollfds_.get(), last = pollfds_.get() + pollfdNum_;
first != last; ++first) {
if(first->fd == socket) {
if(socketEntry.eventEmpty()) {
if(pollfdNum_ >= 2) {
*first = *(last-1);
}
--pollfdNum_;
socketEntries_.erase(i);
} else {
*first = socketEntry.getEvents();
}
break;
}
}
return true;
}
#ifdef ENABLE_ASYNC_DNS
@ -242,31 +239,31 @@ bool PollEventPoll::deleteEvents
bool PollEventPoll::addNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
std::shared_ptr<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
auto itr = nameResolverEntries_.lower_bound(entry);
if(itr != nameResolverEntries_.end() && *(*itr) == *entry) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.lower_bound(key);
if(itr != std::end(nameResolverEntries_) && (*itr).first == key) {
return false;
} else {
nameResolverEntries_.insert(itr, entry);
entry->addSocketEvents(this);
return true;
}
itr = nameResolverEntries_.insert
(itr, std::make_pair(key, KAsyncNameResolverEntry(resolver, command)));
(*itr).second.addSocketEvents(this);
return true;
}
bool PollEventPoll::deleteNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
std::shared_ptr<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
auto itr = nameResolverEntries_.find(entry);
if(itr == nameResolverEntries_.end()) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.find(key);
if(itr == std::end(nameResolverEntries_)) {
return false;
} else {
(*itr)->removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
(*itr).second.removeSocketEvents(this);
nameResolverEntries_.erase(itr);
return true;
}
#endif // ENABLE_ASYNC_DNS

View File

@ -39,7 +39,7 @@
# include <poll.h>
#include <set>
#include <map>
#include "Event.h"
#include "a2functional.h"
@ -64,19 +64,20 @@ private:
public:
KSocketEntry(sock_t socket);
KSocketEntry(const KSocketEntry&) = delete;
KSocketEntry(KSocketEntry&&) = default;
struct pollfd getEvents();
};
friend int accumulateEvent(int events, const KEvent& event);
private:
typedef std::set<std::shared_ptr<KSocketEntry>,
DerefLess<std::shared_ptr<KSocketEntry> > > KSocketEntrySet;
typedef std::map<sock_t, KSocketEntry> KSocketEntrySet;
KSocketEntrySet socketEntries_;
#ifdef ENABLE_ASYNC_DNS
typedef std::set<std::shared_ptr<KAsyncNameResolverEntry>,
DerefLess<std::shared_ptr<KAsyncNameResolverEntry> > >
KAsyncNameResolverEntrySet;
typedef std::map<std::pair<AsyncNameResolver*, Command*>,
KAsyncNameResolverEntry> KAsyncNameResolverEntrySet;
KAsyncNameResolverEntrySet nameResolverEntries_;
#endif // ENABLE_ASYNC_DNS
@ -86,7 +87,7 @@ private:
// The number of valid struct pollfd in pollfds_.
int pollfdNum_;
struct pollfd* pollfds_;
std::unique_ptr<struct pollfd[]> pollfds_;
bool addEvents(sock_t socket, const KEvent& event);

View File

@ -179,7 +179,7 @@ bool PortEventPoll::addEvents(sock_t socket,
int r = 0;
int errNum = 0;
if(i != socketEntries_.end() && *(*i) == *socketEntry) {
event.addSelf(*i);
event.addSelf((*i).get());
A2PortEvent pv = (*i)->getEvents();
r = port_associate(port_, PORT_SOURCE_FD, (*i)->getSocket(),
pv.events, pv.socketEntry);
@ -191,7 +191,7 @@ bool PortEventPoll::addEvents(sock_t socket,
delete [] portEvents_;
portEvents_ = new port_event_t[portEventsSize_];
}
event.addSelf(socketEntry);
event.addSelf(socketEntry.get());
A2PortEvent pv = socketEntry->getEvents();
r = port_associate(port_, PORT_SOURCE_FD, socketEntry->getSocket(),
pv.events, pv.socketEntry);
@ -230,7 +230,7 @@ bool PortEventPoll::deleteEvents(sock_t socket,
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.", socket));
return false;
} else {
event.removeSelf(*i);
event.removeSelf((*i).get());
int r = 0;
int errNum = 0;
if((*i)->eventEmpty()) {

View File

@ -177,8 +177,9 @@ void SelectEventPoll::poll(const struct timeval& tv)
#endif // __MINGW32__
#ifdef ENABLE_ASYNC_DNS
for(auto& entry : nameResolverEntries_) {
int fd = entry->getFds(&rfds, &wfds);
for(auto& i : nameResolverEntries_) {
auto& entry = i.second;
int fd = entry.getFds(&rfds, &wfds);
// TODO force error if fd == 0
if(fdmax_ < fd) {
fdmax_ = fd;
@ -196,15 +197,16 @@ void SelectEventPoll::poll(const struct timeval& tv)
#endif // !__MINGW32__
} while(retval == -1 && errno == EINTR);
if(retval > 0) {
for(auto& e: socketEntries_) {
for(auto& i: socketEntries_) {
auto& e = i.second;
int events = 0;
if(FD_ISSET(e->getSocket(), &rfds)) {
if(FD_ISSET(e.getSocket(), &rfds)) {
events |= EventPoll::EVENT_READ;
}
if(FD_ISSET(e->getSocket(), &wfds)) {
if(FD_ISSET(e.getSocket(), &wfds)) {
events |= EventPoll::EVENT_WRITE;
}
e->processEvents(events);
e.processEvents(events);
}
} else if(retval == -1) {
int errNum = errno;
@ -212,8 +214,8 @@ void SelectEventPoll::poll(const struct timeval& tv)
}
#ifdef ENABLE_ASYNC_DNS
for(auto & e: nameResolverEntries_) {
e->process(&rfds, &wfds);
for(auto& i: nameResolverEntries_) {
i.second.process(&rfds, &wfds);
}
#endif // ENABLE_ASYNC_DNS
@ -240,8 +242,9 @@ void SelectEventPoll::updateFdSet()
#endif // !__MINGW32__
FD_ZERO(&rfdset_);
FD_ZERO(&wfdset_);
for(auto& e: socketEntries_) {
sock_t fd = e->getSocket();
for(auto& i: socketEntries_) {
auto&e = i.second;
sock_t fd = e.getSocket();
#ifndef __MINGW32__
if(fd < 0 || FD_SETSIZE <= fd) {
A2_LOG_WARN("Detected file descriptor >= FD_SETSIZE or < 0. "
@ -249,7 +252,7 @@ void SelectEventPoll::updateFdSet()
continue;
}
#endif // !__MINGW32__
int events = e->getEvents();
int events = e.getEvents();
if(events&EventPoll::EVENT_READ) {
#ifdef __MINGW32__
checkFdCountMingw(rfdset_);
@ -271,13 +274,12 @@ void SelectEventPoll::updateFdSet()
bool SelectEventPoll::addEvents(sock_t socket, Command* command,
EventPoll::EventType events)
{
std::shared_ptr<SocketEntry> socketEntry(new SocketEntry(socket));
auto i = socketEntries_.lower_bound(socketEntry);
if(i != socketEntries_.end() && *(*i) == *socketEntry) {
(*i)->addCommandEvent(command, events);
auto i = socketEntries_.lower_bound(socket);
if(i != std::end(socketEntries_) && (*i).first == socket) {
(*i).second.addCommandEvent(command, events);
} else {
socketEntries_.insert(i, socketEntry);
socketEntry->addCommandEvent(command, events);
i = socketEntries_.insert(i, std::make_pair(socket, SocketEntry(socket)));
(*i).second.addCommandEvent(command, events);
}
updateFdSet();
return true;
@ -286,42 +288,42 @@ bool SelectEventPoll::addEvents(sock_t socket, Command* command,
bool SelectEventPoll::deleteEvents(sock_t socket, Command* command,
EventPoll::EventType events)
{
std::shared_ptr<SocketEntry> socketEntry(new SocketEntry(socket));
auto i = socketEntries_.find(socketEntry);
if(i == socketEntries_.end()) {
auto i = socketEntries_.find(socket);
if(i == std::end(socketEntries_)) {
A2_LOG_DEBUG(fmt("Socket %d is not found in SocketEntries.", socket));
return false;
} else {
(*i)->removeCommandEvent(command, events);
if((*i)->eventEmpty()) {
socketEntries_.erase(i);
}
updateFdSet();
return true;
}
auto& socketEntry = (*i).second;
socketEntry.removeCommandEvent(command, events);
if(socketEntry.eventEmpty()) {
socketEntries_.erase(i);
}
updateFdSet();
return true;
}
#ifdef ENABLE_ASYNC_DNS
bool SelectEventPoll::addNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
std::shared_ptr<AsyncNameResolverEntry> entry
(new AsyncNameResolverEntry(resolver, command));
auto itr = nameResolverEntries_.lower_bound(entry);
if(itr != nameResolverEntries_.end() && *(*itr) == *entry) {
auto key = std::make_pair(resolver.get(), command);
auto itr = nameResolverEntries_.lower_bound(key);
if(itr != std::end(nameResolverEntries_) && (*itr).first == key) {
return false;
} else {
nameResolverEntries_.insert(itr, entry);
return true;
}
nameResolverEntries_.insert
(itr, std::make_pair(key, AsyncNameResolverEntry(resolver, command)));
return true;
}
bool SelectEventPoll::deleteNameResolver
(const std::shared_ptr<AsyncNameResolver>& resolver, Command* command)
{
std::shared_ptr<AsyncNameResolverEntry> entry
(new AsyncNameResolverEntry(resolver, command));
return nameResolverEntries_.erase(entry) == 1;
auto key = std::make_pair(resolver.get(), command);
return nameResolverEntries_.erase(key) == 1;
}
#endif // ENABLE_ASYNC_DNS

View File

@ -38,7 +38,7 @@
#include "EventPoll.h"
#include <deque>
#include <set>
#include <map>
#include "a2functional.h"
#ifdef ENABLE_ASYNC_DNS
@ -100,6 +100,9 @@ private:
public:
SocketEntry(sock_t socket);
SocketEntry(const SocketEntry&) = delete;
SocketEntry(SocketEntry&&) = default;
bool operator==(const SocketEntry& entry) const
{
return socket_ == entry.socket_;
@ -141,6 +144,9 @@ private:
AsyncNameResolverEntry(const std::shared_ptr<AsyncNameResolver>& nameResolver,
Command* command);
AsyncNameResolverEntry(const AsyncNameResolverEntry&) = delete;
AsyncNameResolverEntry(AsyncNameResolverEntry&&) = default;
bool operator==(const AsyncNameResolverEntry& entry)
{
return *nameResolver_ == *entry.nameResolver_ &&
@ -164,13 +170,11 @@ private:
fd_set wfdset_;
sock_t fdmax_;
typedef std::set<std::shared_ptr<SocketEntry>,
DerefLess<std::shared_ptr<SocketEntry> > > SocketEntrySet;
typedef std::map<sock_t, SocketEntry> SocketEntrySet;
SocketEntrySet socketEntries_;
#ifdef ENABLE_ASYNC_DNS
typedef std::set<std::shared_ptr<AsyncNameResolverEntry>,
DerefLess<std::shared_ptr<AsyncNameResolverEntry> > >
AsyncNameResolverEntrySet;
typedef std::map<std::pair<AsyncNameResolver*, Command*>,
AsyncNameResolverEntry> AsyncNameResolverEntrySet;
AsyncNameResolverEntrySet nameResolverEntries_;
#endif // ENABLE_ASYNC_DNS