summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTatsuhiro Tsujikawa <tatsuhiro.t@gmail.com>2014-09-13 15:36:27 +0900
committerTatsuhiro Tsujikawa <tatsuhiro.t@gmail.com>2014-09-13 17:33:22 +0900
commit596e5c616218c9681adf4eb45077608510e2876e (patch)
treebe2c7fb20b14f0c4dc533c82dbb5dad23529d693
parentf55c16c7edeb319a82217a91a87a35cddfd606b0 (diff)
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.
-rw-r--r--src/EpollEventPoll.cc122
-rw-r--r--src/EpollEventPoll.h13
-rw-r--r--src/Event.h19
-rw-r--r--src/KqueueEventPoll.cc111
-rw-r--r--src/KqueueEventPoll.h15
-rw-r--r--src/LibuvEventPoll.cc6
-rw-r--r--src/PollEventPoll.cc125
-rw-r--r--src/PollEventPoll.h15
-rw-r--r--src/PortEventPoll.cc6
-rw-r--r--src/SelectEventPoll.cc76
-rw-r--r--src/SelectEventPoll.h16
11 files changed, 268 insertions, 256 deletions
diff --git a/src/EpollEventPoll.cc b/src/EpollEventPoll.cc
index 982ce8af..fc93f6d9 100644
--- a/src/EpollEventPoll.cc
+++ b/src/EpollEventPoll.cc
@@ -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
diff --git a/src/EpollEventPoll.h b/src/EpollEventPoll.h
index 0ec4fd60..902f9f64 100644
--- a/src/EpollEventPoll.h
+++ b/src/EpollEventPoll.h
@@ -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
diff --git a/src/Event.h b/src/Event.h
index 19238f87..1d9d24ee 100644
--- a/src/Event.h
+++ b/src/Event.h
@@ -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_ &&
diff --git a/src/KqueueEventPoll.cc b/src/KqueueEventPoll.cc
index b44e83da..c1b85804 100644
--- a/src/KqueueEventPoll.cc
+++ b/src/KqueueEventPoll.cc
@@ -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
diff --git a/src/KqueueEventPoll.h b/src/KqueueEventPoll.h
index 9e0ea47e..a5fdcf47 100644
--- a/src/KqueueEventPoll.h
+++ b/src/KqueueEventPoll.h
@@ -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;
diff --git a/src/LibuvEventPoll.cc b/src/LibuvEventPoll.cc
index 09d314b6..3fc76fda 100644
--- a/src/LibuvEventPoll.cc
+++ b/src/LibuvEventPoll.cc
@@ -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()) {
diff --git a/src/PollEventPoll.cc b/src/PollEventPoll.cc
index 5c64ef84..79f3d15e 100644
--- a/src/PollEventPoll.cc
+++ b/src/PollEventPoll.cc
@@ -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();
+ }
+
+ 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);
}
- break;
+ --pollfdNum_;
+ socketEntries_.erase(i);
+ } else {
+ *first = socketEntry.getEvents();
}
+ break;
}
- return true;
}
+ 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
diff --git a/src/PollEventPoll.h b/src/PollEventPoll.h
index 0b88baef..bd754712 100644
--- a/src/PollEventPoll.h
+++ b/src/PollEventPoll.h
@@ -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);
diff --git a/src/PortEventPoll.cc b/src/PortEventPoll.cc
index 0610cd97..9607729e 100644
--- a/src/PortEventPoll.cc
+++ b/src/PortEventPoll.cc
@@ -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()) {
diff --git a/src/SelectEventPoll.cc b/src/SelectEventPoll.cc
index 0aec1051..87efee56 100644
--- a/src/SelectEventPoll.cc
+++ b/src/SelectEventPoll.cc
@@ -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
diff --git a/src/SelectEventPoll.h b/src/SelectEventPoll.h
index 0a7a4b54..c3af5345 100644
--- a/src/SelectEventPoll.h
+++ b/src/SelectEventPoll.h
@@ -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