github
/
aria2
mirror of https://github.com/aria2/aria2
1
0
Fork 0
Code Issues Releases Wiki Activity

2010-04-20 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net> 

Externalized Event, CommandEvent, ADNSEvent, SocketEntry and
	AsyncNameResolverEntry class.
	* src/Event.h
	* src/Makefile.am
	* src/PollEventPoll.cc
	* src/PollEventPoll.h
pull/1/head
Tatsuhiro Tsujikawa 2010-04-19 15:04:38 +00:00
parent 0ec50a8d50
commit e9ef6f3af9
6 changed files with 454 additions and 447 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
ChangeLog
View File

@ -1,3 +1,12 @@
2010-04-20 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Externalized Event, CommandEvent, ADNSEvent, SocketEntry and
AsyncNameResolverEntry class.
* src/Event.h
* src/Makefile.am
* src/PollEventPoll.cc
* src/PollEventPoll.h
2010-04-19 Tatsuhiro Tsujikawa <t-tujikawa@users.sourceforge.net>
Supported poll() for socket event notification. --event-poll can

378
src/Event.h Normal file
View File

@ -0,0 +1,378 @@
/* <!-- copyright */
/*
* aria2 - The high speed download utility
*
* Copyright (C) 2010 Tatsuhiro Tsujikawa
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*
* In addition, as a special exception, the copyright holders give
* permission to link the code of portions of this program with the
* OpenSSL library under certain conditions as described in each
* individual source file, and distribute linked combinations
* including the two.
* You must obey the GNU General Public License in all respects
* for all of the code used other than OpenSSL. If you modify
* file(s) with this exception, you may extend this exception to your
* version of the file(s), but you are not obligated to do so. If you
* do not wish to do so, delete this exception statement from your
* version. If you delete this exception statement from all source
* files in the program, then also delete it here.
*/
/* copyright --> */
#ifndef _D_EVENT_H_
#define _D_EVENT_H_
#include "common.h"
#include <deque>
#include <algorithm>
#include <functional>
#include "SharedHandle.h"
#include "a2netcompat.h"
#include "Command.h"
#ifdef ENABLE_ASYNC_DNS
# include "AsyncNameResolver.h"
#endif // ENABLE_ASYNC_DNS
namespace aria2 {
template<typename SocketEntry>
class Event {
public:
virtual ~Event() {}
virtual void processEvents(int events) = 0;
virtual int getEvents() const = 0;
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const = 0;
virtual void removeSelf
(const SharedHandle<SocketEntry>& socketEntry) const = 0;
};
template<typename SocketEntry, typename EventPoll>
class CommandEvent : public Event<SocketEntry> {
private:
Command* _command;
int _events;
public:
CommandEvent(Command* command, int events):
_command(command), _events(events) {}
Command* getCommand() const
{
return _command;
}
void addEvents(int events)
{
_events |= events;
}
void removeEvents(int events)
{
_events &= (~events);
}
bool eventsEmpty() const
{
return _events == 0;
}
bool operator==(const CommandEvent& commandEvent) const
{
return _command == commandEvent._command;
}
virtual int getEvents() const
{
return _events;
}
virtual void processEvents(int events)
{
if((_events&events) ||
((EventPoll::IEV_ERROR|EventPoll::IEV_HUP)&events)) {
_command->setStatusActive();
}
if(EventPoll::IEV_READ&events) {
_command->readEventReceived();
}
if(EventPoll::IEV_WRITE&events) {
_command->writeEventReceived();
}
if(EventPoll::IEV_ERROR&events) {
_command->errorEventReceived();
}
if(EventPoll::IEV_HUP&events) {
_command->hupEventReceived();
}
}
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->addCommandEvent(*this);
}
virtual void removeSelf(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->removeCommandEvent(*this);
}
};
#ifdef ENABLE_ASYNC_DNS
template<typename SocketEntry, typename EventPoll>
class ADNSEvent : public Event<SocketEntry> {
private:
SharedHandle<AsyncNameResolver> _resolver;
Command* _command;
sock_t _socket;
int _events;
public:
ADNSEvent(const SharedHandle<AsyncNameResolver>& resolver, Command* command,
sock_t socket, int events):
_resolver(resolver), _command(command), _socket(socket), _events(events) {}
bool operator==(const ADNSEvent& event) const
{
return _resolver == event._resolver;
}
virtual int getEvents() const
{
return _events;
}
virtual void processEvents(int events)
{
ares_socket_t readfd;
ares_socket_t writefd;
if(events&(EventPoll::IEV_READ|EventPoll::IEV_ERROR|
EventPoll::IEV_HUP)) {
readfd = _socket;
} else {
readfd = ARES_SOCKET_BAD;
}
if(events&(EventPoll::IEV_WRITE|EventPoll::IEV_ERROR|
EventPoll::IEV_HUP)) {
writefd = _socket;
} else {
writefd = ARES_SOCKET_BAD;
}
_resolver->process(readfd, writefd);
_command->setStatusActive();
}
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->addADNSEvent(*this);
}
virtual void removeSelf(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->removeADNSEvent(*this);
}
};
#else // !ENABLE_ASYNC_DNS
template<typename SocketEntry, typename EventPoll>
class ADNSEvent : public Event<SocketEntry> {};
#endif // !ENABLE_ASYNC_DNS
template<typename CommandEvent, typename ADNSEvent, typename EventRType>
class SocketEntry {
protected:
sock_t _socket;
std::deque<CommandEvent> _commandEvents;
#ifdef ENABLE_ASYNC_DNS
std::deque<ADNSEvent> _adnsEvents;
#endif // ENABLE_ASYNC_DNS
public:
SocketEntry(sock_t socket):_socket(socket) {}
virtual ~SocketEntry() {}
bool operator==(const SocketEntry& entry) const
{
return _socket == entry._socket;
}
bool operator<(const SocketEntry& entry) const
{
return _socket < entry._socket;
}
void addCommandEvent(const CommandEvent& cev)
{
typename std::deque<CommandEvent>::iterator i =
std::find(_commandEvents.begin(), _commandEvents.end(), cev);
if(i == _commandEvents.end()) {
_commandEvents.push_back(cev);
} else {
(*i).addEvents(cev.getEvents());
}
}
void removeCommandEvent(const CommandEvent& cev)
{
typename std::deque<CommandEvent>::iterator i =
std::find(_commandEvents.begin(), _commandEvents.end(), cev);
if(i == _commandEvents.end()) {
// not found
} else {
(*i).removeEvents(cev.getEvents());
if((*i).eventsEmpty()) {
_commandEvents.erase(i);
}
}
}
#ifdef ENABLE_ASYNC_DNS
void addADNSEvent(const ADNSEvent& aev)
{
typename std::deque<ADNSEvent>::iterator i =
std::find(_adnsEvents.begin(), _adnsEvents.end(), aev);
if(i == _adnsEvents.end()) {
_adnsEvents.push_back(aev);
}
}
void removeADNSEvent(const ADNSEvent& aev)
{
typename std::deque<ADNSEvent>::iterator i =
std::find(_adnsEvents.begin(), _adnsEvents.end(), aev);
if(i == _adnsEvents.end()) {
// not found
} else {
_adnsEvents.erase(i);
}
}
#endif // ENABLE_ASYNC_DNS
virtual EventRType getEvents() = 0;
sock_t getSocket() const
{
return _socket;
}
void setSocket(sock_t socket)
{
_socket = socket;
}
bool eventEmpty() const
{
#ifdef ENABLE_ASYNC_DNS
return _commandEvents.empty() && _adnsEvents.empty();
#else // !ENABLE_ASYNC_DNS
return _commandEvents.empty();
#endif // !ENABLE_ASYNC_DNS)
}
void processEvents(int events)
{
std::for_each(_commandEvents.begin(), _commandEvents.end(),
std::bind2nd(std::mem_fun_ref
(&CommandEvent::processEvents),
events));
#ifdef ENABLE_ASYNC_DNS
std::for_each(_adnsEvents.begin(), _adnsEvents.end(),
std::bind2nd(std::mem_fun_ref
(&ADNSEvent::processEvents),
events));
#endif // ENABLE_ASYNC_DNS
}
};
#ifdef ENABLE_ASYNC_DNS
template<typename EventPoll>
class AsyncNameResolverEntry {
private:
SharedHandle<AsyncNameResolver> _nameResolver;
Command* _command;
size_t _socketsSize;
sock_t _sockets[ARES_GETSOCK_MAXNUM];
public:
AsyncNameResolverEntry(const SharedHandle<AsyncNameResolver>& nameResolver,
Command* command):
_nameResolver(nameResolver), _command(command), _socketsSize(0) {}
bool operator==(const AsyncNameResolverEntry& entry)
{
return _nameResolver == entry._nameResolver &&
_command == entry._command;
}
void addSocketEvents(EventPoll* e)
{
_socketsSize = 0;
int mask = _nameResolver->getsock(_sockets);
if(mask == 0) {
return;
}
size_t i;
for(i = 0; i < ARES_GETSOCK_MAXNUM; ++i) {
int events = 0;
if(ARES_GETSOCK_READABLE(mask, i)) {
events |= EventPoll::IEV_READ;
}
if(ARES_GETSOCK_WRITABLE(mask, i)) {
events |= EventPoll::IEV_WRITE;
}
if(events == 0) {
// assume no further sockets are returned.
break;
}
e->addEvents(_sockets[i], _command, events, _nameResolver);
}
_socketsSize = i;
}
void removeSocketEvents(EventPoll* e)
{
for(size_t i = 0; i < _socketsSize; ++i) {
e->deleteEvents(_sockets[i], _command, _nameResolver);
}
}
// Calls AsyncNameResolver::process(ARES_SOCKET_BAD,
// ARES_SOCKET_BAD).
void processTimeout()
{
_nameResolver->process(ARES_SOCKET_BAD, ARES_SOCKET_BAD);
}
};
#else // !ENABLE_ASYNC_DNS
template<typename EventPoll>
class AsyncNameResolverEntry {};
#endif // !ENABLE_ASYNC_DNS
} // namespace aria2
#endif // _D_EVENT_H_

