Skip to content

Commit e1e988b

Browse files
RamazanKaraRamazanKara
committed
feat: use VRBT tree for multiple HTTP call responses
Replace the single http_response struct with a red-black tree (VRBT) keyed by token_id. This enables proxy-wasm modules to dispatch multiple concurrent dispatch_http_call invocations and receive each response via on_http_call_response(token_id). Changes: - proxy_wasm.h: add vwasm_http_call_entry with VRBT_ENTRY, tree head, comparison function, and VRBT_GENERATE_STATIC - proxy_wasm_http.c: allocate tree entry per call, assign unique incrementing token_id - wasm_engine.c: iterate VRBT_FOREACH_SAFE to deliver deferred callbacks for all pending responses - proxy_wasm_headers.c: read headers from active_http_call entry - proxy_wasm.c: free all tree entries on cleanup, read buffer from active_http_call
1 parent 4d4a9aa commit e1e988b

5 files changed

Lines changed: 151 additions & 101 deletions

File tree

‎src/proxy_wasm.c‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -433,9 +433,10 @@ pw_proxy_get_buffer_bytes(void *env, wasmtime_caller_t *caller,
433433
data_len = ctx->plugin_config_len;
434434
break;
435435
case PROXY_BUFFER_HTTP_CALL_BODY:
436-
if (ctx->http_response.valid) {
437-
data = (const char *)ctx->http_response.body;
438-
data_len = ctx->http_response.body_len;
436+
if (ctx->active_http_call != NULL &&
437+
ctx->active_http_call->response.valid) {
438+
data = (const char *)ctx->active_http_call->response.body;
439+
data_len = ctx->active_http_call->response.body_len;
439440
}
440441
break;
441442
case PROXY_BUFFER_HTTP_REQUEST_BODY:
@@ -568,8 +569,9 @@ pw_proxy_get_buffer_status(void *env, wasmtime_caller_t *caller,
568569
data_len = ctx->plugin_config_len;
569570
break;
570571
case PROXY_BUFFER_HTTP_CALL_BODY:
571-
if (ctx->http_response.valid)
572-
data_len = ctx->http_response.body_len;
572+
if (ctx->active_http_call != NULL &&
573+
ctx->active_http_call->response.valid)
574+
data_len = ctx->active_http_call->response.body_len;
573575
break;
574576
case PROXY_BUFFER_HTTP_REQUEST_BODY:
575577
if (ctx->body_modified && ctx->modified_body != NULL)
@@ -1192,12 +1194,16 @@ vwasm_proxy_ctx_cleanup(struct vwasm_proxy_ctx *ctx)
11921194
if (ctx == NULL)
11931195
return;
11941196

1195-
/* Free HTTP call response buffer */
1196-
if (ctx->http_response.raw_buf != NULL) {
1197-
free(ctx->http_response.raw_buf);
1198-
ctx->http_response.raw_buf = NULL;
1199-
ctx->http_response.body = NULL;
1200-
ctx->http_response.valid = 0;
1197+
/* Free all HTTP call response entries */
1198+
{
1199+
struct vwasm_http_call_entry *ent, *tent;
1200+
VRBT_FOREACH_SAFE(ent, vwasm_http_call_tree,
1201+
&ctx->http_calls, tent) {
1202+
vwasm_http_call_tree_VRBT_REMOVE(&ctx->http_calls,
1203+
ent);
1204+
free(ent->response.raw_buf);
1205+
free(ent);
1206+
}
12011207
}
12021208

12031209
/* Free cached request body (only if heap-allocated) */

‎src/proxy_wasm.h‎

Lines changed: 29 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
#include <stdint.h>
1515
#include <pthread.h>
1616
#include <wasmtime.h>
17+
#include "vtree.h"
1718

1819
/* ----------------------------------------------------------------
1920
* Proxy-Wasm ABI enums
@@ -144,6 +145,29 @@ struct vwasm_http_call_response {
144145
int valid;
145146
};
146147

148+
/* Red-black tree entry for multiple outstanding HTTP callouts */
149+
struct vwasm_http_call_entry {
150+
VRBT_ENTRY(vwasm_http_call_entry) entry;
151+
uint32_t token_id;
152+
struct vwasm_http_call_response response;
153+
};
154+
155+
VRBT_HEAD(vwasm_http_call_tree, vwasm_http_call_entry);
156+
157+
static inline int
158+
vwasm_http_call_cmp(const struct vwasm_http_call_entry *a,
159+
const struct vwasm_http_call_entry *b)
160+
{
161+
if (a->token_id < b->token_id)
162+
return (-1);
163+
if (a->token_id > b->token_id)
164+
return (1);
165+
return (0);
166+
}
167+
168+
VRBT_GENERATE_STATIC(vwasm_http_call_tree, vwasm_http_call_entry, entry,
169+
vwasm_http_call_cmp)
170+
147171
/* ----------------------------------------------------------------
148172
* Proxy-Wasm execution context
149173
*
@@ -199,14 +223,12 @@ struct vwasm_proxy_ctx {
199223
/* Metric store (global, not per-ctx) */
200224
struct vwasm_metric_store *metric_store;
201225

202-
/* HTTP call response (for proxy_on_http_call_response) */
203-
struct vwasm_http_call_response http_response;
226+
/* HTTP call responses (VRBT keyed by token_id) */
227+
struct vwasm_http_call_tree http_calls;
228+
uint32_t http_call_next_token;
204229

205-
/* Deferred HTTP call callback (invoked after header fn returns) */
206-
int http_call_pending;
207-
uint32_t http_call_token_id;
208-
uint32_t http_call_num_headers;
209-
size_t http_call_body_len;
230+
/* Active call context (set during proxy_on_http_call_response) */
231+
struct vwasm_http_call_entry *active_http_call;
210232

211233
/* HTTP request/response body access */
212234
const uint8_t *request_body;

‎src/proxy_wasm_headers.c‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -175,8 +175,9 @@ pw_trailer_clear(struct vwasm_trailer_map *tm)
175175
/* ----------------------------------------------------------------
176176
* HTTP call response header helpers
177177
*
178-
* Parse raw HTTP response stored in ctx->http_response to extract
179-
* headers. The raw format is: "HTTP/1.1 200 OK\r\n<headers>\r\n\r\n<body>"
178+
* Parse raw HTTP response stored in ctx->active_http_call->response
179+
* to extract headers. The raw format is:
180+
* "HTTP/1.1 200 OK\r\n<headers>\r\n\r\n<body>"
180181
* ---------------------------------------------------------------- */
181182

182183
/*
@@ -192,11 +193,13 @@ pw_http_call_response_find_header(const struct vwasm_proxy_ctx *ctx,
192193
const char *raw, *end, *p, *line_end;
193194
size_t key_len;
194195

195-
if (!ctx->http_response.valid || ctx->http_response.raw_buf == NULL)
196+
if (ctx->active_http_call == NULL ||
197+
!ctx->active_http_call->response.valid ||
198+
ctx->active_http_call->response.raw_buf == NULL)
196199
return (NULL);
197200

198-
raw = (const char *)ctx->http_response.raw_buf;
199-
end = raw + ctx->http_response.raw_len;
201+
raw = (const char *)ctx->active_http_call->response.raw_buf;
202+
end = raw + ctx->active_http_call->response.raw_len;
200203

201204
/* Find end of status line */
202205
p = raw;
@@ -638,8 +641,9 @@ pw_proxy_get_header_map_pairs(void *env, wasmtime_caller_t *caller,
638641
uint32_t hcount, hoffset;
639642
struct { const char *k; size_t kl; const char *v; size_t vl; } hdrs[64];
640643

641-
if (!ctx->http_response.valid ||
642-
ctx->http_response.raw_buf == NULL) {
644+
if (ctx->active_http_call == NULL ||
645+
!ctx->active_http_call->response.valid ||
646+
ctx->active_http_call->response.raw_buf == NULL) {
643647
if (pw_return_bytes(ctx, NULL, 0,
644648
(uint32_t)args[1].of.i32,
645649
(uint32_t)args[2].of.i32) != 0) {
@@ -650,8 +654,8 @@ pw_proxy_get_header_map_pairs(void *env, wasmtime_caller_t *caller,
650654
return (NULL);
651655
}
652656

653-
raw = (const char *)ctx->http_response.raw_buf;
654-
raw_end = raw + ctx->http_response.raw_len;
657+
raw = (const char *)ctx->active_http_call->response.raw_buf;
658+
raw_end = raw + ctx->active_http_call->response.raw_len;
655659
hcount = 0;
656660

657661
/* Parse status line → :status pseudo-header */

‎src/proxy_wasm_http.c‎

Lines changed: 32 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,6 @@ pw_proxy_http_call(void *env, wasmtime_caller_t *caller,
231231
uint8_t *response_buf = NULL;
232232
size_t response_len = 0;
233233
ssize_t n;
234-
uint32_t token_id = 1;
235234
const char *method;
236235
const char *path;
237236
char extra_headers[4096];
@@ -577,36 +576,40 @@ pw_proxy_http_call(void *env, wasmtime_caller_t *caller,
577576
}
578577
}
579578

580-
/* Store response in context for get_buffer_bytes(HTTP_CALL_BODY) */
581-
ctx->http_response.raw_buf = response_buf;
582-
ctx->http_response.raw_len = response_len;
583-
ctx->http_response.body = body_start;
584-
ctx->http_response.body_len = resp_body_len;
585-
ctx->http_response.num_headers = resp_num_headers;
586-
ctx->http_response.valid = 1;
579+
/* Store response in VRBT tree keyed by token_id */
580+
{
581+
struct vwasm_http_call_entry *ent;
582+
uint32_t token_id;
587583

588-
/* Write token ID */
589-
if (pw_write_u32(ctx, (uint32_t)args[9].of.i32, token_id) != 0) {
590-
free(response_buf);
591-
ctx->http_response.raw_buf = NULL;
592-
ctx->http_response.valid = 0;
593-
results[0].of.i32 = PROXY_INTERNAL;
594-
return (NULL);
595-
}
584+
token_id = ++ctx->http_call_next_token;
596585

597-
/*
598-
* Defer proxy_on_http_call_response to after the current host
599-
* function returns. The proxy-wasm SDK uses RefCell internally
600-
* and panics on re-entrant borrow if we call the callback from
601-
* within the same execution of proxy_on_http_request_headers.
602-
*
603-
* The engine (wasm_engine.c) checks http_call_pending after the
604-
* header function returns and invokes the callback then.
605-
*/
606-
ctx->http_call_pending = 1;
607-
ctx->http_call_token_id = token_id;
608-
ctx->http_call_num_headers = resp_num_headers;
609-
ctx->http_call_body_len = resp_body_len;
586+
ent = calloc(1, sizeof(*ent));
587+
if (ent == NULL) {
588+
free(response_buf);
589+
results[0].of.i32 = PROXY_INTERNAL;
590+
return (NULL);
591+
}
592+
ent->token_id = token_id;
593+
ent->response.raw_buf = response_buf;
594+
ent->response.raw_len = response_len;
595+
ent->response.body = body_start;
596+
ent->response.body_len = resp_body_len;
597+
ent->response.num_headers = resp_num_headers;
598+
ent->response.valid = 1;
599+
600+
vwasm_http_call_tree_VRBT_INSERT(&ctx->http_calls, ent);
601+
602+
/* Write token ID back to the module */
603+
if (pw_write_u32(ctx, (uint32_t)args[9].of.i32,
604+
token_id) != 0) {
605+
vwasm_http_call_tree_VRBT_REMOVE(&ctx->http_calls,
606+
ent);
607+
free(response_buf);
608+
free(ent);
609+
results[0].of.i32 = PROXY_INTERNAL;
610+
return (NULL);
611+
}
612+
}
610613

611614
results[0].of.i32 = PROXY_OK;
612615
return (NULL);

‎src/wasm_engine.c‎

Lines changed: 60 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1127,61 +1127,76 @@ proxy_wasm_execute(struct vwasm_engine *engine,
11271127
}
11281128

11291129
/*
1130-
* Deferred HTTP call callback: if proxy_http_call stored a
1131-
* response during the header function, invoke the callback now
1132-
* that the header function has returned (avoids RefCell panic
1130+
* Deferred HTTP call callbacks: if proxy_http_call stored one or
1131+
* more responses during the header function, invoke the callbacks
1132+
* now that the header function has returned (avoids RefCell panic
11331133
* in the proxy-wasm Rust SDK from re-entrant borrows).
1134+
*
1135+
* Iterate the VRBT in insertion order (ascending token_id).
11341136
*/
1135-
if (proxy_ctx.http_call_pending) {
1137+
if (!VRBT_EMPTY(&proxy_ctx.http_calls)) {
11361138
wasmtime_extern_t cb_item;
1137-
proxy_ctx.http_call_pending = 0;
1139+
struct vwasm_http_call_entry *ent, *tent;
11381140

11391141
if (wasmtime_instance_export_get(context, &instance,
11401142
"proxy_on_http_call_response", 27, &cb_item) &&
11411143
cb_item.kind == WASMTIME_EXTERN_FUNC) {
1142-
wasmtime_val_t cb_args[5];
1143-
wasmtime_error_t *cb_err;
1144-
wasm_trap_t *cb_trap = NULL;
1145-
1146-
wasmtime_context_set_epoch_deadline(context,
1147-
VWASM_DEFAULT_EPOCH_DEADLINE_MS);
1148-
1149-
cb_args[0].kind = WASMTIME_I32;
1150-
cb_args[0].of.i32 =
1151-
(int32_t)proxy_ctx.stream_context_id;
1152-
cb_args[1].kind = WASMTIME_I32;
1153-
cb_args[1].of.i32 =
1154-
(int32_t)proxy_ctx.http_call_token_id;
1155-
cb_args[2].kind = WASMTIME_I32;
1156-
cb_args[2].of.i32 =
1157-
(int32_t)proxy_ctx.http_call_num_headers;
1158-
cb_args[3].kind = WASMTIME_I32;
1159-
cb_args[3].of.i32 =
1160-
(int32_t)proxy_ctx.http_call_body_len;
1161-
cb_args[4].kind = WASMTIME_I32;
1162-
cb_args[4].of.i32 = 0; /* num_trailers */
1163-
1164-
cb_err = wasmtime_func_call(context,
1165-
&cb_item.of.func, cb_args, 5,
1166-
NULL, 0, &cb_trap);
1167-
if (cb_err != NULL)
1168-
wasmtime_error_delete(cb_err);
1169-
if (cb_trap != NULL)
1170-
wasm_trap_delete(cb_trap);
11711144

1145+
VRBT_FOREACH_SAFE(ent, vwasm_http_call_tree,
1146+
&proxy_ctx.http_calls, tent) {
1147+
wasmtime_val_t cb_args[5];
1148+
wasmtime_error_t *cb_err;
1149+
wasm_trap_t *cb_trap = NULL;
1150+
1151+
wasmtime_context_set_epoch_deadline(context,
1152+
VWASM_DEFAULT_EPOCH_DEADLINE_MS);
1153+
1154+
/* Set active call so header maps can look up
1155+
* the correct response */
1156+
proxy_ctx.active_http_call = ent;
1157+
1158+
cb_args[0].kind = WASMTIME_I32;
1159+
cb_args[0].of.i32 =
1160+
(int32_t)proxy_ctx.stream_context_id;
1161+
cb_args[1].kind = WASMTIME_I32;
1162+
cb_args[1].of.i32 =
1163+
(int32_t)ent->token_id;
1164+
cb_args[2].kind = WASMTIME_I32;
1165+
cb_args[2].of.i32 =
1166+
(int32_t)ent->response.num_headers;
1167+
cb_args[3].kind = WASMTIME_I32;
1168+
cb_args[3].of.i32 =
1169+
(int32_t)ent->response.body_len;
1170+
cb_args[4].kind = WASMTIME_I32;
1171+
cb_args[4].of.i32 = 0; /* num_trailers */
1172+
1173+
cb_err = wasmtime_func_call(context,
1174+
&cb_item.of.func, cb_args, 5,
1175+
NULL, 0, &cb_trap);
1176+
if (cb_err != NULL)
1177+
wasmtime_error_delete(cb_err);
1178+
if (cb_trap != NULL)
1179+
wasm_trap_delete(cb_trap);
1180+
1181+
proxy_ctx.active_http_call = NULL;
1182+
1183+
/*
1184+
* If the callback sent a local response,
1185+
* stop iterating and return immediately.
1186+
*/
1187+
if (proxy_ctx.local_response_set) {
1188+
*status_code =
1189+
proxy_ctx.local_response_code;
1190+
ret = 0;
1191+
goto cleanup;
1192+
}
1193+
}
11721194
/*
1173-
* After successful callback, reset action to
1174-
* CONTINUE so the request proceeds (the original
1175-
* PAUSE was only to wait for the HTTP call).
1176-
* If the callback sent a local response, return
1177-
* the status code immediately.
1195+
* All callbacks succeeded without sending a local
1196+
* response — reset action to CONTINUE so the
1197+
* request proceeds (the original PAUSE was only to
1198+
* wait for the HTTP call(s)).
11781199
*/
1179-
if (proxy_ctx.local_response_set) {
1180-
*status_code =
1181-
proxy_ctx.local_response_code;
1182-
ret = 0;
1183-
goto cleanup;
1184-
}
11851200
action = 0;
11861201
}
11871202
}

0 commit comments

Comments
 (0)