Skip to content

Commit cd423d1

Browse files
PagghiuPagghiu
committed
Http: Start integrating AsyncStreams
1 parent c1b763c commit cd423d1

File tree

7 files changed

+186
-77
lines changed

7 files changed

+186
-77
lines changed

‎Documentation/Libraries/Http.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,8 @@
77
[SaneCppHttp.h](https://github.com/Pagghiu/SaneCppLibraries/releases/latest/download/SaneCppHttp.h) is a library implementing a hand-written http 1.1 parser, client and server.
88

99
# Dependencies
10-
- Dependencies: [Async](@ref library_async), [Memory](@ref library_memory)
11-
- All dependencies: [Async](@ref library_async), [File](@ref library_file), [FileSystem](@ref library_file_system), [Foundation](@ref library_foundation), [Memory](@ref library_memory), [Socket](@ref library_socket), [Threading](@ref library_threading)
10+
- Dependencies: [AsyncStreams](@ref library_async_streams), [Memory](@ref library_memory)
11+
- All dependencies: [Async](@ref library_async), [AsyncStreams](@ref library_async_streams), [File](@ref library_file), [FileSystem](@ref library_file_system), [Foundation](@ref library_foundation), [Memory](@ref library_memory), [Socket](@ref library_socket), [Threading](@ref library_threading)
1212

1313
![Dependency Graph](Http.svg)
1414

‎Libraries/Http/HttpAsyncServer.cpp‎

Lines changed: 93 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,19 @@ Result HttpAsyncServer::start(AsyncEventLoop& loop, StringSpan address, uint16_t
1919
asyncServerAccept.callback.bind<HttpAsyncServer, &HttpAsyncServer::onNewClient>(*this);
2020
SC_TRY(asyncServerAccept.start(*eventLoop, serverSocket));
2121
started = true;
22-
2322
return httpServer.start(memory);
2423
}
2524

25+
void HttpAsyncServer::setUseAsyncStreams(bool useAsyncStreams, Span<AsyncReadableStream::Request> readQueue,
26+
Span<AsyncWritableStream::Request> writeQueue, Span<AsyncBufferView> buffers)
27+
{
28+
readQueues = readQueue;
29+
writeQueues = writeQueue;
30+
useStreams = useAsyncStreams;
31+
32+
buffersPool.buffers = buffers;
33+
}
34+
2635
Result HttpAsyncServer::stopAsync()
2736
{
2837
stopping = true;
@@ -34,10 +43,7 @@ Result HttpAsyncServer::stopAsync()
3443

3544
for (HttpServerClient& it : httpServer.clients)
3645
{
37-
if (it.state != HttpServerClient::State::Free)
38-
{
39-
closeAsync(it);
40-
}
46+
closeAsync(it);
4147
}
4248
return Result(true);
4349
}
@@ -72,19 +78,64 @@ void HttpAsyncServer::onNewClient(AsyncSocketAccept::Result& result)
7278
SC_ASSERT_RELEASE(httpServer.allocateClient(idx));
7379

7480
HttpServerClient& client = httpServer.clients[idx];
81+
client.socket = move(acceptedClient);
82+
if (useStreams)
83+
{
84+
const size_t readQueueLen = readQueues.sizeInElements() / httpServer.clients.sizeInElements();
85+
const size_t writeQueueLen = writeQueues.sizeInElements() / httpServer.clients.sizeInElements();
86+
SC_TRUST_RESULT(readQueueLen > 0);
87+
SC_TRUST_RESULT(writeQueueLen > 0);
88+
Span<AsyncReadableStream::Request> readQueue;
89+
Span<AsyncWritableStream::Request> writeQueue;
90+
SC_TRUST_RESULT(readQueues.sliceStartLength(idx * readQueueLen, readQueueLen, readQueue));
91+
SC_TRUST_RESULT(writeQueues.sliceStartLength(idx * writeQueueLen, writeQueueLen, writeQueue));
92+
SC_TRUST_RESULT(client.readableSocketStream.init(buffersPool, readQueue, *eventLoop, client.socket));
93+
SC_TRUST_RESULT(client.writableSocketStream.init(buffersPool, writeQueue, *eventLoop, client.socket));
94+
95+
auto onData = [this, idx](AsyncBufferView::ID bufferID) { onStreamReceive(httpServer.clients[idx], bufferID); };
96+
SC_TRUST_RESULT(client.readableSocketStream.eventData.addListener(onData));
97+
SC_TRUST_RESULT(client.readableSocketStream.start());
98+
}
99+
else
100+
{
101+
client.asyncSend.setDebugName(client.debugName);
102+
client.asyncReceive.setDebugName(client.debugName);
103+
client.asyncReceive.callback.bind<HttpAsyncServer, &HttpAsyncServer::onReceive>(*this);
75104

76-
client.socket = move(acceptedClient);
77-
client.asyncSend.setDebugName(client.debugName);
78-
client.asyncReceive.setDebugName(client.debugName);
79-
client.asyncReceive.callback.bind<HttpAsyncServer, &HttpAsyncServer::onReceive>(*this);
80-
81-
// This cannot fail because start reports only incorrect API usage (AsyncRequest already in use etc.)
82-
SC_TRUST_RESULT(client.asyncReceive.start(*eventLoop, client.socket, client.request.availableHeader));
105+
// This cannot fail because start reports only incorrect API usage (AsyncRequest already in use etc.)
106+
SC_TRUST_RESULT(client.asyncReceive.start(*eventLoop, client.socket, client.request.availableHeader));
107+
}
83108

84109
// Only reactivate asyncAccept if arena is not full (otherwise it's being reactivated in closeAsync)
85110
result.reactivateRequest(httpServer.canAcceptMoreClients());
86111
}
87112

