Async messages improvements

makefile

@@ -63,14 +63,14 @@ FLAGS = $(CPPFLAGS) -Isrc/ $(CPPSTD) -Iexternal/bin/libs/release
 else ifeq ($(CONFIG),debug) # debug
 MAIN_FILE = src/main.cpp
 BIN_NAME = http-debug
-#ADDR_SANITIZER = -fsanitize=address
+ADDR_SANITIZER = -fsanitize=address
 FLAGS = $(CPPFLAGS) -DDEBUG -g -Isrc/ $(ADDR_SANITIZER) $(CPPSTD) -Iexternal/bin/libs/debug 
 
 ### Test settings
 else ifeq ($(CONFIG),test) # test
 MAIN_FILE = testbench/tests.cpp
 BIN_NAME = http-test
-#ADDR_SANITIZER = -fsanitize=address
+ADDR_SANITIZER = -fsanitize=address
 FLAGS = $(CPPFLAGS) -DDEBUG -DTESTING -g -Isrc/ $(ADDR_SANITIZER) $(CPPSTD) -Iexternal/bin/libs/debug 
 LIBRARIES += external/bin/libs/debug/bftest/libbftest-debug.a
 endif # ($(CONFIG),...)

src/log.cpp

@@ -44,6 +44,7 @@ void _LogWriteEntry(BFFileWriter * filewriter, int mode, ...) {
 		logstr
 	);
 
