From 8106a0f628891ed0818b7edb80a3a59004f1eebf Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sun, 23 Sep 2018 21:48:29 +0200 Subject: [PATCH] Loop::defer and almost working async streaming --- examples/helpers/AsyncFileReader.h | 46 ++++++++++++++++++++++++------ main.cpp | 13 +++++---- src/HttpContext.h | 7 +++++ src/HttpResponse.h | 29 ++++++++++++++++++- src/Loop.h | 32 ++++++++++++++++++++- src/LoopData.h | 13 +++++++++ 6 files changed, 123 insertions(+), 17 deletions(-) diff --git a/examples/helpers/AsyncFileReader.h b/examples/helpers/AsyncFileReader.h index 9537dcc..7364b86 100644 --- a/examples/helpers/AsyncFileReader.h +++ b/examples/helpers/AsyncFileReader.h @@ -3,6 +3,7 @@ #include #include #include +#include /* This is just a very simple and inefficient demo of async responses, * please do roll your own variant or use a database or Node.js's async @@ -11,7 +12,8 @@ struct AsyncFileReader { private: /* The cache we have in memory for this file */ std::string cache; - int cacheOffset = 0; + int cacheOffset; + bool hasCache; /* The pending async file read (yes we only support one pending read) */ std::function pendingReadCb; @@ -38,6 +40,7 @@ public: fin.seekg(0, fin.beg); fin.read(cache.data(), cache.length()); cacheOffset = 0; + hasCache = true; // get loop for thread @@ -47,7 +50,7 @@ public: /* Returns any data already cached for this offset */ std::string_view peek(int offset) { /* Did we hit the cache? */ - if (offset >= cacheOffset && ((offset - cacheOffset) < cache.length())) { + if (hasCache && offset >= cacheOffset && ((offset - cacheOffset) < cache.length())) { /* Cache hit */ //std::cout << "Cache hit!" << std::endl; return std::string_view(cache.data() + offset - cacheOffset, cache.length() - offset + cacheOffset); @@ -61,14 +64,39 @@ public: /* Asynchronously request more data at offset */ void request(int offset, std::function cb) { - // std::async this - //std::cout << "Caching 1 MB at offset = " << offset << std::endl; - fin.seekg(offset, fin.beg); - fin.read(cache.data(), cache.length()); - cacheOffset = offset; + // in this case, what do we do? + // we need to queue up this chunk request and callback! + // if queue is full, either block or close the connection via abort! + if (!hasCache) { + // already requesting a chunk! + std::cout << "ERROR: already requesting a chunk!" << std::endl; + return; + } - // Loop.defer this - cb(std::string_view(cache.data(), cache.length())); + // disable cache + hasCache = false; + + std::async(std::launch::async, [this, cb, offset]() { + std::cout << "ASYNC Caching 1 MB at offset = " << offset << std::endl; + + + + fin.seekg(offset, fin.beg); + fin.read(cache.data(), cache.length()); + cacheOffset = offset; + + loop->defer([this, cb]() { + + int chunkSize = std::min(cache.length(), fileSize - offset); + + if (chunkSize != cache.length()) { + std::cout << "LESS THAN A CACHE 1 MB!" << std::endl; + } + + hasCache = true; + cb(std::string_view(cache.data(), chunkSize)); + }); + }); } /* Abort any pending async. request */ diff --git a/main.cpp b/main.cpp index a7c666f..8edae8e 100644 --- a/main.cpp +++ b/main.cpp @@ -43,10 +43,12 @@ int main(int argc, char **argv) { return std::pair(false, chunk); } else { - std::string_view outerChunk; + //std::string_view outerChunk; /* We had nothing readily available right now, request async chunk and pause the stream until we have */ - asyncFileReader.request(offset, [&outerChunk, res](std::string_view chunk) { + asyncFileReader.request(offset, [res](std::string_view chunk) { + + std::cout << "We came here!" << std::endl; /* We were aborted */ if (!chunk.length()) { @@ -55,13 +57,12 @@ int main(int argc, char **argv) { // we need a way to NOT resume a paused socket! essentially close! } else { /* We finally got the data, resume stream with this chunk */ - //res->resume(chunk); - outerChunk = chunk; + res->resume(chunk); } }); - std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl; - return std::pair(false, outerChunk); + //std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl; + //return std::pair(false, outerChunk); // what if we resumed before we paused! we cannot do that! diff --git a/src/HttpContext.h b/src/HttpContext.h index c2574f2..bc4a402 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -168,6 +168,13 @@ private: // todo: share this path with HttpResponse::write (it is exatly the same logic!) while (true) { auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); + + if (chunk.length() == 0) { + std::cout << "onwritable paused!" << std::endl; + httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; + break; + } + int written = asyncSocket->mergeDrain(chunk); httpResponseData->offset += written; // this is not correct, we can reach the end! diff --git a/src/HttpResponse.h b/src/HttpResponse.h index a18a1ca..df8154b 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -96,6 +96,7 @@ public: /* Do nothing if not even paused */ if (!(httpResponseData->state & HttpResponseData::HTTP_PAUSED_STREAM_OUT)) { + std::cout << "Resue called but we are not even in paused state!" << std::endl; return; } @@ -104,11 +105,33 @@ public: /* Remove paused status */ httpResponseData->state &= ~HttpResponseData::HTTP_PAUSED_STREAM_OUT; - if (chunk.length()) { + /*if (chunk.length()) { int written = AsyncSocket::write(chunk.data(), chunk.length(), true); if (written == chunk.length()) { // pull a new chunk from the callback (basically call onWritable) + std::cout << "Wrote everything off!" << std::endl; + } + }*/ + + AsyncSocket *asyncSocket = this; + + // again, this path is shared with onwritable, write and here! + while (true) { + auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); + + // break on pause! + if (chunk.length() == 0) { + std::cout << "Resume paused!" << std::endl; + httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; + break; + } + + int written = asyncSocket->mergeDrain(chunk); + httpResponseData->offset += written; + // this is not correct, we can reach the end! + if (written < chunk.length()) { + break; } } @@ -164,6 +187,10 @@ public: /* Disable timeout and mark this stream as paused (important to silence spurious onWritable events) */ AsyncSocket::timeout(0); httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; + + // forgot about this path, if we sent things off and then ended up pausing! + httpResponseData->offset = offset; + httpResponseData->outStream = cb; } return; } diff --git a/src/Loop.h b/src/Loop.h index 70e0f42..41cf964 100644 --- a/src/Loop.h +++ b/src/Loop.h @@ -7,13 +7,29 @@ #include -#include + + + +#include namespace uWS { struct Loop { private: static void wakeupCb(us_loop *loop) { + std::cout << "wakeupCB called" << std::endl; + LoopData *loopData = (LoopData *) us_loop_ext(loop); + /* Swap current deferQueue */ + loopData->deferMutex.lock(); + int oldDeferQueue = loopData->currentDeferQueue; + loopData->currentDeferQueue = (loopData->currentDeferQueue + 1) % 2; + loopData->deferMutex.unlock(); + + /* Drain the queue */ + for (auto &x : loopData->deferQueues[oldDeferQueue]) { + x(); + } + loopData->deferQueues[oldDeferQueue].clear(); } static void preCb(us_loop *loop) { @@ -63,6 +79,20 @@ public: us_loop_free((us_loop *) this); } + /* Defer this callback on Loop's thread of execution */ + void defer(std::function cb) { + LoopData *loopData = (LoopData *) us_loop_ext((us_loop *) this); + + std::cout << "defer called" << std::endl; + //if (std::thread::get_id() == ) // todo: add fast path for same thread id + loopData->deferMutex.lock(); + loopData->deferQueues[loopData->currentDeferQueue].emplace_back(cb); + loopData->deferMutex.unlock(); + + std::cout << "us_wakeup_loop called" << std::endl; + us_wakeup_loop((us_loop *) this); + } + /* Actively block and run this loop */ void run() { us_loop_run((us_loop *) this); diff --git a/src/LoopData.h b/src/LoopData.h index e28804a..df40ed3 100644 --- a/src/LoopData.h +++ b/src/LoopData.h @@ -1,8 +1,19 @@ #ifndef LOOPDATA_H #define LOOPDATA_H +#include +#include +#include +#include + +namespace uWS { + struct LoopData { + friend struct Loop; private: + std::mutex deferMutex; + int currentDeferQueue = 0; + std::vector> deferQueues[2]; public: /* Good 16k for SSL perf. */ @@ -14,4 +25,6 @@ public: bool corked = false; }; +} + #endif // LOOPDATA_H