113+
void HttpAsyncServer::onStreamReceive(HttpServerClient& client, AsyncBufferView::ID bufferID)
114+
{
115+
Span<char> readData;
116+
buffersPool.getWritableData(bufferID, readData);
117+
// TODO: Handle error for available headers not big enough
118+
SC_ASSERT_RELEASE(readData.sizeInBytes() <= client.request.availableHeader.sizeInBytes());
119+
::memcpy(client.request.availableHeader.data(), readData.data(), readData.sizeInBytes());
120+
121+
if (not client.request.parse(httpServer.maxHeaderSize, readData))
122+
{
123+
// TODO: Invoke on error
124+
return;
125+
}
126+
else if (client.request.allHeadersReceived())
127+
{
128+
httpServer.onRequest(client.request, client.response);
129+
}
130+
131+
if (client.response.mustBeFlushed())
132+
{
133+
auto onAfterWrite = [this, &client](AsyncBufferView::ID) { closeAsync(client); };
134+
SC_TRUST_RESULT(client.writableStream->write(client.response.getSpan(), onAfterWrite));
135+
client.writableStream->end(); // TODO: This must be called only if actually ended...
136+
}
137+
}
138+
88139
void HttpAsyncServer::onReceive(AsyncSocketReceive::Result& result)
89140
{
90141
SC_COMPILER_WARNING_PUSH_OFFSETOF
@@ -98,10 +149,19 @@ void HttpAsyncServer::onReceive(AsyncSocketReceive::Result& result)
98149
return;
99150
}
100151

