diff options
| author | Tatsuhiro Tsujikawa <tatsuhiro.t@gmail.com> | 2014-09-13 15:36:27 +0900 |
|---|---|---|
| committer | Tatsuhiro Tsujikawa <tatsuhiro.t@gmail.com> | 2014-09-13 17:33:22 +0900 |
| commit | 596e5c616218c9681adf4eb45077608510e2876e (patch) | |
| tree | be2c7fb20b14f0c4dc533c82dbb5dad23529d693 | |
| parent | f55c16c7edeb319a82217a91a87a35cddfd606b0 (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.cc | 122 | ||||
| -rw-r--r-- | src/EpollEventPoll.h | 13 | ||||
| -rw-r--r-- | src/Event.h | 19 | ||||
| -rw-r--r-- | src/KqueueEventPoll.cc | 111 | ||||
| -rw-r--r-- | src/KqueueEventPoll.h | 15 | ||||
| -rw-r--r-- | src/LibuvEventPoll.cc | 6 | ||||
| -rw-r--r-- | src/PollEventPoll.cc | 125 | ||||
| -rw-r--r-- | src/PollEventPoll.h | 15 | ||||
| -rw-r--r-- | src/PortEventPoll.cc | 6 | ||||
| -rw-r--r-- | src/SelectEventPoll.cc | 76 | ||||
| -rw-r--r-- | src/SelectEventPoll.h | 16 |
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 |
