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

More work on new streams

Browse Source
This commit is contained in:
Alex Hultman
2018-09-27 23:44:46 +02:00
parent ee81e9b0b0
commit 2c511ec904
7 changed files with 266 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files
15.pro
+4 -3
View File
@@ -28,8 +28,9 @@ HEADERS += \
src/AsyncSocket.h \
src/AsyncSocketData.h \
src/Loop.h \
src/App.h
src/App.h \
src/Utilities.h
INCLUDEPATH += uSockets/src src
QMAKE_CXXFLAGS += -fsanitize=address
LIBS += -lasan -lssl -lcrypto
#QMAKE_CXXFLAGS += -fsanitize=address
LIBS += -pthread -lssl -lcrypto
main.cpp
+64 -24
View File
@@ -2,45 +2,85 @@
#include "examples/helpers/AsyncFileReader.h"
AsyncFileReader asyncFileReader("/home/alexhultman/v0.15/sintel.mkv");
AsyncFileReader asyncFileReader("/home/alexhultman/sintel_small.mp4");
inline std::string slurp(const std::string &path) {
std::ostringstream buf;
std::ifstream input (path.c_str());
buf << input.rdbuf();
return buf.str();
}
std::string sintelMovie = slurp("/home/alexhultman/sintel_small.mp4");
int main(int argc, char **argv) {
std::cout << "Sintel movie is " << sintelMovie.length() << " bytes" << std::endl;
// läs in hela sintel-filmen här, testa strömmarna med den sen!
uWS::/*SSL*/App(/*{
.key_file_name = "/home/alexhultman/uWebSockets/misc/ssl/key.pem",
.cert_file_name = "/home/alexhultman/uWebSockets/misc/ssl/cert.pem",
.dh_params_file_name = "/home/alexhultman/dhparams.pem",
.passphrase = "1234"
}*/).get("/", [](auto *res, auto *req) {
}*/).get("/*", [](auto *res, auto *req) {
res->writeStatus(uWS::HTTP_200_OK);
res->writeHeader("Content-Type", "text/html;charset=utf-8");
//res->writeStatus(uWS::HTTP_200_OK);
//res->writeHeader("Content-Type", "text/html;charset=utf-8");
res->write("<h1>Hallå!</h1>Din user-agent är: ");
res->end(req->getHeader("user-agent"));
std::cout << "Endar nu" << std::endl;
// buffer it up and end by draining it
res->end(sintelMovie);
// stream it here
/*if (!res->tryEnd(sintelMovie)) {
res->onWritable([res](int offset) {
std::cout << "Streaming data at offset " << offset << std::endl;
res->tryEnd(std::string_view(sintelMovie).substr(offset));
});
}*/
// tryWrite / tryEnd
// vad om man skriver tryEnd("<h1>Hallå!</h1>Din user-agent är: ", totalLength)
//res->write("<h1>Hallå!</h1>Din user-agent är: ");
//res->end(req->getHeader("user-agent"));
}).get("/async/sintel.mkv", [](auto *res, auto *req) {
// asyncFileReader.getFileSize()
/* Peek from cache */
std::string_view chunk = asyncFileReader.peek(/*offset*/ 0);
if (chunk.length()) {
/* We had parts of this file cached already */
res->write(chunk);
} else {
/* We had nothing readily available right now, request async chunk and pause the stream until we have */
asyncFileReader.request(/*offset*/ 0, [res](std::string_view chunk) {
/* We were aborted */
if (!chunk.length()) {
std::cout << "Async File Read request was aborted!" << std::endl;
// close the socket here?
// we need a way to NOT resume a paused socket! essentially close!
// I guess it should return wherer or not it wants to be called again?
auto streamIt = [res](int offset) {
while(true) {
/* Peek from cache */
std::string_view chunk = asyncFileReader.peek(offset);
if (chunk.length()) {
/* We had parts of this file cached already */
res->tryEnd(chunk, asyncFileReader.getFileSize());
} else {
/* We finally got the data, resume stream with this chunk */
res->write(chunk);
/* We had nothing readily available right now, request async chunk and pause the stream until we have */
asyncFileReader.request(offset, [res](std::string_view chunk) {
/* We were aborted */
if (!chunk.length()) {
std::cout << "Async File Read request was aborted!" << std::endl;
// close the socket here?
// we need a way to NOT resume a paused socket! essentially close!
} else {
/* We finally got the data, resume stream with this chunk */
res->tryEnd(chunk, asyncFileReader.getFileSize());
}
});
}
});
return true;
}
};
// basically do this
if (!streamIt(0)) {
// what should it return? will onWritable be called again depending on what it returns?
res->onWritable(streamIt);
}
}).listen(3000, [](auto *token) {
src/AsyncSocket.h
+44 -47
View File
@@ -11,6 +11,7 @@ namespace uWS {
template <bool SSL>
struct AsyncSocket : StaticDispatch<SSL> {
template <bool> friend struct HttpContext;
protected:
using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE;
using StaticDispatch<SSL>::static_dispatch;
@@ -23,39 +24,68 @@ protected:
}
}
public:
void *getExt() {
return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
}
void timeout(unsigned int seconds) {
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds);
}
void shutdown() {
static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this);
}
SOCKET_TYPE *close() {
return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
}
/* Cork this socket. Only one socket may ever be corked per-loop at any given time */
void cork() {
std::cout << "Cork called" << std::endl;
LoopData *loopData = getLoopData();
loopData->corked = true;
}
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer */
/* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. */
// todo: consider supporting nextLength = UNKNOWN as -1 (more but unknown size)
int write(const char *src, int length, bool optionally = false, int nextLength = 0) {
LoopData *loopData = getLoopData();
//std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl;
std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl;
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
/* Do nothing for a null sized chunk */
if (length == 0) {
if (length == 0 && !asyncSocketData->buffer.length()) {
//std::cout << "Write returned: 0" << std::endl;
return 0;
}
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
/* Do not write anything if we have a per-socket buffer */
if (asyncSocketData->buffer.length()) {
if (optionally) {
//std::cout << "Write returned: 0" << std::endl;
// we have buffer and we are optionally, if drain then drain else quit
// drain here
std::cout << "Drain path" << std::endl;
// will just end up in a loop!
int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), nextLength != 0);//write(asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), optionally, 0, true);
// removeBuffer
asyncSocketData->buffer = asyncSocketData->buffer.substr(written);
// should we really return this here? should be 0 as we took 0 new data!
return 0;
} else {
std::cout << "Buffering at top of write!" << std::endl;
std::cout << "Buffering at top of write (really bad)!" << std::endl;
/* At least we can reserve room for next chunk if we know it up front */
if (nextLength) {
@@ -104,7 +134,7 @@ public:
return written;
}
std::cout << "Buffering at bottom of write!" << std::endl;
std::cout << "Buffering at bottom of write (okay)!" << std::endl;
/* Fall back to worst possible case (should be very rare for HTTP) */
/* At least we can reserve room for next chunk if we know it up front */
@@ -124,6 +154,9 @@ public:
/* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */
/* It does NOT count bytes written from cork buffer (they are already accounted for in the write call responsible for its corking)! */
int uncork(const char *src = nullptr, int length = 0, bool optionally = false) {
std::cout << "Uncork called with length: " << length << std::endl;
LoopData *loopData = getLoopData();
if (loopData->corked) {
@@ -137,48 +170,12 @@ public:
/* We should only return with new writes, not things written to cork already */
return write(src, length, optionally, 0);
} else {
std::cout << "Not even corked!" << std::endl;
}
return 0;
}
/* Check if this socket has buffered data */
bool hasBuffer() {
return ((AsyncSocketData<SSL> *) getExt())->buffer.length() > 0;
}
/* Drain any socket-buffer while also optionally sending a chunk */
int mergeDrain(std::string_view optionalChunk = {}) {
// strategy: if we have two parts and both will fit in cork buffer then cork them and recursively send them off
// write any per-socket buffer and optionally more
AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
// not handled yet
if (asyncSocketData->buffer.length()) {
std::cout << "ERROR! has socket buffer!" << std::endl;
exit(0);
}
/* Write the optional part */
return write(optionalChunk.data(), optionalChunk.length(), true, 0);
}
/* These should not be public to the user! */
void timeout(unsigned int seconds) {
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds);
}
void shutdown() {
static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this);
}
SOCKET_TYPE *close() {
return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
}
};
}
src/HttpContext.h
+23 -45
View File
@@ -135,60 +135,38 @@ private:
/* Handle HTTP write out */
static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) {
//std::cout << "Writable event!" << std::endl;
std::cout << "Writable event!" << std::endl;
/* Silence any spurious writable events due to SSL_read failing to write */
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT) {
return s;
}
/* Writing data should reset the timeout */
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
/* Are we already ended and just waiting for a drain / shutdown? */
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_ENDED_STREAM_OUT) {
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
/* Try and send everything buffered up */
asyncSocket->mergeDrain();
/* If we succeed with drainage we can finally shut down */
if (!asyncSocket->hasBuffer()) {
asyncSocket->shutdown();
}
/* Nothing here for us */
return s;
}
if (httpResponseData->outStream) {
/* Regular path, request more data */
// 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!
if (written < chunk.length()) {
break;
}
}
// todo: we should loop until we cannot send anymore just like we do in HttpResponse::write(stream)!
// if this, then it means it finished with no issues so we need to empty any buffers?
if (httpResponseData->onWritable) {
httpResponseData->onWritable(httpResponseData->offset);
} else {
/* We can come here if we only have socket buffers to drain yet no attached stream */
asyncSocket->mergeDrain();
// lets drain here
std::cout << "LEts drain!" << std::endl;
// mergeDrain
asyncSocket->write(nullptr, 0, true, 0);
}
// bascially just empty the buffer and if successful also call onWritable (bad strategy!)
//asyncSocket->mergeDrain();
// what we want is to immediately call onWritable and have AsyncSocket::write calls always try and empty any buffers at the same time?
// AsyncSocket::write can take boolean drain = true to know it should try and drain the buffers according to whatever strategy
// mergeDrain is basically AsyncSocket::write with boolean drain = true!
// on writable should return whether it wants more data or not?
return s;
});
src/HttpResponse.h
+79 -127
View File
@@ -5,6 +5,7 @@
#include "AsyncSocket.h"
#include "HttpResponseData.h"
#include "Utilities.h"
namespace uWS {
@@ -13,62 +14,28 @@ const char *HTTP_200_OK = "200 OK";
template <bool SSL>
struct HttpResponse : public AsyncSocket<SSL> {
typedef AsyncSocket<SSL> Super;
private:
HttpResponseData<SSL> *getHttpResponseData() {
return (HttpResponseData<SSL> *) AsyncSocket<SSL>::getExt();
}
int u32toaHex(uint32_t value, char *dst) {
char palette[] = "0123456789abcdef";
char temp[10];
char *p = temp;
do {
*p++ = palette[value % 16];
value /= 16;
} while (value > 0);
int ret = p - temp;
do {
*dst++ = *--p;
} while (p != temp);
return ret;
return (HttpResponseData<SSL> *) Super::getExt();
}
/* Write an unsigned 32-bit integer in hex */
void writeUnsignedHex(unsigned int value) {
char buf[10];
int length = u32toaHex(value, buf);
int length = utils::u32toaHex(value, buf);
/* For now we do this copy */
AsyncSocket<SSL>::write(buf, length);
}
int u32toa(uint32_t value, char *dst) {
char temp[10];
char *p = temp;
do {
*p++ = (char) (value % 10) + '0';
value /= 10;
} while (value > 0);
int ret = p - temp;
do {
*dst++ = *--p;
} while (p != temp);
return ret;
Super::write(buf, length);
}
/* Write an unsigned 32-bit integer */
void writeUnsigned(unsigned int value) {
char buf[10];
int length = u32toa(value, buf);
int length = utils::u32toa(value, buf);
/* For now we do this copy */
AsyncSocket<SSL>::write(buf, length);
Super::write(buf, length);
}
public:
@@ -77,59 +44,61 @@ public:
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
/* Do not allow writing more than one status */
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_STATUS_SENT) {
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_STATUS_CALLED) {
return this;
}
/* Update status */
httpResponseData->state |= HttpResponseData<SSL>::HTTP_STATUS_SENT;
httpResponseData->state |= HttpResponseData<SSL>::HTTP_STATUS_CALLED;
AsyncSocket<SSL>::write("HTTP/1.1 ", 9);
AsyncSocket<SSL>::write(status.data(), status.length());
AsyncSocket<SSL>::write("\r\n", 2);
Super::write("HTTP/1.1 ", 9);
Super::write(status.data(), status.length());
Super::write("\r\n", 2);
return this;
}
/* Write an HTTP header with string value */
HttpResponse *writeHeader(std::string_view key, std::string_view value) {
AsyncSocket<SSL>::write(key.data(), key.length());
AsyncSocket<SSL>::write(": ", 2);
AsyncSocket<SSL>::write(value.data(), value.length());
AsyncSocket<SSL>::write("\r\n", 2);
Super::write(key.data(), key.length());
Super::write(": ", 2);
Super::write(value.data(), value.length());
Super::write("\r\n", 2);
return this;
}
/* Write an HTTP header with unsigned int value */
HttpResponse *writeHeader(std::string_view key, unsigned int value) {
AsyncSocket<SSL>::write(key.data(), key.length());
AsyncSocket<SSL>::write(": ", 2);
Super::write(key.data(), key.length());
Super::write(": ", 2);
writeUnsigned(value);
AsyncSocket<SSL>::write("\r\n", 2);
Super::write("\r\n", 2);
return this;
}
/* End the response with an optional data chunk */
void end(std::string_view data = {}) {
writeStatus(HTTP_200_OK);
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_WRITE_CALLED) {
/* Do not allow sending 0 chunk here */
if (data.length()) {
AsyncSocket<SSL>::write("\r\n", 2);
Super::write("\r\n", 2);
writeUnsignedHex(data.length());
AsyncSocket<SSL>::write("\r\n", 2);
AsyncSocket<SSL>::write(data.data(), data.length());
Super::write("\r\n", 2);
Super::write(data.data(), data.length());
}
/* Terminating 0 chunk */
AsyncSocket<SSL>::write("\r\n0\r\n\r\n", 7);
Super::write("\r\n0\r\n\r\n", 7);
} else {
/* We have a known send size */
AsyncSocket<SSL>::write("Content-Length: ", 16);
Super::write("Content-Length: ", 16);
writeUnsigned(data.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
Super::write("\r\n\r\n", 4);
AsyncSocket<SSL>::write(data.data(), data.length());
Super::write(data.data(), data.length());
}
}
@@ -147,94 +116,77 @@ public:
httpResponseData->state |= HttpResponseData<SSL>::HTTP_WRITE_CALLED;
}
AsyncSocket<SSL>::write("\r\n", 2);
Super::write("\r\n", 2);
writeUnsignedHex(data.length());
AsyncSocket<SSL>::write("\r\n", 2);
AsyncSocket<SSL>::write(data.data(), data.length());
Super::write("\r\n", 2);
Super::write(data.data(), data.length());
// are we corked still?
return true;
}
// we really want tryEnd(data) integer to try and stream something with known size
// write/tryWrite called first should enter into chunked?
// tryWrite(char *, length) int
// write(char *, length) bool
/* Attach an output stream function. Chunks may be read more than once. Negative offset mean broken stream */
void writeOldRemoveMe(std::function<std::pair<bool, std::string_view>(int)> cb, int length = 0) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
/* Do not allow write if already called */
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_WRITE_CALLED) {
return;
}
/* Write 200 OK if not already written any status */
// todo: share this code in a function
bool tryEnd(std::string_view data, int totalSize = 0) {
/* Write status if not already done */
writeStatus(HTTP_200_OK);
/* Update status */
httpResponseData->state |= HttpResponseData<SSL>::HTTP_WRITE_CALLED;
if (length) {
httpResponseData->state |= HttpResponseData<SSL>::HTTP_KNOWN_STREAM_OUT_SIZE;
/* Rely on FIN to signal end if we do not pass any length */
AsyncSocket<SSL>::write("Content-Length: ", 16);
writeUnsigned(length);
AsyncSocket<SSL>::write("\r\n", 2);
/* If no total size given then assume this chunk is everything */
if (!totalSize) {
totalSize = data.length();
}
/* HTTP body separator */
AsyncSocket<SSL>::write("\r\n", 2);
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
for (int offset = 0; length == 0 || offset < length; ) {
/* Pull a chunk from stream */
auto [msg_more, chunk] = cb(offset);
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_WRITE_CALLED) {
/* Do not allow sending 0 chunk here */
if (data.length()) {
Super::write("\r\n", 2);
writeUnsignedHex(data.length());
Super::write("\r\n", 2);
/* Handle PAUSE and FIN */
if (chunk.length() == 0) {
/* FIN */
if (chunk.data()) {
/* Try flush and shut down */
AsyncSocket<SSL>::uncork();
if (!AsyncSocket<SSL>::hasBuffer()) {
/* Uncork finished with no buffered data */
AsyncSocket<SSL>::shutdown();
} else {
/* Let it shut down when drained */
httpResponseData->state |= HttpResponseData<SSL>::HTTP_ENDED_STREAM_OUT;
}
} else {
/* 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;
// should be optional
Super::write(data.data(), data.length());
}
/* Send off the chunk */
int written = AsyncSocket<SSL>::write(chunk.data(), chunk.length(), true);
/* Terminating 0 chunk */
Super::write("\r\n0\r\n\r\n", 7);
} else {
// if not already ended! should we call tryWrite after this? we need an extra flag! end called!
/* If we failed to send everything, exit */
if (written < chunk.length()) {
httpResponseData->offset = offset + written;
httpResponseData->outStream = cb;
return;
if (!(httpResponseData->state & HttpResponseData<SSL>::HTTP_END_CALLED)) {
/* We have a known send size */
Super::write("Content-Length: ", 16);
writeUnsigned(data.length());
Super::write("\r\n\r\n", 4);
/* Mark end called */
httpResponseData->state |= HttpResponseData<SSL>::HTTP_END_CALLED;
}
offset += written;
/* Write as much as possible without causing backpressure */
httpResponseData->offset += Super::write(data.data(), data.length(), true);
}
return httpResponseData->offset == totalSize;
}
/* Attach handler for writable HTTP response */
HttpResponse *onWritable(std::function<void(int)> handler) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
httpResponseData->onWritable = handler;
return this;
}
/* Attach handler for aborted HTTP request */
HttpResponse *onAborted(std::function<void()> handler) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
httpResponseData->onAborted = handler;
return this;
}
// onData(chunk, remaining == -1 or 0 or actual remaining)?
/* Attach a read handler for data sent. Will be called with a chunk of size 0 when FIN */
void read(std::function<void(std::string_view)> handler) {
HttpResponseData<SSL> *data = getHttpResponseData();
src/HttpResponseData.h
+3 -3
View File
@@ -16,15 +16,15 @@ struct HttpResponseData : HttpParser, AsyncSocketData<SSL> {
private:
/* Bits of status */
enum {
HTTP_STATUS_SENT = 1, // used
HTTP_STATUS_CALLED = 1, // used
HTTP_WRITE_CALLED = 2, // used
HTTP_KNOWN_STREAM_OUT_SIZE = 4, // not used
HTTP_END_CALLED = 4, // used
HTTP_PAUSED_STREAM_OUT = 8, // not used
HTTP_ENDED_STREAM_OUT = 16 // not used
};
/* Per socket event handlers */
std::function<void()> onWritable;
std::function<void(int)> onWritable;
//std::function<void()> onData;
std::function<void(std::string_view)> inStream;
src/Utilities.h
+49
View File
@@ -0,0 +1,49 @@
#ifndef UTILITIES_H
#define UTILITIES_H
/* Various common utilities */
#include <cstdint>
namespace uWS {
namespace utils {
int u32toaHex(uint32_t value, char *dst) {
char palette[] = "0123456789abcdef";
char temp[10];
char *p = temp;
do {
*p++ = palette[value % 16];
value /= 16;
} while (value > 0);
int ret = p - temp;
do {
*dst++ = *--p;
} while (p != temp);
return ret;
}
int u32toa(uint32_t value, char *dst) {
char temp[10];
char *p = temp;
do {
*p++ = (char) (value % 10) + '0';
value /= 10;
} while (value > 0);
int ret = p - temp;
do {
*dst++ = *--p;
} while (p != temp);
return ret;
}
}
}
#endif // UTILITIES_H