101-
const size_t idx = static_cast<size_t>(&client - &httpServer.clients[0]);
102-
Span<char> outspan = httpServer.processClientReceivedData(idx, readData);
103-
if (not outspan.empty())
152+
if (not client.request.parse(httpServer.maxHeaderSize, readData))
104153
{
154+
// TODO: Invoke on error
155+
return;
156+
}
157+
else if (client.request.allHeadersReceived())
158+
{
159+
httpServer.onRequest(client.request, client.response);
160+
}
161+
162+
if (client.response.mustBeFlushed())
163+
{
164+
Span<const char> outspan = client.response.getSpan();
105165
client.asyncSend.setDebugName(client.debugName);
106166
client.asyncSend.callback.bind<HttpAsyncServer, &HttpAsyncServer::onAfterSend>(*this);
107167
auto res = client.asyncSend.start(*eventLoop, client.socket, outspan);
@@ -130,15 +190,28 @@ void HttpAsyncServer::onAfterSend(AsyncSocketSend::Result& result)
130190

131191
void HttpAsyncServer::closeAsync(HttpServerClient& requestClient)
132192
{
133-
if (not requestClient.asyncSend.isFree())
193+
if (requestClient.state == HttpServerClient::State::Free)
194+
{
195+
return;
196+
}
197+
198+
if (useStreams)
134199
{
135-
(void)requestClient.asyncSend.stop(*eventLoop);
200+
requestClient.readableSocketStream.destroy(); // emits 'eventClose' cancelling pending reads
201+
requestClient.writableSocketStream.end(); // emits 'eventFinish' cancelling pending writes
136202
}
137-
if (not requestClient.asyncReceive.isFree())
203+
else
138204
{
139-
(void)requestClient.asyncReceive.stop(*eventLoop);
205+
if (not requestClient.asyncSend.isFree())
206+
{
207+
(void)requestClient.asyncSend.stop(*eventLoop);
208+
}
209+
if (not requestClient.asyncReceive.isFree())
210+
{
211+
(void)requestClient.asyncReceive.stop(*eventLoop);
212+
}
213+
SC_TRUST_RESULT(requestClient.socket.close());
140214
}
141-
SC_TRUST_RESULT(requestClient.socket.close());
142215
const bool wasFull = not httpServer.canAcceptMoreClients();
143216

144217
SC_TRUST_RESULT(httpServer.deallocateClient(requestClient));

‎Libraries/Http/HttpAsyncServer.h‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,17 +27,30 @@ struct SC_COMPILER_EXPORT HttpAsyncServer
2727
/// @brief Returns true if the server has been started
2828
[[nodiscard]] bool isStarted() const { return started; }
2929

30+
/// @brief Enables using AsyncStreams instead of raw Async Send and Receive
31+
void setUseAsyncStreams(bool useAsyncStreams, Span<AsyncReadableStream::Request> readQueue,
32+
Span<AsyncWritableStream::Request> writeQueue, Span<AsyncBufferView> buffers);
33+
3034
/// @brief The underlying http server
3135
HttpServer httpServer;
3236

3337
private:
34-
bool started = false;
35-
bool stopping = false;
38+
AsyncBuffersPool buffersPool;
39+
40+
Span<AsyncReadableStream::Request> readQueues;
41+
Span<AsyncWritableStream::Request> writeQueues;
42+
43+
bool useStreams = false;
44+
bool started = false;
45+
bool stopping = false;
3646

3747
void onNewClient(AsyncSocketAccept::Result& result);
48+
void closeAsync(HttpServerClient& requestClient);
49+
3850
void onReceive(AsyncSocketReceive::Result& result);
3951
void onAfterSend(AsyncSocketSend::Result& result);
40-
void closeAsync(HttpServerClient& requestClient);
52+
53+
void onStreamReceive(HttpServerClient& client, AsyncBufferView::ID bufferID);
4154

4255
AsyncEventLoop* eventLoop = nullptr;
4356
SocketDescriptor serverSocket;

‎Libraries/Http/HttpServer.cpp‎

Lines changed: 9 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,15 @@ void HttpRequest::reset()
3333

3434
Result HttpRequest::parse(const uint32_t maxSize, Span<const char> readData)
3535
{
36+
readHeaders = {readHeaders.data(), readHeaders.sizeInBytes() + readData.sizeInBytes()};
37+
const bool hasHeaderSpace = availableHeader.sliceStart(readData.sizeInBytes(), availableHeader);
38+
39+
if (not hasHeaderSpace)
40+
{
41+
parsedSuccessfully = false;
42+
return Result::Error("Header space is finished");
43+
}
44+
3645
if (readHeaders.sizeInBytes() > maxSize)
3746
{
3847
parsedSuccessfully = false;
@@ -136,7 +145,6 @@ Result HttpResponse::end()
136145
//-------------------------------------------------------------------------------------------------------
137146
Result HttpServer::start(Memory& memory)
138147
{
139-
140148
clients = memory.clients;
141149
headersMemory = &memory.headersMemory;
142150
return Result(true);
@@ -177,33 +185,4 @@ bool HttpServer::deallocateClient(HttpServerClient& client)
177185
}
178186
}
179187

180-
Span<char> HttpServer::processClientReceivedData(size_t idx, Span<const char> readData)
181-
{
182-
HttpServerClient& client = clients[idx];
183-
client.request.readHeaders = {client.request.readHeaders.data(),
184-
client.request.readHeaders.sizeInBytes() + readData.sizeInBytes()};
185-
const bool hasHeaderSpace =
186-
client.request.availableHeader.sliceStart(readData.sizeInBytes(), client.request.availableHeader);
187-
188-
if (not client.request.parse(maxHeaderSize, readData))
189-
{
190-
// TODO: Invoke on error
191-
return {};
192-
}
193-
else if (not hasHeaderSpace)
194-
{
195-
// TODO: Invoke on error (no more header space)
196-
return {};
197-
}
198-
if (client.request.headersEndReceived)
199-
{
200-
onRequest(client.request, client.response);
201-
}
202-
if (client.response.mustBeFlushed())
203-
{
204-
return client.response.outputBuffer.toSpan();
205-
}
206-
return {};
207-
}
208-
209188
} // namespace SC

‎Libraries/Http/HttpServer.h‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
#include "HttpParser.h"
55

66
#include "../Async/Async.h"
7+
#include "../AsyncStreams/AsyncRequestStreams.h"
78
#include "../Foundation/Function.h"
89
#include "../Foundation/StringSpan.h"
910
#include "../Memory/Buffer.h"
@@ -51,6 +52,8 @@ struct SC_COMPILER_EXPORT HttpRequest
5152
Span<char> readHeaders;
5253
Span<char> availableHeader; ///< Space to save headers to
5354

55+
bool allHeadersReceived() const { return headersEndReceived; }
56+
5457
private:
5558
friend struct HttpServer;
5659
using HttpHeaderOffset = detail::HttpHeaderOffset; // TODO: hide class implementation
@@ -89,9 +92,11 @@ struct SC_COMPILER_EXPORT HttpResponse
8992
outputBuffer.clear();
9093
}
9194

95+
Span<const char> getSpan() const { return outputBuffer.toSpanConst(); }
96+
[[nodiscard]] bool mustBeFlushed() const { return responseEnded or outputBuffer.size() > highwaterMark; }
97+
9298
private:
9399
friend struct HttpServer;
94-
[[nodiscard]] bool mustBeFlushed() const { return responseEnded or outputBuffer.size() > highwaterMark; }
95100

96101
Buffer outputBuffer;
97102

@@ -121,6 +126,12 @@ struct SC_COMPILER_EXPORT HttpServerClient
121126

122127
char debugName[16] = {0};
123128

129+
ReadableSocketStream readableSocketStream;
130+
WritableSocketStream writableSocketStream;
131+
132+
AsyncReadableStream* readableStream = &readableSocketStream;
133+
AsyncWritableStream* writableStream = &writableSocketStream;
134+
124135
SocketDescriptor socket;
125136
AsyncSocketReceive asyncReceive;
126137
AsyncSocketSend asyncSend;
@@ -164,15 +175,13 @@ struct SC_COMPILER_EXPORT HttpServer
164175
[[nodiscard]] bool allocateClient(size_t& idx);
165176
[[nodiscard]] bool deallocateClient(HttpServerClient& client);
166177

167-
[[nodiscard]] Span<char> processClientReceivedData(size_t idx, Span<const char> readData);
178+
uint32_t maxHeaderSize = 8 * 1024;
168179

169180
private:
170181
void closeAsync(HttpServerClient& requestClient);
171182

172183
IGrowableBuffer* headersMemory = nullptr;
173184
size_t numClients = 0;
174-
175-
uint32_t maxHeaderSize = 8 * 1024;
176185
};
177186
//! @}
178187
#if SC_COMPILER_MSVC

‎Support/Dependencies/Dependencies.json‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -151,17 +151,19 @@
151151
"Http": {
152152
"direct_dependencies": [
153153
"Async",
154+
"AsyncStreams",
154155
"FileSystem",
155156
"Foundation",
156157
"Memory",
157158
"Socket"
158159
],
159160
"minimal_dependencies": [
160-
"Async",
161+
"AsyncStreams",
161162
"Memory"
162163
],
163164
"all_dependencies": [
164165
"Async",
166+
"AsyncStreams",
165167
"File",
166168
"FileSystem",
167169
"Foundation",

0 commit comments

Comments
 (0)