baraba/bux-lang
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(nexus): HTTP/1.1 keep-alive (~2× throughput)

Browse Source 
Reuse TCP connections for multiple requests, honor Connection close,
and raise wrk health RPS from ~45k to ~80k+ on the same machine.

- Loop HandleConnection up to 1000 reqs/fd with Connection keep-alive
- RawRequest_WantsKeepAlive (HTTP/1.1 default; HTTP/1.0 needs keep-alive)
- Nexus 0.3.0 banner; benches note before/after RPS
This commit is contained in:
dimgigov
2026-07-19 22:53:59 +03:00
parent abb2b54dce
commit e30d8a5eb9
6 changed files with 134 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/nexus/src/Handlers.bux
+4 -3
View File
@@ -63,16 +63,17 @@ module Handlers {
pub func HandleApiHealth() -> HttpResponse {
return Http_NewResponse(200, "application/json; charset=utf-8",
"{\"status\":\"ok\",\"server\":\"Nexus\",\"version\":\"0.2.0\"}");
"{\"status\":\"ok\",\"server\":\"Nexus\",\"version\":\"0.3.0\"}");
}
pub func HandleApiInfo() -> HttpResponse {
return Http_NewResponse(200, "application/json; charset=utf-8",
"{\"name\":\"Nexus\",\"language\":\"Bux\",\"features\":[\"HTTP/1.1\",\"thread-pool\",\"algebraic-enums\"]}");
"{\"name\":\"Nexus\",\"language\":\"Bux\",\"features\":[\"HTTP/1.1\",\"keep-alive\",\"thread-pool\",\"algebraic-enums\"]}");
}
pub func HandleWebSocketUpgrade(req: HttpRequest) -> HttpResponse {
let wsKey: String = RequestHeader_Get(req, "Sec-WebSocket-Key");
var reqMut: HttpRequest = req;
let wsKey: String = RequestHeader_Get(&reqMut, "Sec-WebSocket-Key");
var resp: HttpResponse;
resp.statusCode = 101;
apps/nexus/src/Http.bux
+54 -5
View File
@@ -1,7 +1,7 @@
module Http {
import Std::Array::{Array};
import Std::String::{String_Eq, String_EndsWith, String_Contains};
import Std::Array::{Array, Array_Len, Array_Get};
import Std::String::{String_Eq, String_EndsWith, String_Contains, String_Len, String_StartsWith};
pub enum HttpMethod {
GET,
@@ -95,13 +95,62 @@ module Http {
return resp;
}
pub func RequestHeader_Get(req: HttpRequest, key: String) -> String {
for entry in req.headers {
if String_Eq(entry.key, key) {
func CharLower(c: int) -> int {
if c >= 65 && c <= 90 {
return c + 32;
}
return c;
}
/// Case-insensitive string equality (HTTP header names/values).
pub func String_EqIgnoreCase(a: String, b: String) -> bool {
let la: uint = String_Len(a);
let lb: uint = String_Len(b);
if la != lb {
return false;
}
var i: uint = 0;
while i < la {
if CharLower(a[i] as int) != CharLower(b[i] as int) {
return false;
}
i = i + 1;
}
return true;
}
pub func RequestHeader_Get(req: *HttpRequest, key: String) -> String {
// Index-based walk — for-in over Array<struct{String}> can corrupt string fields.
let n: uint = Array_Len<HeaderEntry>(&req.headers);
var i: uint = 0;
while i < n {
let entry: HeaderEntry = Array_Get<HeaderEntry>(&req.headers, i);
if String_EqIgnoreCase(entry.key, key) {
return entry.value;
}
i = i + 1;
}
return "";
}
/// Decide keep-alive from the raw request bytes (avoids fragile header Array walk).
/// HTTP/1.1 defaults to keep-alive; Connection: close forces close;
/// HTTP/1.0 needs explicit keep-alive.
pub func RawRequest_WantsKeepAlive(raw: String) -> bool {
if String_Contains(raw, "Connection: close") || String_Contains(raw, "connection: close") ||
String_Contains(raw, "CONNECTION: CLOSE") {
return false;
}
// HTTP/1.0 without Keep-Alive → close
if String_Contains(raw, "HTTP/1.0") {
if String_Contains(raw, "Connection: keep-alive") ||
String_Contains(raw, "Connection: Keep-Alive") ||
String_Contains(raw, "connection: keep-alive") {
return true;
}
return false;
}
return true;
}
}
apps/nexus/src/Server.bux
+51 -26
View File
@@ -5,7 +5,7 @@ module Server {
import Std::String::{String_Len, String_StartsWith};
import Std::Channel::{Channel, Channel_New, Channel_Send, Channel_Recv};
import Config::{ServerConfig};
import Http::{HttpRequest, HttpResponse, Http_StatusText, Http_NewResponse};
import Http::{HttpRequest, HttpResponse, Http_StatusText, Http_NewResponse, RawRequest_WantsKeepAlive};
import Errors::{ParseResult};
import Parser::{ParseRequest};
import Router::{Router, Router_Dispatch};
@@ -17,11 +17,14 @@ module Server {
extern func bux_sb_build(sb: *void) -> String;
extern func bux_sb_free(sb: *void);
/// Cap requests per TCP connection (safety + fair scheduling).
const MAX_KEEPALIVE_REQUESTS: int = 1000;
pub struct ConnectionTask {
fd: int;
}
pub func BuildResponse(resp: HttpResponse) -> String {
pub func BuildResponse(resp: HttpResponse, keepAlive: bool) -> String {
let sb: *void = bux_sb_new(4096);
bux_sb_append(sb, "HTTP/1.1 ");
@@ -30,7 +33,7 @@ module Server {
bux_sb_append(sb, Http_StatusText(resp.statusCode));
bux_sb_append(sb, "\r\n");
bux_sb_append(sb, "Server: Nexus/0.2.0 (Bux)\r\n");
bux_sb_append(sb, "Server: Nexus/0.3.0 (Bux)\r\n");
if bux_strlen(resp.extraHeaders) > 0 {
bux_sb_append(sb, resp.extraHeaders);
@@ -46,7 +49,13 @@ module Server {
bux_sb_append(sb, "Content-Length: ");
bux_sb_append_int(sb, bodyLen as int64);
bux_sb_append(sb, "\r\n");
bux_sb_append(sb, "Connection: close\r\n");
if keepAlive {
bux_sb_append(sb, "Connection: keep-alive\r\n");
bux_sb_append(sb, "Keep-Alive: timeout=5, max=1000\r\n");
} else {
bux_sb_append(sb, "Connection: close\r\n");
}
bux_sb_append(sb, "\r\n");
if bodyLen > 0 {
@@ -58,28 +67,44 @@ module Server {
return result;
}
/// Serve one TCP client: zero or more HTTP requests (HTTP/1.1 keep-alive).
pub func HandleConnection(fd: int, router: Router) {
let raw: String = Net_Recv(fd, 8192);
if String_Len(raw) == 0 {
return;
}
var reqCount: int = 0;
while reqCount < MAX_KEEPALIVE_REQUESTS {
let raw: String = Net_Recv(fd, 8192);
if String_Len(raw) == 0 {
return;
}
// HTTP/2 preface detection
if String_StartsWith(raw, "PRI * HTTP/2.0") {
let resp: HttpResponse = Http_NewResponse(200, "text/plain; charset=utf-8",
"HTTP/2 detected — full support planned for future release.\r\n");
Net_Send(fd, BuildResponse(resp));
return;
}
// HTTP/2 preface detection — one-shot response, then close
if String_StartsWith(raw, "PRI * HTTP/2.0") {
let resp: HttpResponse = Http_NewResponse(200, "text/plain; charset=utf-8",
"HTTP/2 detected — full support planned for future release.\r\n");
Net_Send(fd, BuildResponse(resp, false));
return;
}
let parsed: ParseResult = ParseRequest(raw);
if parsed.tag == ParseResult_Ok {
let req: HttpRequest = parsed.data.Ok_0;
let resp: HttpResponse = Router_Dispatch(router, req);
Net_Send(fd, BuildResponse(resp));
} else {
let resp: HttpResponse = Http_NewResponse(400, "text/plain; charset=utf-8", "Bad Request");
Net_Send(fd, BuildResponse(resp));
let parsed: ParseResult = ParseRequest(raw);
var keepAlive: bool = false;
if parsed.tag == ParseResult_Ok {
let req: HttpRequest = parsed.data.Ok_0;
keepAlive = RawRequest_WantsKeepAlive(raw);
// Last request on the connection quota must close
if reqCount + 1 >= MAX_KEEPALIVE_REQUESTS {
keepAlive = false;
}
let resp: HttpResponse = Router_Dispatch(router, req);
Net_Send(fd, BuildResponse(resp, keepAlive));
} else {
let resp: HttpResponse = Http_NewResponse(400, "text/plain; charset=utf-8", "Bad Request");
Net_Send(fd, BuildResponse(resp, false));
return;
}
reqCount = reqCount + 1;
if !keepAlive {
return;
}
}
}
@@ -113,8 +138,8 @@ module Server {
pub func RunServer(config: ServerConfig, router: Router) -> int {
PrintLine("================================================");
PrintLine(" Nexus HTTP Server v0.2.0");
PrintLine(" Production-ready HTTP/1.1 with thread-pool");
PrintLine(" Nexus HTTP Server v0.3.0");
PrintLine(" HTTP/1.1 keep-alive + thread-pool");
PrintLine(" Built with Bux");
PrintLine("================================================");
PrintLine("");
@@ -148,7 +173,7 @@ module Server {
PrintInt(config.port);
PrintLine("");
PrintInt(config.workerCount);
PrintLine(" worker threads | static: ./public/");
PrintLine(" worker threads | keep-alive: on | static: ./public/");
PrintLine("Endpoints: / /api/health /api/info /ws");
PrintLine("Press Ctrl+C to stop.");
PrintLine("");