talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Loop::defer and almost working async streaming

Browse Source
This commit is contained in:
Alex Hultman
2018-09-23 21:48:29 +02:00
parent 2fb87b67c7
commit 8106a0f628
6 changed files with 123 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/helpers/AsyncFileReader.h
+37 -9
View File
@@ -3,6 +3,7 @@
#include <fstream>
#include <sstream>
#include <iostream>
#include <future>
/* 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<void(std::string_view)> 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<void(std::string_view)> 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 */
main.cpp
+7 -6
View File
@@ -43,10 +43,12 @@ int main(int argc, char **argv) {
return std::pair<bool, std::string_view>(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<bool, std::string_view>(false, outerChunk);
//std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl;
//return std::pair<bool, std::string_view>(false, outerChunk);
// what if we resumed before we paused! we cannot do that!
src/HttpContext.h
+7
View File
@@ -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<SSL>::HTTP_PAUSED_STREAM_OUT;
break;
}
int written = asyncSocket->mergeDrain(chunk);
httpResponseData->offset += written;
// this is not correct, we can reach the end!
src/HttpResponse.h
+28 -1
View File
@@ -96,6 +96,7 @@ public:
/* Do nothing if not even paused */
if (!(httpResponseData->state & HttpResponseData<SSL>::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<SSL>::HTTP_PAUSED_STREAM_OUT;
if (chunk.length()) {
/*if (chunk.length()) {
int written = AsyncSocket<SSL>::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<SSL> *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<SSL>::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<SSL>::timeout(0);
httpResponseData->state |= HttpResponseData<SSL>::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;
}
src/Loop.h
+31 -1
View File
@@ -7,13 +7,29 @@
#include <libusockets.h>
#include <thread>
#include <iostream>
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<void()> 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);
src/LoopData.h
+13
View File
@@ -1,8 +1,19 @@
#ifndef LOOPDATA_H
#define LOOPDATA_H
#include <thread>
#include <functional>
#include <vector>
#include <mutex>
namespace uWS {
struct LoopData {
friend struct Loop;
private:
std::mutex deferMutex;
int currentDeferQueue = 0;
std::vector<std::function<void()>> deferQueues[2];
public:
/* Good 16k for SSL perf. */
@@ -14,4 +25,6 @@ public:
bool corked = false;
};
}
#endif // LOOPDATA_H