+#ifdef ENABLE_LOG_CONSOLE_PRINT
 	printf(
 		format,
 		dt.month,
@@ -56,6 +57,7 @@ void _LogWriteEntry(BFFileWriter * filewriter, int mode, ...) {
 	);
 	printf("\n");
 	fflush(stdout);
+#endif // ENABLE_LOG_CONSOLE_PRINT
 
 	va_end(arg0);
 	va_end(arg1);

src/log.hpp

@@ -10,6 +10,8 @@ extern "C" {
 #include <bflibc/filewriter.h>
 }
 
+//#define ENABLE_LOG_CONSOLE_PRINT
+
 #define CHAT_LOG_PATH "/tmp/http.log"
 
 extern BFFileWriter gFileWriter;

src/main.cpp

@@ -32,7 +32,7 @@ void help(const char * toolname) {
 }
 
 void __NewConnection(Connection * sc) {
-	LOG_WRITE("new connection made");
+	LOG_DEBUG("new connection made");
 }
 
 int __ReadArguments(int argc, char * argv[]) {

src/office.cpp

@@ -11,6 +11,7 @@
 #include <bflibcpp/bflibcpp.hpp>
 #include <bfnet/bfnet.hpp>
 #include <iostream>
+#include <unistd.h>
 
 using namespace BF::Net;
 using namespace BF;
@@ -23,35 +24,50 @@ using namespace std;
  * once popped from queue
  */
 Atomic<Queue<Envelope *>> _incomingRequests;
-BFThreadAsyncID _tidRequestQueue = NULL;
+
+/**
+ * requests will come in asynchronously from clients, we will queue 
+ * each one and have our worker threads pick off the queue
+ *
+ * we have n worker threads, working concurrently on every available request
+ *
+ * i don't see a performance boost on the client side between 2 and 4 threads
+ */
+const unsigned char numWorkerThreads = 2;
+BFThreadAsyncID _tidRequestQueue[numWorkerThreads];
+
 BFLock _queueSema;
 
 void Office::envelopeReceive(Envelope * envelope) {
-	_incomingRequests.lock();
-
-	BFRetain(envelope);
-	_incomingRequests.unsafeget().push(envelope);
+	int attempts = 0;
+	int threshold = 25;
+	while (_incomingRequests.get<int>([=] (auto & q) {
+		return q.size();
+	}) >= threshold) {
+		usleep(500);
+	}
 
-	_incomingRequests.unlock();
+	_incomingRequests.get([=] (Queue<Envelope *> & q) {
+		BFRetain(envelope);
+		q.push(envelope);
+	});
 	BFLockRelease(&_queueSema);
 }
 
 void __IncomingRequestsWorkerThread(void * in) {
-	while (!BFThreadAsyncIsCanceled(_tidRequestQueue)) {
-		if (_incomingRequests.get().empty()) { 
+	const BFThreadAsyncID tid = BFThreadAsyncGetID();
+	while (!BFThreadAsyncIsCanceled(tid)) {
+		_incomingRequests.lock();
+		if (_incomingRequests.unsafeget().empty()) {
+			_incomingRequests.unlock();
 			BFLockWait(&_queueSema);
 		} else {
-			_incomingRequests.lock();
-
-			// get first item from the queue
 			Envelope * envelope = _incomingRequests.unsafeget().front();
-
-			// pop off
 			_incomingRequests.unsafeget().pop();
-
 			_incomingRequests.unlock();
 
 			if (envelope->data()->size() == 0) {
+				BFRelease(envelope);
 				continue;
 			}
 
@@ -74,15 +90,32 @@ void __IncomingRequestsWorkerThread(void * in) {
 
 void Office::start() {
 	BFLockCreate(&_queueSema);
-	_tidRequestQueue = BFThreadAsync(__IncomingRequestsWorkerThread, NULL);
+
+	for (int i = 0; i < numWorkerThreads; i++) {
+		_tidRequestQueue[i] = BFThreadAsync(__IncomingRequestsWorkerThread, NULL);
+	}
 }
 
 void Office::stop() {
-	BFThreadAsyncCancel(_tidRequestQueue);
-	BFLockRelease(&_queueSema);
-	BFThreadAsyncWait(_tidRequestQueue);
-	BFThreadAsyncDestroy(_tidRequestQueue);
+	for (int i = 0; i < numWorkerThreads; i++) {
+		BFLockRelease(&_queueSema);
+		BFThreadAsyncCancel(_tidRequestQueue[i]);
+	}
+
+	for (int i = 0; i < numWorkerThreads; i++) {
+		BFLockRelease(&_queueSema);
+		BFThreadAsyncWait(_tidRequestQueue[i]);
+		BFThreadAsyncDestroy(_tidRequestQueue[i]);
+	}
 
+	// flush q
+	_incomingRequests.lock();
+	while (!_incomingRequests.unsafeget().empty()) {
+		Envelope * e = _incomingRequests.unsafeget().front();
+		_incomingRequests.unsafeget().pop();
+		BFRelease(e);
+	}
+	_incomingRequests.unlock();
 	BFLockDestroy(&_queueSema);
 }
 

src/request.cpp

@@ -11,38 +11,110 @@ extern "C" {
 #include <bflibc/bflibc.h>
 }
 
-#include <regex>
+#include <sstream>
 
 using namespace BF;
 
 Request::Request(const Data * data) : _message(data == NULL ? 0 : (const char *) data->buffer(), data == NULL ? 0 : data->size()) {
-	LOG_WRITE("Request length = %ld", _message.size());
-	LOG_WRITE("Request content = \n%s", _message.c_str());
+	LOG_DEBUG("Request length = %ld", _message.size());
+	LOG_DEBUG("Request content = \n%s", _message.c_str());
+
+	this->parse();
+}
+
+Request::~Request() { }
+
+void __RequestParseStatusLine(std::string & in, String & method, String & target, String & protocol) {
+	std::stringstream ss(in);
+	std::string buf;
+	char del = ' ';
+
+	if (getline(ss, buf, del)) {
+		method = buf;
+	}
+
+	if (getline(ss, buf, del)) {
+		target = buf;
+	}
+
+	if (getline(ss, buf, del)) {
+		protocol = buf;
+	}
+}
+
+void __RequestParseHeader(std::string & in, HashMap<String, String> & header) {
+	std::string del = ": ";
+	auto pos = in.find(del);
+	std::string key, value;
+
+	if (pos == std::string::npos) {
+		return;
+	}
+
+	key = in.substr(0, pos);
+	in.erase(0, pos + del.length());
+	pos = in.find(del);
+
+	value = in.substr(0, pos);
+
+	header.insert(key, value);
 }
 
-Request::~Request() {
+void Request::parse() {
+	if (this->_message.empty()) {
+		return;
+	}
+
+	std::stringstream ss(this->_message);
+	std::string buf;
+	char del = '\n';
+	int statusLineParsed = false;
+	int headerParsed = false;
+	while (getline(ss, buf, del)) {
+		buf.erase(
+			std::remove_if(
+				buf.begin(),
+				buf.end(),
+				[](unsigned char c) {
+					return c == '\t' || c == '\n' || c == '\r'; 
+				}
+			),
+			buf.end()
+		);
+
+		if (!statusLineParsed) {
+			__RequestParseStatusLine(buf, this->_method, this->_target, this->_protocol);
+			statusLineParsed = true;
+		} else if (!headerParsed) {
+			if (buf.empty()) {
+				headerParsed = true;
+			} else {
+				__RequestParseHeader(buf, this->_header);
+			}
+		} else { // read the body
+			this->_body.append(buf);
+		}
+	}
 }
 
-String Request::method() const {
-	std::regex methodRegex(R"((GET|POST|PUT|DELETE|HEAD|OPTIONS|CONNECT|TRACE|PATCH)\s+)");
-	std::smatch match;
+const String & Request::method() const {
+	return this->_method;
+}
 
-    if (std::regex_search(this->_message, match, methodRegex)) {
-        return match[1].str(); // The second capturing group contains the target
-    } else {
-        return ""; // Or throw an exception if no target is found
-    }
+const String & Request::target() const {
+	return this->_target;
 }
 
-String Request::target() const {
-	std::regex targetRegex(R"((?:GET|POST|PUT|DELETE|HEAD|OPTIONS|CONNECT|TRACE|PATCH)\s+([^\s]+)\s+HTTP/\d\.\d)");
-    std::smatch match;
+const String & Request::protocol() const {
+	return this->_protocol;
+}
 
-    if (std::regex_search(this->_message, match, targetRegex)) {
-        return match[1].str(); // The second capturing group contains the target
-    } else {
-        return ""; // Or throw an exception if no target is found
-    }
+const HashMap<String, String> & Request::header() const {
+	return this->_header;
+}
+
+const String & Request::body() const {
+	return this->_body;
 }
 
 // component: 0=path, 1=query
@@ -121,18 +193,3 @@ HashMap<String, String> Request::targetQuery() const {
 	return res;
 }
 
-String Request::protocol() const {
-    std::regex protocolRegex(R"((?:GET|POST|PUT|DELETE|HEAD|OPTIONS|CONNECT|TRACE|PATCH)\s+[^\s]+\s+(HTTP/\d\.\d))");
-    std::smatch match;
-
-    if (std::regex_search(this->_message, match, protocolRegex)) {
-        return match[1].str(); // The second capturing group contains the protocol
-    } else {
-        return ""; // Or throw an exception if no protocol is found
-    }
-}
-
-String Request::host() const {
-	return "";
-}
-

src/request.hpp

@@ -11,6 +11,11 @@
 #include <bflibcpp/data.hpp>
 #include <string>
 
+#ifdef LINUX
+#include <linux/limits.h>
+#endif
+#include <limits.h>
+
 typedef enum {
 	kRequestMethodNone = 0,
 	kRequestMethodGet,
@@ -22,20 +27,32 @@ class Request : public BF::Object {
 	Request(const BF::Data * data);
 	virtual ~Request();
 
-	BF::String method() const;
-	BF::String target() const;
-	BF::String protocol() const;
+	const BF::String & method() const;
+	const BF::String & target() const;
+	const BF::String & protocol() const;
+
+	const BF::HashMap<BF::String, BF::String> & header() const;
+	const BF::String & body() const;
 
 	// returns path without query string
 	BF::String targetPath() const;
 
 	// returns query data
 	BF::HashMap<BF::String, BF::String> targetQuery() const;
 
-	BF::String host() const;
-
 private:
+	void parse();
+
 	std::string _message;
+
+	// status line
+	BF::String _method;
+	BF::String _target;
+	BF::String _protocol;
+
+	BF::HashMap<BF::String, BF::String> _header;
+
+	BF::String _body;
 };
 
 #endif // REQUEST_HPP

src/resource.cpp

@@ -33,3 +33,8 @@ bool Resource::setRootFolder(const String & rootFolder) {
 	return true;
 }
 
+bool Resource::targetValid(const URL & target) {
+	URL rootURL = _rootFolder;
+	return target.absURL().isSubPath(rootURL.absURL());
+}
+

src/resource.hpp

@@ -9,10 +9,12 @@
 #define RESOURCE_HPP
 
 #include <bflibcpp/data.hpp>
+#include <bflibcpp/url.hpp>
 
 namespace Resource {
 	bool setRootFolder(const BF::String & rootFolder);
 	const BF::String & getRootFolder();
+	bool targetValid(const BF::URL & target);
 }
 
 #endif // RESOURCE_HPP