3
src/Makefile.am
View File

@ -201,7 +201,8 @@ SRCS = Socket.h\
wallclock.h\
download_helper.cc download_helper.h\
MetadataInfo.cc MetadataInfo.h\
SessionSerializer.cc SessionSerializer.h
SessionSerializer.cc SessionSerializer.h\
Event.h
if ENABLE_XML_RPC
SRCS += XmlRpcRequestParserController.cc XmlRpcRequestParserController.h\

4
src/Makefile.in
View File

@ -434,7 +434,7 @@ am__libaria2c_a_SOURCES_DIST = Socket.h SocketCore.cc SocketCore.h \
CreateRequestCommand.h DownloadResultCode.h wallclock.h \
download_helper.cc download_helper.h MetadataInfo.cc \
MetadataInfo.h SessionSerializer.cc SessionSerializer.h \
XmlRpcRequestParserController.cc \
Event.h XmlRpcRequestParserController.cc \
XmlRpcRequestParserController.h \
XmlRpcRequestParserStateMachine.cc \
XmlRpcRequestParserStateMachine.h XmlRpcRequestParserState.h \
@ -1204,7 +1204,7 @@ SRCS = Socket.h SocketCore.cc SocketCore.h BinaryStream.h Command.cc \
CreateRequestCommand.h DownloadResultCode.h wallclock.h \
download_helper.cc download_helper.h MetadataInfo.cc \
MetadataInfo.h SessionSerializer.cc SessionSerializer.h \
$(am__append_1) $(am__append_2) $(am__append_3) \
Event.h $(am__append_1) $(am__append_2) $(am__append_3) \
$(am__append_4) $(am__append_5) $(am__append_6) \
$(am__append_7) $(am__append_8) $(am__append_9) \
$(am__append_10) $(am__append_11) $(am__append_12) \

