Unverified Commit ded5e71e authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #411 from knowledge4igor/first_step_of_memory_leaks_fix_in_client

First step of fixing memory leaks in Client
parents 1200e55c 35298e3c
...@@ -196,7 +196,7 @@ public: ...@@ -196,7 +196,7 @@ public:
void registerPoller(Polling::Epoll& poller) override; void registerPoller(Polling::Epoll& poller) override;
Async::Promise<void> Async::Promise<void>
asyncConnect(const std::shared_ptr<Connection>& connection, const struct sockaddr* address, socklen_t addr_len); asyncConnect(std::shared_ptr<Connection> connection, const struct sockaddr* address, socklen_t addr_len);
Async::Promise<ssize_t> asyncSendRequest( Async::Promise<ssize_t> asyncSendRequest(
const std::shared_ptr<Connection>& connection, const std::shared_ptr<Connection>& connection,
...@@ -216,14 +216,14 @@ private: ...@@ -216,14 +216,14 @@ private:
std::shared_ptr<Connection> connection, const struct sockaddr* addr, socklen_t addr_len) std::shared_ptr<Connection> connection, const struct sockaddr* addr, socklen_t addr_len)
: resolve(std::move(resolve)) : resolve(std::move(resolve))
, reject(std::move(reject)) , reject(std::move(reject))
, connection(std::move(connection)) , connection(connection)
, addr(addr) , addr(addr)
, addr_len(addr_len) , addr_len(addr_len)
{ } { }
Async::Resolver resolve; Async::Resolver resolve;
Async::Rejection reject; Async::Rejection reject;
std::shared_ptr<Connection> connection; std::weak_ptr<Connection> connection;
const struct sockaddr* addr; const struct sockaddr* addr;
socklen_t addr_len; socklen_t addr_len;
}; };
...@@ -261,7 +261,7 @@ private: ...@@ -261,7 +261,7 @@ private:
void handleRequestsQueue(); void handleRequestsQueue();
void handleConnectionQueue(); void handleConnectionQueue();
void handleIncoming(const 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);
......
...@@ -141,9 +141,16 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -141,9 +141,16 @@ Transport::onReady(const Aio::FdSet& fds) {
else if (entry.isReadable()) { else if (entry.isReadable()) {
auto tag = entry.getTag(); auto tag = entry.getTag();
auto fd = tag.value(); auto fd = tag.value();
auto reqIt = connections.find(fd); auto connIt = connections.find(fd);
if (reqIt != std::end(connections)) if (connIt != std::end(connections)) {
handleIncoming(reqIt->second.connection); auto connection = connIt->second.connection.lock();
if (connection) {
handleIncoming(connection);
}
else {
throw std::runtime_error("Connection error");
}
}
else { else {
auto timerIt = timeouts.find(fd); auto timerIt = timeouts.find(fd);
if (timerIt != std::end(timeouts)) if (timerIt != std::end(timeouts))
...@@ -164,7 +171,12 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -164,7 +171,12 @@ Transport::onReady(const Aio::FdSet& fds) {
else { else {
conn.resolve(); conn.resolve();
// We are connected, we can start reading data now // We are connected, we can start reading data now
reactor()->modifyFd(key(), conn.connection->fd, NotifyOn::Read); auto connection = connIt->second.connection.lock();
if (connection) {
reactor()->modifyFd(key(), connection->fd, NotifyOn::Read);
} else {
throw std::runtime_error("Connection error");
}
} }
} else { } else {
throw std::runtime_error("Unknown fd"); throw std::runtime_error("Unknown fd");
...@@ -180,7 +192,7 @@ Transport::registerPoller(Polling::Epoll& poller) { ...@@ -180,7 +192,7 @@ Transport::registerPoller(Polling::Epoll& poller) {
} }
Async::Promise<void> Async::Promise<void>
Transport::asyncConnect(const std::shared_ptr<Connection>& connection, const struct sockaddr* address, socklen_t addr_len) Transport::asyncConnect(std::shared_ptr<Connection> connection, const struct sockaddr* address, socklen_t addr_len)
{ {
return Async::Promise<void>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<void>([=](Async::Resolver& resolve, Async::Rejection& reject) {
ConnectionEntry entry(std::move(resolve), std::move(reject), connection, address, addr_len); ConnectionEntry entry(std::move(resolve), std::move(reject), connection, address, addr_len);
...@@ -266,9 +278,11 @@ Transport::handleConnectionQueue() { ...@@ -266,9 +278,11 @@ Transport::handleConnectionQueue() {
auto entry = connectionsQueue.popSafe(); auto entry = connectionsQueue.popSafe();
if (!entry) break; if (!entry) break;
auto &data = entry->data(); auto &data = entry->data();
const auto& conn = data.connection; auto conn = data.connection.lock();
if (!conn) {
throw std::runtime_error("Connection error");
}
int res = ::connect(conn->fd, data.addr, data.addr_len); int res = ::connect(conn->fd, data.addr, data.addr_len);
if (res == -1) { if (res == -1) {
if (errno == EINPROGRESS) { if (errno == EINPROGRESS) {
...@@ -284,7 +298,7 @@ Transport::handleConnectionQueue() { ...@@ -284,7 +298,7 @@ Transport::handleConnectionQueue() {
} }
void void
Transport::handleIncoming(const std::shared_ptr<Connection>& connection) { Transport::handleIncoming(std::shared_ptr<Connection> connection) {
char buffer[Const::MaxBuffer] = {0}; char buffer[Const::MaxBuffer] = {0};
ssize_t totalBytes = 0; ssize_t totalBytes = 0;
......
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