Unverified Commit 1e116eb0 authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #504 from knowledge4igor/refactoring_01

Refactoing in Client and Epoll
parents 894f5d01 65ba973e
...@@ -260,6 +260,9 @@ private: ...@@ -260,6 +260,9 @@ private:
void handleRequestsQueue(); void handleRequestsQueue();
void handleConnectionQueue(); void handleConnectionQueue();
void handleReadableEntry(const Aio::FdSet::Entry& entry);
void handleWritableEntry(const Aio::FdSet::Entry& entry);
void handleHangupEntry(const Aio::FdSet::Entry& entry);
void handleIncoming(std::shared_ptr<Connection> connection); void handleIncoming(std::shared_ptr<Connection> connection);
void handleResponsePacket(const std::shared_ptr<Connection>& connection, const char* buffer, size_t totalBytes); void handleResponsePacket(const std::shared_ptr<Connection>& connection, const char* buffer, size_t totalBytes);
void handleTimeout(const std::shared_ptr<Connection>& connection); void handleTimeout(const std::shared_ptr<Connection>& connection);
......
...@@ -98,7 +98,7 @@ struct Event { ...@@ -98,7 +98,7 @@ struct Event {
class Epoll { class Epoll {
public: public:
explicit Epoll(size_t max = 128); Epoll();
void addFd(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level); void addFd(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level);
void addFdOneShot(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level); void addFdOneShot(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level);
...@@ -107,8 +107,7 @@ public: ...@@ -107,8 +107,7 @@ public:
void rearmFd(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level); void rearmFd(Fd fd, Flags<NotifyOn> interest, Tag tag, Mode mode = Mode::Level);
int poll(std::vector<Event>& events, int poll(std::vector<Event>& events,
size_t maxEvents = Const::MaxEvents, const std::chrono::milliseconds timeout = std::chrono::milliseconds(-1)) const;
std::chrono::milliseconds timeout = std::chrono::milliseconds(0)) const;
private: private:
static int toEpollEvents(const Flags<NotifyOn>& interest); static int toEpollEvents(const Flags<NotifyOn>& interest);
......
...@@ -132,55 +132,15 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -132,55 +132,15 @@ Transport::onReady(const Aio::FdSet& fds) {
else if (entry.getTag() == requestsQueue.tag()) { else if (entry.getTag() == requestsQueue.tag()) {
handleRequestsQueue(); handleRequestsQueue();
} }
else if (entry.isReadable()) { else if (entry.isReadable()) {
auto tag = entry.getTag(); handleReadableEntry(entry);
auto fd = tag.value();
auto connIt = connections.find(fd);
if (connIt != std::end(connections)) {
auto connection = connIt->second.connection.lock();
if (connection) {
handleIncoming(connection);
}
else {
throw std::runtime_error("Connection error: problem with reading data from server");
}
}
else {
Guard guard(timeoutsLock);
auto timerIt = timeouts.find(fd);
if (timerIt != std::end(timeouts))
{
auto connection = timerIt->second.lock();
if (connection)
{
handleTimeout(connection);
timeouts.erase(fd);
}
}
}
}
else {
auto tag = entry.getTag();
auto fd = tag.value();
auto connIt = connections.find(fd);
if (connIt != std::end(connections)) {
auto& connectionEntry = connIt->second;
if (entry.isHangup())
connectionEntry.reject(Error::system("Could not connect"));
else {
auto connection = connIt->second.connection.lock();
if (connection) {
connectionEntry.resolve();
// We are connected, we can start reading data now
reactor()->modifyFd(key(), connection->fd(), NotifyOn::Read);
} else {
connectionEntry.reject(Error::system("Connection lost"));
}
} }
else if (entry.isWritable()) {
handleWritableEntry(entry);
} else if (entry.isHangup()) {
handleHangupEntry(entry);
} else { } else {
throw std::runtime_error("Unknown fd"); assert(false && "Unexpected event in entry");
}
} }
} }
} }
...@@ -300,6 +260,87 @@ Transport::handleConnectionQueue() { ...@@ -300,6 +260,87 @@ Transport::handleConnectionQueue() {
} }
} }
void Transport::handleReadableEntry(const Aio::FdSet::Entry& entry)
{
assert(entry.isReadable() && "Entry must be readable");
auto tag = entry.getTag();
const Fd fd = tag.value();
auto connIt = connections.find(fd);
if (connIt != std::end(connections))
{
auto connection = connIt->second.connection.lock();
if (connection)
{
handleIncoming(connection);
}
else
{
throw std::runtime_error("Connection error: problem with reading data from server");
}
}
else
{
Guard guard(timeoutsLock);
auto timerIt = timeouts.find(fd);
if (timerIt != std::end(timeouts))
{
auto connection = timerIt->second.lock();
if (connection)
{
handleTimeout(connection);
timeouts.erase(fd);
}
}
}
}
void Transport::handleWritableEntry(const Aio::FdSet::Entry& entry)
{
assert(entry.isWritable() && "Entry must be writable");
auto tag = entry.getTag();
const Fd fd = tag.value();
auto connIt = connections.find(fd);
if (connIt != std::end(connections))
{
auto& connectionEntry = connIt->second;
auto connection = connIt->second.connection.lock();
if (connection)
{
connectionEntry.resolve();
// We are connected, we can start reading data now
reactor()->modifyFd(key(), connection->fd(), NotifyOn::Read);
}
else
{
connectionEntry.reject(Error::system("Connection lost"));
}
}
else
{
throw std::runtime_error("Unknown fd");
}
}
void Transport::handleHangupEntry(const Aio::FdSet::Entry& entry)
{
assert(entry.isHangup() && "Entry must be hangup");
auto tag = entry.getTag();
const Fd fd = tag.value();
auto connIt = connections.find(fd);
if (connIt != std::end(connections))
{
auto& connectionEntry = connIt->second;
connectionEntry.reject(Error::system("Could not connect"));
}
else
{
throw std::runtime_error("Unknown fd");
}
}
void void
Transport::handleIncoming(std::shared_ptr<Connection> connection) { Transport::handleIncoming(std::shared_ptr<Connection> connection) {
char buffer[Const::MaxBuffer] = {0}; char buffer[Const::MaxBuffer] = {0};
......
...@@ -143,8 +143,8 @@ namespace Polling { ...@@ -143,8 +143,8 @@ namespace Polling {
, tag(_tag) , tag(_tag)
{ } { }
Epoll::Epoll(size_t max) { Epoll::Epoll() {
epoll_fd = TRY_RET(epoll_create(max)); epoll_fd = TRY_RET(epoll_create(Const::MaxEvents));
} }
void void
...@@ -188,12 +188,12 @@ namespace Polling { ...@@ -188,12 +188,12 @@ namespace Polling {
} }
int int
Epoll::poll(std::vector<Event>& events, size_t maxEvents, std::chrono::milliseconds timeout) const { Epoll::poll(std::vector<Event>& events, const std::chrono::milliseconds timeout) const {
struct epoll_event evs[Const::MaxEvents]; struct epoll_event evs[Const::MaxEvents];
int ready_fds = -1; int ready_fds = -1;
do { do {
ready_fds = epoll_wait(epoll_fd, evs, maxEvents, timeout.count()); ready_fds = epoll_wait(epoll_fd, evs, Const::MaxEvents, timeout.count());
} while (ready_fds < 0 && errno == EINTR); } while (ready_fds < 0 && errno == EINTR);
for (int i = 0; i < ready_fds; ++i) { for (int i = 0; i < ready_fds; ++i) {
......
...@@ -135,20 +135,17 @@ public: ...@@ -135,20 +135,17 @@ public:
if (handlers_.empty()) if (handlers_.empty())
throw std::runtime_error("You need to set at least one handler"); throw std::runtime_error("You need to set at least one handler");
std::chrono::milliseconds timeout(-1);
for (;;) { for (;;) {
std::vector<Polling::Event> events; std::vector<Polling::Event> events;
int ready_fds; int ready_fds = poller.poll(events);
switch (ready_fds = poller.poll(events, 1024, timeout)) {
switch (ready_fds) {
case -1: break; case -1: break;
case 0: break; case 0: break;
default: default:
if (shutdown_) return; if (shutdown_) return;
handleFds(std::move(events)); handleFds(std::move(events));
timeout = std::chrono::milliseconds(-1);
} }
} }
} }
......
...@@ -258,8 +258,8 @@ Listener::run() { ...@@ -258,8 +258,8 @@ Listener::run() {
for (;;) { for (;;) {
std::vector<Polling::Event> events; std::vector<Polling::Event> events;
int ready_fds = poller.poll(events);
int ready_fds = poller.poll(events, 128, std::chrono::milliseconds(-1));
if (ready_fds == -1) { if (ready_fds == -1) {
if (errno == EINTR && g_listen_fd == -1) return; if (errno == EINTR && g_listen_fd == -1) return;
throw Error::system("Polling"); throw Error::system("Polling");
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment