MIRROR: javascript for 🐜's, a tiny runtime with big ambitions
improve blob streaming
+58
-20
+1
examples/demo/event_loop.js
+1
examples/demo/event_loop.js
···
19
19
else formatted = String(Math.round(rate));
20
20
21
21
console.log(`\x1b[1;36m${formatted} event loop iterations/sec\x1b[0m \x1b[2m(${count.toLocaleString()} in ${(elapsed / 1000).toFixed(1)}s)\x1b[0m`);
22
+
if (typeof globalThis.Ant !== 'undefined') console.log('ant version:', Ant.version);
22
23
});
+7
-5
include/streams/readable.h
+7
-5
include/streams/readable.h
···
53
53
ant_value_t rs_cancel_resolve(ant_t *js, ant_value_t *args, int nargs);
54
54
ant_value_t readable_stream_cancel(ant_t *js, ant_value_t stream_obj, ant_value_t reason);
55
55
ant_value_t rs_create_stream(ant_t *js, ant_value_t pull_fn, ant_value_t cancel_fn, double hwm);
56
-
ant_value_t rs_create_stream(ant_t *js, ant_value_t pull_fn, ant_value_t cancel_fn, double hwm);
57
56
58
-
bool rs_reader_has_reqs(ant_t *js, ant_value_t reader_obj);
59
-
bool rs_default_controller_can_close_or_enqueue(rs_controller_t *ctrl, rs_stream_t *stream);
60
-
57
+
void rs_controller_close(ant_t *js, ant_value_t ctrl_obj);
61
58
void rs_default_controller_clear_algorithms(ant_value_t ctrl_obj);
59
+
void rs_ctrl_queue_push(ant_t *js, ant_value_t ctrl_obj, ant_value_t value);
60
+
void rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk);
62
61
void rs_default_controller_call_pull_if_needed(ant_t *js, ant_value_t controller_obj);
63
62
void rs_default_reader_error_read_requests(ant_t *js, ant_value_t reader_obj, ant_value_t e);
64
63
void rs_fulfill_read_request(ant_t *js, ant_value_t stream_obj, ant_value_t chunk, bool done);
65
-
void rs_ctrl_queue_push(ant_t *js, ant_value_t ctrl_obj, ant_value_t value);
64
+
66
65
void readable_stream_close(ant_t *js, ant_value_t stream_obj);
67
66
void readable_stream_error(ant_t *js, ant_value_t stream_obj, ant_value_t e);
67
+
68
+
bool rs_reader_has_reqs(ant_t *js, ant_value_t reader_obj);
69
+
bool rs_default_controller_can_close_or_enqueue(rs_controller_t *ctrl, rs_stream_t *stream);
68
70
69
71
#endif
+7
-15
src/modules/blob.c
+7
-15
src/modules/blob.c
···
12
12
#include "internal.h"
13
13
#include "descriptors.h"
14
14
15
-
#include "silver/engine.h"
16
15
#include "modules/blob.h"
17
16
#include "modules/buffer.h"
18
17
#include "modules/symbol.h"
···
284
283
static ant_value_t blob_stream_pull(ant_t *js, ant_value_t *args, int nargs) {
285
284
ant_value_t blob_obj = js_get_slot(js->current_func, SLOT_DATA);
286
285
blob_data_t *bd = blob_get_data(blob_obj);
287
-
ant_value_t controller = (nargs > 0) ? args[0] : js_mkundef();
286
+
ant_value_t ctrl = (nargs > 0) ? args[0] : js_mkundef();
288
287
289
288
if (bd && bd->size > 0 && bd->data) {
290
-
ArrayBufferData *ab = create_array_buffer_data(bd->size);
291
-
if (ab) {
289
+
ArrayBufferData *ab = create_array_buffer_data(bd->size);
290
+
if (ab) {
292
291
memcpy(ab->data, bd->data, bd->size);
293
-
ant_value_t chunk = create_typed_array(js, TYPED_ARRAY_UINT8, ab, 0, bd->size, "Uint8Array");
294
-
ant_value_t enqueue_fn = js_get(js, controller, "enqueue");
295
-
if (is_callable(enqueue_fn)) {
296
-
ant_value_t enq_args[1] = { chunk };
297
-
sv_vm_call(js->vm, js, enqueue_fn, controller, enq_args, 1, NULL, false);
298
-
}}
299
-
}
292
+
rs_controller_enqueue(js, ctrl, create_typed_array(js, TYPED_ARRAY_UINT8, ab, 0, bd->size, "Uint8Array"));
293
+
}}
300
294
301
-
ant_value_t close_fn = js_get(js, controller, "close");
302
-
if (is_callable(close_fn)) sv_vm_call(js->vm, js, close_fn, controller, NULL, 0, NULL, false);
303
-
295
+
rs_controller_close(js, ctrl);
304
296
return js_mkundef();
305
297
}
306
298
307
299
static ant_value_t js_blob_stream(ant_t *js, ant_value_t *args, int nargs) {
308
300
ant_value_t pull_fn = js_heavy_mkfun(js, blob_stream_pull, js->this_val);
309
-
return rs_create_stream(js, pull_fn, js_mkundef(), 0);
301
+
return rs_create_stream(js, pull_fn, js_mkundef(), 1);
310
302
}
311
303
312
304
static ant_value_t js_blob_ctor(ant_t *js, ant_value_t *args, int nargs) {
+43
src/streams/readable.c
+43
src/streams/readable.c
···
495
495
return js_mkundef();
496
496
}
497
497
498
+
void rs_controller_enqueue(ant_t *js, ant_value_t ctrl_obj, ant_value_t chunk) {
499
+
rs_controller_t *ctrl = rs_get_controller(ctrl_obj);
500
+
if (!ctrl) return;
501
+
502
+
ant_value_t stream_obj = rs_ctrl_stream(ctrl_obj);
503
+
rs_stream_t *stream = rs_get_stream(stream_obj);
504
+
if (!rs_default_controller_can_close_or_enqueue(ctrl, stream)) return;
505
+
506
+
ant_value_t reader_obj = rs_stream_reader(stream_obj);
507
+
if (is_object_type(reader_obj) && rs_reader_has_reqs(js, reader_obj)) {
508
+
rs_fulfill_read_request(js, stream_obj, chunk, false);
509
+
rs_default_controller_call_pull_if_needed(js, ctrl_obj);
510
+
return;
511
+
}
512
+
513
+
rs_ctrl_queue_push(js, ctrl_obj, chunk);
514
+
if (ctrl->queue_sizes_len >= ctrl->queue_sizes_cap) {
515
+
uint32_t new_cap = ctrl->queue_sizes_cap ? ctrl->queue_sizes_cap * 2 : 4;
516
+
double *new_sizes = realloc(ctrl->queue_sizes, new_cap * sizeof(double));
517
+
if (new_sizes) { ctrl->queue_sizes = new_sizes; ctrl->queue_sizes_cap = new_cap; }
518
+
}
519
+
520
+
if (ctrl->queue_sizes_len < ctrl->queue_sizes_cap)
521
+
ctrl->queue_sizes[ctrl->queue_sizes_len++] = 1;
522
+
ctrl->queue_total_size += 1;
523
+
rs_default_controller_call_pull_if_needed(js, ctrl_obj);
524
+
}
525
+
526
+
void rs_controller_close(ant_t *js, ant_value_t ctrl_obj) {
527
+
rs_controller_t *ctrl = rs_get_controller(ctrl_obj);
528
+
if (!ctrl) return;
529
+
530
+
ant_value_t stream_obj = rs_ctrl_stream(ctrl_obj);
531
+
rs_stream_t *stream = rs_get_stream(stream_obj);
532
+
if (!rs_default_controller_can_close_or_enqueue(ctrl, stream)) return;
533
+
ctrl->close_requested = true;
534
+
535
+
if (rs_ctrl_queue_len(js, ctrl_obj) == 0) {
536
+
rs_default_controller_clear_algorithms(ctrl_obj);
537
+
readable_stream_close(js, stream_obj);
538
+
}
539
+
}
540
+
498
541
static ant_value_t js_rs_reader_get_closed(ant_t *js, ant_value_t *args, int nargs) {
499
542
return rs_reader_closed(js->this_val);
500
543
}