SelectEventPoll.cc 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361
  1. /* <!-- copyright */
  2. /*
  3. * aria2 - The high speed download utility
  4. *
  5. * Copyright (C) 2009 Tatsuhiro Tsujikawa
  6. *
  7. * This program is free software; you can redistribute it and/or modify
  8. * it under the terms of the GNU General Public License as published by
  9. * the Free Software Foundation; either version 2 of the License, or
  10. * (at your option) any later version.
  11. *
  12. * This program is distributed in the hope that it will be useful,
  13. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. * GNU General Public License for more details.
  16. *
  17. * You should have received a copy of the GNU General Public License
  18. * along with this program; if not, write to the Free Software
  19. * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  20. *
  21. * In addition, as a special exception, the copyright holders give
  22. * permission to link the code of portions of this program with the
  23. * OpenSSL library under certain conditions as described in each
  24. * individual source file, and distribute linked combinations
  25. * including the two.
  26. * You must obey the GNU General Public License in all respects
  27. * for all of the code used other than OpenSSL. If you modify
  28. * file(s) with this exception, you may extend this exception to your
  29. * version of the file(s), but you are not obligated to do so. If you
  30. * do not wish to do so, delete this exception statement from your
  31. * version. If you delete this exception statement from all source
  32. * files in the program, then also delete it here.
  33. */
  34. /* copyright --> */
  35. #include "SelectEventPoll.h"
  36. #include <cstring>
  37. #include <algorithm>
  38. #include <numeric>
  39. #include <iostream>
  40. #include "Command.h"
  41. #include "LogFactory.h"
  42. #include "Logger.h"
  43. namespace aria2 {
  44. SelectEventPoll::CommandEvent::CommandEvent(Command* command, int events):
  45. _command(command), _events(events) {}
  46. bool SelectEventPoll::CommandEvent::operator==(const CommandEvent& commandEvent) const
  47. {
  48. return _command == commandEvent._command;
  49. }
  50. Command* SelectEventPoll::CommandEvent::getCommand() const
  51. {
  52. return _command;
  53. }
  54. int SelectEventPoll::CommandEvent::getEvents() const
  55. {
  56. return _events;
  57. }
  58. void SelectEventPoll::CommandEvent::addEvents(int events)
  59. {
  60. _events |= events;
  61. }
  62. void SelectEventPoll::CommandEvent::removeEvents(int events)
  63. {
  64. _events &= (~events);
  65. }
  66. bool SelectEventPoll::CommandEvent::eventsEmpty() const
  67. {
  68. return _events == 0;
  69. }
  70. void SelectEventPoll::CommandEvent::processEvents(int events)
  71. {
  72. if((_events&events) ||
  73. ((EventPoll::EVENT_ERROR|EventPoll::EVENT_HUP)&events)) {
  74. _command->setStatusActive();
  75. }
  76. if(EventPoll::EVENT_READ&events) {
  77. _command->readEventReceived();
  78. }
  79. if(EventPoll::EVENT_WRITE&events) {
  80. _command->writeEventReceived();
  81. }
  82. if(EventPoll::EVENT_ERROR&events) {
  83. _command->errorEventReceived();
  84. }
  85. if(EventPoll::EVENT_HUP&events) {
  86. _command->hupEventReceived();
  87. }
  88. }
  89. SelectEventPoll::SocketEntry::SocketEntry(sock_t socket):_socket(socket) {}
  90. bool SelectEventPoll::SocketEntry::operator==(const SocketEntry& entry) const
  91. {
  92. return _socket == entry._socket;
  93. }
  94. bool SelectEventPoll::SocketEntry::operator<(const SocketEntry& entry) const
  95. {
  96. return _socket < entry._socket;
  97. }
  98. void SelectEventPoll::SocketEntry::addCommandEvent
  99. (Command* command, int events)
  100. {
  101. CommandEvent cev(command, events);
  102. std::deque<CommandEvent>::iterator i = std::find(_commandEvents.begin(),
  103. _commandEvents.end(),
  104. cev);
  105. if(i == _commandEvents.end()) {
  106. _commandEvents.push_back(cev);
  107. } else {
  108. (*i).addEvents(events);
  109. }
  110. }
  111. void SelectEventPoll::SocketEntry::removeCommandEvent
  112. (Command* command, int events)
  113. {
  114. CommandEvent cev(command, events);
  115. std::deque<CommandEvent>::iterator i = std::find(_commandEvents.begin(),
  116. _commandEvents.end(),
  117. cev);
  118. if(i == _commandEvents.end()) {
  119. // not found
  120. } else {
  121. (*i).removeEvents(events);
  122. if((*i).eventsEmpty()) {
  123. _commandEvents.erase(i);
  124. }
  125. }
  126. }
  127. void SelectEventPoll::SocketEntry::processEvents(int events)
  128. {
  129. std::for_each(_commandEvents.begin(), _commandEvents.end(),
  130. std::bind2nd(std::mem_fun_ref(&CommandEvent::processEvents),
  131. events));
  132. }
  133. sock_t SelectEventPoll::SocketEntry::getSocket() const
  134. {
  135. return _socket;
  136. }
  137. bool SelectEventPoll::SocketEntry::eventEmpty() const
  138. {
  139. return _commandEvents.empty();
  140. }
  141. int accumulateEvent(int events, const SelectEventPoll::CommandEvent& event)
  142. {
  143. return events|event.getEvents();
  144. }
  145. int SelectEventPoll::SocketEntry::getEvents()
  146. {
  147. return
  148. std::accumulate(_commandEvents.begin(), _commandEvents.end(), 0,
  149. accumulateEvent);
  150. }
  151. #ifdef ENABLE_ASYNC_DNS
  152. SelectEventPoll::AsyncNameResolverEntry::AsyncNameResolverEntry
  153. (const SharedHandle<AsyncNameResolver>& nameResolver, Command* command):
  154. _nameResolver(nameResolver), _command(command) {}
  155. bool SelectEventPoll::AsyncNameResolverEntry::operator==
  156. (const AsyncNameResolverEntry& entry)
  157. {
  158. return _nameResolver == entry._nameResolver &&
  159. _command == entry._command;
  160. }
  161. int SelectEventPoll::AsyncNameResolverEntry::getFds
  162. (fd_set* rfdsPtr, fd_set* wfdsPtr)
  163. {
  164. return _nameResolver->getFds(rfdsPtr, wfdsPtr);
  165. }
  166. void SelectEventPoll::AsyncNameResolverEntry::process
  167. (fd_set* rfdsPtr, fd_set* wfdsPtr)
  168. {
  169. _nameResolver->process(rfdsPtr, wfdsPtr);
  170. switch(_nameResolver->getStatus()) {
  171. case AsyncNameResolver::STATUS_SUCCESS:
  172. case AsyncNameResolver::STATUS_ERROR:
  173. _command->setStatusActive();
  174. break;
  175. default:
  176. break;
  177. }
  178. }
  179. #endif // ENABLE_ASYNC_DNS
  180. SelectEventPoll::SelectEventPoll()
  181. {
  182. updateFdSet();
  183. }
  184. SelectEventPoll::~SelectEventPoll() {}
  185. void SelectEventPoll::poll(const struct timeval& tv)
  186. {
  187. fd_set rfds;
  188. fd_set wfds;
  189. memcpy(&rfds, &_rfdset, sizeof(fd_set));
  190. memcpy(&wfds, &_wfdset, sizeof(fd_set));
  191. #ifdef ENABLE_ASYNC_DNS
  192. for(std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator itr =
  193. _nameResolverEntries.begin(); itr != _nameResolverEntries.end();
  194. ++itr) {
  195. SharedHandle<AsyncNameResolverEntry>& entry = *itr;
  196. int fd = entry->getFds(&rfds, &wfds);
  197. // TODO force error if fd == 0
  198. if(_fdmax < fd) {
  199. _fdmax = fd;
  200. }
  201. }
  202. #endif // ENABLE_ASYNC_DNS
  203. int retval;
  204. do {
  205. struct timeval ttv = tv;
  206. retval = select(_fdmax+1, &rfds, &wfds, NULL, &ttv);
  207. } while(retval == -1 && errno == EINTR);
  208. if(retval > 0) {
  209. for(std::deque<SharedHandle<SocketEntry> >::iterator i =
  210. _socketEntries.begin(); i != _socketEntries.end(); ++i) {
  211. int events = 0;
  212. if(FD_ISSET((*i)->getSocket(), &rfds)) {
  213. events |= EventPoll::EVENT_READ;
  214. }
  215. if(FD_ISSET((*i)->getSocket(), &wfds)) {
  216. events |= EventPoll::EVENT_WRITE;
  217. }
  218. (*i)->processEvents(events);
  219. }
  220. }
  221. #ifdef ENABLE_ASYNC_DNS
  222. for(std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator i =
  223. _nameResolverEntries.begin(); i != _nameResolverEntries.end(); ++i) {
  224. (*i)->process(&rfds, &wfds);
  225. }
  226. #endif // ENABLE_ASYNC_DNS
  227. }
  228. void SelectEventPoll::updateFdSet()
  229. {
  230. _fdmax = 0;
  231. FD_ZERO(&_rfdset);
  232. FD_ZERO(&_wfdset);
  233. for(std::deque<SharedHandle<SocketEntry> >::iterator i =
  234. _socketEntries.begin(); i != _socketEntries.end(); ++i) {
  235. sock_t fd = (*i)->getSocket();
  236. int events = (*i)->getEvents();
  237. if(events&EventPoll::EVENT_READ) {
  238. FD_SET(fd, &_rfdset);
  239. }
  240. if(events&EventPoll::EVENT_WRITE) {
  241. FD_SET(fd, &_wfdset);
  242. }
  243. if(_fdmax < fd) {
  244. _fdmax = fd;
  245. }
  246. }
  247. }
  248. bool SelectEventPoll::addEvents(sock_t socket, Command* command,
  249. EventPoll::EventType events)
  250. {
  251. SharedHandle<SocketEntry> socketEntry(new SocketEntry(socket));
  252. std::deque<SharedHandle<SocketEntry> >::iterator i =
  253. std::lower_bound(_socketEntries.begin(), _socketEntries.end(), socketEntry);
  254. int r = 0;
  255. if(i != _socketEntries.end() && (*i) == socketEntry) {
  256. (*i)->addCommandEvent(command, events);
  257. } else {
  258. _socketEntries.insert(i, socketEntry);
  259. socketEntry->addCommandEvent(command, events);
  260. }
  261. updateFdSet();
  262. if(r == -1) {
  263. _logger->debug("Failed to add socket event %d:%s", socket, strerror(errno));
  264. return false;
  265. } else {
  266. return true;
  267. }
  268. }
  269. bool SelectEventPoll::deleteEvents(sock_t socket, Command* command,
  270. EventPoll::EventType events)
  271. {
  272. SharedHandle<SocketEntry> socketEntry(new SocketEntry(socket));
  273. std::deque<SharedHandle<SocketEntry> >::iterator i =
  274. std::lower_bound(_socketEntries.begin(), _socketEntries.end(), socketEntry);
  275. if(i != _socketEntries.end() && (*i) == socketEntry) {
  276. (*i)->removeCommandEvent(command, events);
  277. int r = 0;
  278. if((*i)->eventEmpty()) {
  279. _socketEntries.erase(i);
  280. }
  281. updateFdSet();
  282. if(r == -1) {
  283. _logger->debug("Failed to delete socket event:%s", strerror(errno));
  284. return false;
  285. } else {
  286. return true;
  287. }
  288. } else {
  289. _logger->debug("Socket %d is not found in SocketEntries.", socket);
  290. return false;
  291. }
  292. }
  293. #ifdef ENABLE_ASYNC_DNS
  294. bool SelectEventPoll::addNameResolver
  295. (const SharedHandle<AsyncNameResolver>& resolver, Command* command)
  296. {
  297. SharedHandle<AsyncNameResolverEntry> entry
  298. (new AsyncNameResolverEntry(resolver, command));
  299. std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator itr =
  300. std::find(_nameResolverEntries.begin(), _nameResolverEntries.end(), entry);
  301. if(itr == _nameResolverEntries.end()) {
  302. _nameResolverEntries.push_back(entry);
  303. return true;
  304. } else {
  305. return false;
  306. }
  307. }
  308. bool SelectEventPoll::deleteNameResolver
  309. (const SharedHandle<AsyncNameResolver>& resolver, Command* command)
  310. {
  311. SharedHandle<AsyncNameResolverEntry> entry
  312. (new AsyncNameResolverEntry(resolver, command));
  313. std::deque<SharedHandle<AsyncNameResolverEntry> >::iterator itr =
  314. std::find(_nameResolverEntries.begin(), _nameResolverEntries.end(), entry);
  315. if(itr == _nameResolverEntries.end()) {
  316. return false;
  317. } else {
  318. _nameResolverEntries.erase(itr);
  319. return true;
  320. }
  321. }
  322. #endif // ENABLE_ASYNC_DNS
  323. } // namespace aria2