292
src/PollEventPoll.cc
View File

@ -44,254 +44,34 @@
namespace aria2 {
PollEventPoll::CommandEvent::CommandEvent(Command* command, int events):
_command(command), _events(events) {}
PollEventPoll::KSocketEntry::KSocketEntry(sock_t s):
SocketEntry<KCommandEvent, KADNSEvent, struct pollfd>(s) {}
int PollEventPoll::CommandEvent::getEvents() const
{
return _events;
}
void PollEventPoll::CommandEvent::processEvents(int events)
{
if((_events&events) ||
((PollEventPoll::EVENT_ERROR|PollEventPoll::EVENT_HUP)&events)) {
_command->setStatusActive();
}
if(PollEventPoll::EVENT_READ&events) {
_command->readEventReceived();
}
if(PollEventPoll::EVENT_WRITE&events) {
_command->writeEventReceived();
}
if(PollEventPoll::EVENT_ERROR&events) {
_command->errorEventReceived();
}
if(PollEventPoll::EVENT_HUP&events) {
_command->hupEventReceived();
}
}
void PollEventPoll::CommandEvent::addSelf
(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->addCommandEvent(*this);
}
void PollEventPoll::CommandEvent::removeSelf
(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->removeCommandEvent(*this);
}
#ifdef ENABLE_ASYNC_DNS
PollEventPoll::ADNSEvent::ADNSEvent
(const SharedHandle<AsyncNameResolver>& resolver,
Command* command,
sock_t socket, int events):
_resolver(resolver), _command(command), _socket(socket), _events(events) {}
int PollEventPoll::ADNSEvent::getEvents() const
{
return _events;
}
void PollEventPoll::ADNSEvent::processEvents(int events)
{
ares_socket_t readfd;
ares_socket_t writefd;
if(events&(PollEventPoll::EVENT_READ|PollEventPoll::EVENT_ERROR|
PollEventPoll::EVENT_HUP)) {
readfd = _socket;
} else {
readfd = ARES_SOCKET_BAD;
}
if(events&(PollEventPoll::EVENT_WRITE|PollEventPoll::EVENT_ERROR|
PollEventPoll::EVENT_HUP)) {
writefd = _socket;
} else {
writefd = ARES_SOCKET_BAD;
}
_resolver->process(readfd, writefd);
_command->setStatusActive();
}
void PollEventPoll::ADNSEvent::addSelf
(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->addADNSEvent(*this);
}
void PollEventPoll::ADNSEvent::removeSelf
(const SharedHandle<SocketEntry>& socketEntry) const
{
socketEntry->removeADNSEvent(*this);
}
#endif // ENABLE_ASYNC_DNS
PollEventPoll::SocketEntry::SocketEntry(sock_t socket):_socket(socket)
{
memset(&_pollEvent, 0, sizeof(struct pollfd));
}
void PollEventPoll::SocketEntry::addCommandEvent(const CommandEvent& cev)
{
std::deque<CommandEvent>::iterator i = std::find(_commandEvents.begin(),
_commandEvents.end(),
cev);
if(i == _commandEvents.end()) {
_commandEvents.push_back(cev);
} else {
(*i).addEvents(cev.getEvents());
}
}
void PollEventPoll::SocketEntry::removeCommandEvent(const CommandEvent& cev)
{
std::deque<CommandEvent>::iterator i = std::find(_commandEvents.begin(),
_commandEvents.end(),
cev);
if(i == _commandEvents.end()) {
// not found
} else {
(*i).removeEvents(cev.getEvents());
if((*i).eventsEmpty()) {
_commandEvents.erase(i);
}
}
}
#ifdef ENABLE_ASYNC_DNS
void PollEventPoll::SocketEntry::addADNSEvent(const ADNSEvent& aev)
{
std::deque<ADNSEvent>::iterator i = std::find(_adnsEvents.begin(),
_adnsEvents.end(),
aev);
if(i == _adnsEvents.end()) {
_adnsEvents.push_back(aev);
}
}
void PollEventPoll::SocketEntry::removeADNSEvent(const ADNSEvent& aev)
{
std::deque<ADNSEvent>::iterator i = std::find(_adnsEvents.begin(),
_adnsEvents.end(),
aev);
if(i == _adnsEvents.end()) {
// not found
} else {
_adnsEvents.erase(i);
}
}
#endif // ENABLE_ASYNC_DNS
void PollEventPoll::SocketEntry::processEvents(int events)
{
std::for_each(_commandEvents.begin(), _commandEvents.end(),
std::bind2nd(std::mem_fun_ref
(&PollEventPoll::CommandEvent::processEvents),
events));
#ifdef ENABLE_ASYNC_DNS
std::for_each(_adnsEvents.begin(), _adnsEvents.end(),
std::bind2nd(std::mem_fun_ref
(&PollEventPoll::ADNSEvent::processEvents),
events));
#endif // ENABLE_ASYNC_DNS
}
bool PollEventPoll::SocketEntry::eventEmpty() const
{
#ifdef ENABLE_ASYNC_DNS
return _commandEvents.empty() && _adnsEvents.empty();
#else // !ENABLE_ASYNC_DNS
return _commandEvents.empty();
#endif // !ENABLE_ASYNC_DNS)
}
int accumulateEvent(int events, const PollEventPoll::Event& event)
int accumulateEvent(int events, const PollEventPoll::KEvent& event)
{
return events|event.getEvents();
}
struct pollfd& PollEventPoll::SocketEntry::getPollEvent()
struct pollfd PollEventPoll::KSocketEntry::getEvents()
{
_pollEvent.fd = _socket;
struct pollfd pollEvent;
pollEvent.fd = _socket;
#ifdef ENABLE_ASYNC_DNS
_pollEvent.events =
pollEvent.events =
std::accumulate(_adnsEvents.begin(),
_adnsEvents.end(),
std::accumulate(_commandEvents.begin(),
_commandEvents.end(), 0, accumulateEvent),
accumulateEvent);
#else // !ENABLE_ASYNC_DNS
_pollEvent.events =
pollEvent.events =
std::accumulate(_commandEvents.begin(), _commandEvents.end(), 0,
accumulateEvent);
#endif // !ENABLE_ASYNC_DNS
return _pollEvent;
pollEvent.revents = 0;
return pollEvent;
}
#ifdef ENABLE_ASYNC_DNS
PollEventPoll::AsyncNameResolverEntry::AsyncNameResolverEntry
(const SharedHandle<AsyncNameResolver>& nameResolver, Command* command):
_nameResolver(nameResolver), _command(command), _socketsSize(0)
{}
void PollEventPoll::AsyncNameResolverEntry::addSocketEvents(PollEventPoll* e)
{
_socketsSize = 0;
int mask = _nameResolver->getsock(_sockets);
if(mask == 0) {
return;
}
size_t i;
for(i = 0; i < ARES_GETSOCK_MAXNUM; ++i) {
int events = 0;
if(ARES_GETSOCK_READABLE(mask, i)) {
events |= PollEventPoll::EVENT_READ;
}
if(ARES_GETSOCK_WRITABLE(mask, i)) {
events |= PollEventPoll::EVENT_WRITE;
}
if(events == 0) {
// assume no further sockets are returned.
break;
}
LogFactory::getInstance()->debug("addSocketEvents, %d", _sockets[i]);
e->addEvents(_sockets[i], _command, events, _nameResolver);
}
_socketsSize = i;
}
void PollEventPoll::AsyncNameResolverEntry::removeSocketEvents(PollEventPoll* e)
{
for(size_t i = 0; i < _socketsSize; ++i) {
e->deleteEvents(_sockets[i], _command, _nameResolver);
}
}
void PollEventPoll::AsyncNameResolverEntry::processTimeout()
{
_nameResolver->process(ARES_SOCKET_BAD, ARES_SOCKET_BAD);
}
#endif // ENABLE_ASYNC_DNS
PollEventPoll::PollEventPoll():
_pollfdCapacity(1024), _pollfdNum(0), _logger(LogFactory::getInstance())
{
@ -311,12 +91,12 @@ void PollEventPoll::poll(const struct timeval& tv)
while((res = ::poll(_pollfds, _pollfdNum, timeout)) == -1 &&
errno == EINTR);
if(res > 0) {
SharedHandle<SocketEntry> se(new SocketEntry(0));
SharedHandle<KSocketEntry> se(new KSocketEntry(0));
for(struct pollfd* first = _pollfds, *last = _pollfds+_pollfdNum;
first != last; ++first) {
if(first->revents) {
se->setSocket(first->fd);
std::deque<SharedHandle<SocketEntry> >::iterator itr =
std::deque<SharedHandle<KSocketEntry> >::iterator itr =
std::lower_bound(_socketEntries.begin(), _socketEntries.end(), se);
if(itr != _socketEntries.end() && (*itr) == se) {
(*itr)->processEvents(first->revents);
@ -334,7 +114,7 @@ void PollEventPoll::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(std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator i =
for(std::deque<SharedHandle<KAsyncNameResolverEntry> >::iterator i =
_nameResolverEntries.begin(), eoi = _nameResolverEntries.end();
i != eoi; ++i) {
(*i)->processTimeout();
@ -351,32 +131,32 @@ int PollEventPoll::translateEvents(EventPoll::EventType events)
{
int newEvents = 0;
if(EventPoll::EVENT_READ&events) {
newEvents |= PollEventPoll::EVENT_READ;
newEvents |= IEV_READ;
}
if(EventPoll::EVENT_WRITE&events) {
newEvents |= PollEventPoll::EVENT_WRITE;
newEvents |= IEV_WRITE;
}
if(EventPoll::EVENT_ERROR&events) {
newEvents |= PollEventPoll::EVENT_ERROR;
newEvents |= IEV_ERROR;
}
if(EventPoll::EVENT_HUP&events) {
newEvents |= PollEventPoll::EVENT_HUP;
newEvents |= IEV_HUP;
}
return newEvents;
}
bool PollEventPoll::addEvents
(sock_t socket, const PollEventPoll::Event& event)
(sock_t socket, const PollEventPoll::KEvent& event)
{
SharedHandle<SocketEntry> socketEntry(new SocketEntry(socket));
std::deque<SharedHandle<SocketEntry> >::iterator i =
SharedHandle<KSocketEntry> socketEntry(new KSocketEntry(socket));
std::deque<SharedHandle<KSocketEntry> >::iterator i =
std::lower_bound(_socketEntries.begin(), _socketEntries.end(), socketEntry);
if(i != _socketEntries.end() && (*i) == socketEntry) {
event.addSelf(*i);
for(struct pollfd* first = _pollfds, *last = _pollfds+_pollfdNum;
first != last; ++first) {
if(first->fd == socket) {
*first = (*i)->getPollEvent();
*first = (*i)->getEvents();
break;
}
}
@ -390,7 +170,7 @@ bool PollEventPoll::addEvents
delete [] _pollfds;
_pollfds = newPollfds;
}
_pollfds[_pollfdNum] = socketEntry->getPollEvent();
_pollfds[_pollfdNum] = socketEntry->getEvents();
++_pollfdNum;
}
return true;
@ -400,7 +180,7 @@ bool PollEventPoll::addEvents
(sock_t socket, Command* command, EventPoll::EventType events)
{
int pollEvents = translateEvents(events);
return addEvents(socket, CommandEvent(command, pollEvents));
return addEvents(socket, KCommandEvent(command, pollEvents));
}
#ifdef ENABLE_ASYNC_DNS
@ -408,15 +188,15 @@ bool PollEventPoll::addEvents
(sock_t socket, Command* command, int events,
const SharedHandle<AsyncNameResolver>& rs)
{
return addEvents(socket, ADNSEvent(rs, command, socket, events));
return addEvents(socket, KADNSEvent(rs, command, socket, events));
}
#endif // ENABLE_ASYNC_DNS
bool PollEventPoll::deleteEvents
(sock_t socket, const PollEventPoll::Event& event)
(sock_t socket, const PollEventPoll::KEvent& event)
{
SharedHandle<SocketEntry> socketEntry(new SocketEntry(socket));
std::deque<SharedHandle<SocketEntry> >::iterator i =
SharedHandle<KSocketEntry> socketEntry(new KSocketEntry(socket));
std::deque<SharedHandle<KSocketEntry> >::iterator i =
std::lower_bound(_socketEntries.begin(), _socketEntries.end(), socketEntry);
if(i != _socketEntries.end() && (*i) == socketEntry) {
event.removeSelf(*i);
@ -430,7 +210,7 @@ bool PollEventPoll::deleteEvents
--_pollfdNum;
_socketEntries.erase(i);
} else {
*first = (*i)->getPollEvent();
*first = (*i)->getEvents();
}
break;
}
@ -448,7 +228,7 @@ bool PollEventPoll::deleteEvents
bool PollEventPoll::deleteEvents
(sock_t socket, Command* command, const SharedHandle<AsyncNameResolver>& rs)
{
return deleteEvents(socket, ADNSEvent(rs, command, socket, 0));
return deleteEvents(socket, KADNSEvent(rs, command, socket, 0));
}
#endif // ENABLE_ASYNC_DNS
@ -456,16 +236,16 @@ bool PollEventPoll::deleteEvents
(sock_t socket, Command* command, EventPoll::EventType events)
{
int pollEvents = translateEvents(events);
return deleteEvents(socket, CommandEvent(command, pollEvents));
return deleteEvents(socket, KCommandEvent(command, pollEvents));
}
#ifdef ENABLE_ASYNC_DNS
bool PollEventPoll::addNameResolver
(const SharedHandle<AsyncNameResolver>& resolver, Command* command)
{
SharedHandle<AsyncNameResolverEntry> entry
(new AsyncNameResolverEntry(resolver, command));
std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator itr =
SharedHandle<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
std::deque<SharedHandle<KAsyncNameResolverEntry> >::iterator itr =
std::find(_nameResolverEntries.begin(), _nameResolverEntries.end(), entry);
if(itr == _nameResolverEntries.end()) {
_nameResolverEntries.push_back(entry);
@ -479,9 +259,9 @@ bool PollEventPoll::addNameResolver
bool PollEventPoll::deleteNameResolver
(const SharedHandle<AsyncNameResolver>& resolver, Command* command)
{
SharedHandle<AsyncNameResolverEntry> entry
(new AsyncNameResolverEntry(resolver, command));
std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator itr =
SharedHandle<KAsyncNameResolverEntry> entry
(new KAsyncNameResolverEntry(resolver, command));
std::deque<SharedHandle<KAsyncNameResolverEntry> >::iterator itr =
std::find(_nameResolverEntries.begin(), _nameResolverEntries.end(), entry);
if(itr == _nameResolverEntries.end()) {
return false;

215
src/PollEventPoll.h
View File

@ -41,6 +41,7 @@
#include <deque>
#include "Event.h"
#ifdef ENABLE_ASYNC_DNS
# include "AsyncNameResolver.h"
#endif // ENABLE_ASYNC_DNS
@ -50,195 +51,29 @@ namespace aria2 {
class Logger;
class PollEventPoll : public EventPoll {
private:
class KSocketEntry;
typedef Event<KSocketEntry> KEvent;
typedef CommandEvent<KSocketEntry, PollEventPoll> KCommandEvent;
typedef ADNSEvent<KSocketEntry, PollEventPoll> KADNSEvent;
typedef AsyncNameResolverEntry<PollEventPoll> KAsyncNameResolverEntry;
friend class AsyncNameResolverEntry<PollEventPoll>;
class KSocketEntry:
public SocketEntry<KCommandEvent, KADNSEvent, struct pollfd> {
public:
KSocketEntry(sock_t socket);
virtual struct pollfd getEvents();
};
friend int accumulateEvent(int events, const KEvent& event);
private:
class SocketEntry;
class Event {
public:
virtual ~Event() {}
virtual void processEvents(int events) = 0;
virtual int getEvents() const = 0;
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const =0;
virtual void removeSelf
(const SharedHandle<SocketEntry>& socketEntry) const =0;
};
friend int accumulateEvent(int events, const Event& event);
class CommandEvent : public Event {
private:
Command* _command;
int _events;
public:
CommandEvent(Command* command, int events);
Command* getCommand() const
{
return _command;
}
void addEvents(int events)
{
_events |= events;
}
void removeEvents(int events)
{
_events &= (~events);
}
bool eventsEmpty() const
{
return _events == 0;
}
bool operator==(const CommandEvent& commandEvent) const
{
return _command == commandEvent._command;
}
virtual int getEvents() const;
virtual void processEvents(int events);
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const;
virtual void removeSelf(const SharedHandle<SocketEntry>& socketEntry) const;
};
std::deque<SharedHandle<KSocketEntry> > _socketEntries;
#ifdef ENABLE_ASYNC_DNS
class ADNSEvent : public Event {
private:
SharedHandle<AsyncNameResolver> _resolver;
Command* _command;
sock_t _socket;
int _events;
public:
ADNSEvent(const SharedHandle<AsyncNameResolver>& resolver, Command* command,
sock_t socket, int events);
bool operator==(const ADNSEvent& event) const
{
return _resolver == event._resolver;
}
virtual int getEvents() const;
virtual void processEvents(int events);
virtual void addSelf(const SharedHandle<SocketEntry>& socketEntry) const;
virtual void removeSelf(const SharedHandle<SocketEntry>& socketEntry) const;
};
#endif // ENABLE_ASYNC_DNS
class SocketEntry {
private:
sock_t _socket;
std::deque<CommandEvent> _commandEvents;
#ifdef ENABLE_ASYNC_DNS
std::deque<ADNSEvent> _adnsEvents;
#endif // ENABLE_ASYNC_DNS
struct pollfd _pollEvent;
public:
SocketEntry(sock_t socket);
bool operator==(const SocketEntry& entry) const
{
return _socket == entry._socket;
}
bool operator<(const SocketEntry& entry) const
{
return _socket < entry._socket;
}
void addCommandEvent(const CommandEvent& cev);
void removeCommandEvent(const CommandEvent& cev);
#ifdef ENABLE_ASYNC_DNS
void addADNSEvent(const ADNSEvent& aev);
void removeADNSEvent(const ADNSEvent& aev);
#endif // ENABLE_ASYNC_DNS
struct pollfd& getPollEvent();
sock_t getSocket() const
{
return _socket;
}
void setSocket(sock_t socket)
{
_socket = socket;
}
bool eventEmpty() const;
void processEvents(int events);
};
#ifdef ENABLE_ASYNC_DNS
class AsyncNameResolverEntry {
private:
SharedHandle<AsyncNameResolver> _nameResolver;
Command* _command;
size_t _socketsSize;
sock_t _sockets[ARES_GETSOCK_MAXNUM];
public:
AsyncNameResolverEntry(const SharedHandle<AsyncNameResolver>& nameResolver,
Command* command);
bool operator==(const AsyncNameResolverEntry& entry)
{
return _nameResolver == entry._nameResolver &&
_command == entry._command;
}
void addSocketEvents(PollEventPoll* socketPoll);
void removeSocketEvents(PollEventPoll* socketPoll);
// Calls AsyncNameResolver::process(ARES_SOCKET_BAD,
// ARES_SOCKET_BAD).
void processTimeout();
};
#endif // ENABLE_ASYNC_DNS
private:
enum PollEventType {
EVENT_READ = POLLIN,
EVENT_WRITE = POLLOUT,
EVENT_ERROR = POLLERR,
EVENT_HUP = POLLHUP,
};
std::deque<SharedHandle<SocketEntry> > _socketEntries;
#ifdef ENABLE_ASYNC_DNS
std::deque<SharedHandle<AsyncNameResolverEntry> > _nameResolverEntries;
std::deque<SharedHandle<KAsyncNameResolverEntry> > _nameResolverEntries;
#endif // ENABLE_ASYNC_DNS
// Allocated the number of struct pollfd in _pollfds.
@ -251,9 +86,9 @@ private:
Logger* _logger;
bool addEvents(sock_t socket, const Event& event);
bool addEvents(sock_t socket, const KEvent& event);
bool deleteEvents(sock_t socket, const Event& event);
bool deleteEvents(sock_t socket, const KEvent& event);
bool addEvents(sock_t socket, Command* command, int events,
const SharedHandle<AsyncNameResolver>& rs);
@ -282,6 +117,10 @@ public:
(const SharedHandle<AsyncNameResolver>& resolver, Command* command);
#endif // ENABLE_ASYNC_DNS
static const int IEV_READ = POLLIN;
static const int IEV_WRITE = POLLOUT;
static const int IEV_ERROR = POLLERR;
static const int IEV_HUP = POLLHUP;
};
} // namespace aria2