Commit f78f293a authored by Ian Roddis's avatar Ian Roddis

Implementing deque for write queue to allow for multiple pending writes and to preserve write order

parent 7bb54435
...@@ -12,6 +12,8 @@ ...@@ -12,6 +12,8 @@
#include <pistache/async.h> #include <pistache/async.h>
#include <pistache/stream.h> #include <pistache/stream.h>
#include <deque>
namespace Pistache { namespace Pistache {
namespace Tcp { namespace Tcp {
...@@ -199,7 +201,7 @@ private: ...@@ -199,7 +201,7 @@ private:
fd becomes ready again, we can write everything fd becomes ready again, we can write everything
*/ */
PollableQueue<WriteEntry> writesQueue; PollableQueue<WriteEntry> writesQueue;
std::unordered_map<Fd, WriteEntry> toWrite; std::unordered_map<Fd, std::deque<WriteEntry> > toWrite;
PollableQueue<TimerEntry> timersQueue; PollableQueue<TimerEntry> timersQueue;
std::unordered_map<Fd, TimerEntry> timers; std::unordered_map<Fd, TimerEntry> timers;
...@@ -227,11 +229,8 @@ private: ...@@ -227,11 +229,8 @@ private:
void armTimerMsImpl(TimerEntry entry); void armTimerMsImpl(TimerEntry entry);
void asyncWriteImpl(Fd fd, WriteEntry& entry, WriteStatus status = FirstTry); // This will attempt to drain the write queue for the fd
void asyncWriteImpl( void asyncWriteImpl(Fd fd);
Fd fd, int flags, const BufferHolder& buffer,
Async::Deferred<ssize_t> deferred,
WriteStatus status = FirstTry);
void handlePeerDisconnection(const std::shared_ptr<Peer>& peer); void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);
void handleIncoming(const std::shared_ptr<Peer>& peer); void handleIncoming(const std::shared_ptr<Peer>& peer);
......
...@@ -99,8 +99,8 @@ Transport::onReady(const Aio::FdSet& fds) { ...@@ -99,8 +99,8 @@ Transport::onReady(const Aio::FdSet& fds) {
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
auto& write = it->second; // Try to drain the queue
asyncWriteImpl(fd, write, Retry); asyncWriteImpl(fd);
} }
} }
} }
...@@ -164,31 +164,45 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) { ...@@ -164,31 +164,45 @@ Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
throw std::runtime_error("Could not find peer to erase"); throw std::runtime_error("Could not find peer to erase");
peers.erase(it); peers.erase(it);
// Clean up buffers
auto & wq = toWrite[fd];
while (wq.size() > 0) {
auto & entry = wq.front();
const BufferHolder & buffer = entry.buffer;
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
}
toWrite.erase(fd); toWrite.erase(fd);
close(fd); close(fd);
} }
void void
Transport::asyncWriteImpl(Fd fd, Transport::WriteEntry& entry, WriteStatus status) { Transport::asyncWriteImpl(Fd fd)
asyncWriteImpl(fd, entry.flags, entry.buffer, std::move(entry.deferred), status);
}
void
Transport::asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Deferred<ssize_t> deferred, WriteStatus status)
{ {
auto & wq = toWrite[fd];
while (wq.size() > 0) {
auto & entry = wq.front();
int flags = entry.flags;
const BufferHolder &buffer = entry.buffer;
Async::Deferred<ssize_t> deferred = std::move(entry.deferred);
auto cleanUp = [&]() { auto cleanUp = [&]() {
if (buffer.isRaw()) { if (buffer.isRaw()) {
auto raw = buffer.raw(); auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data; if (raw.isOwned) delete[] raw.data;
} }
wq.pop_front();
if (status == Retry) if (wq.size() == 0) {
toWrite.erase(fd); toWrite.erase(fd);
reactor()->modifyFd(key(), fd, NotifyOn::Read, Polling::Mode::Edge);
}
}; };
bool halt = false;
size_t totalWritten = buffer.offset(); size_t totalWritten = buffer.offset();
for (;;) { for (;;) {
ssize_t bytesWritten = 0; ssize_t bytesWritten = 0;
...@@ -204,20 +218,14 @@ Transport::asyncWriteImpl( ...@@ -204,20 +218,14 @@ Transport::asyncWriteImpl(
} }
if (bytesWritten < 0) { if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
// save for a future retry with the totalWritten offset. wq.pop_front();
if (status == Retry) { wq.push_front(WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags));
toWrite.erase(fd);
}
toWrite.insert(
std::make_pair(fd,
WriteEntry(std::move(deferred), buffer.detach(totalWritten), flags)));
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
} }
else { else {
cleanUp(); cleanUp();
deferred.reject(Pistache::Error::system("Could not write data")); deferred.reject(Pistache::Error::system("Could not write data"));
halt = true;
} }
break; break;
} }
...@@ -239,6 +247,8 @@ Transport::asyncWriteImpl( ...@@ -239,6 +247,8 @@ Transport::asyncWriteImpl(
} }
} }
} }
if (halt) break;
}
} }
void void
...@@ -299,7 +309,9 @@ Transport::handleWriteQueue() { ...@@ -299,7 +309,9 @@ Transport::handleWriteQueue() {
if (!entry) break; if (!entry) break;
auto &write = entry->data(); auto &write = entry->data();
asyncWriteImpl(write.peerFd, write); auto fd = write.peerFd;
toWrite[fd].push_back(std::move(write));
reactor()->modifyFd(key(), fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
} }
} }
......
...@@ -14,7 +14,7 @@ using namespace Pistache; ...@@ -14,7 +14,7 @@ using namespace Pistache;
static const size_t N_LETTERS = 26; static const size_t N_LETTERS = 26;
static const size_t LETTER_REPEATS = 100000; static const size_t LETTER_REPEATS = 100000;
static const size_t SET_REPEATS = 10; static const size_t SET_REPEATS = 10;
static const uint16_t PORT = 9080; static const uint16_t PORT = 9082;
void dumpData(const Rest::Request&req, Http::ResponseWriter response) { void dumpData(const Rest::Request&req, Http::ResponseWriter response) {
UNUSED(req); UNUSED(req);
......
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