The unpac monorepo manager self-hosting as a monorepo using unpac
Merge pull request #752 from talex5/linux-get-sched
eio_linux: refactor fixed buffer code
authored by
Thomas Leonard
and committed by
GitHub
ed212e14
2b800a45
+34
-39
+1
vendor/opam/eio/lib_eio/core/eio__core.ml
+1
vendor/opam/eio/lib_eio/core/eio__core.ml
+1
vendor/opam/eio/lib_eio/core/eio__core.mli
+1
vendor/opam/eio/lib_eio/core/eio__core.mli
+26
-3
vendor/opam/eio/lib_eio_linux/low_level.ml
+26
-3
vendor/opam/eio/lib_eio_linux/low_level.ml
···
207
207
raise @@ Err.wrap (Uring.error_of_errno res) "write" ""
208
208
)
209
209
210
-
let alloc_fixed () = Effect.perform Sched.Alloc
210
+
let alloc_fixed () =
211
+
let s = Sched.get () in
212
+
match s.mem with
213
+
| None -> None
214
+
| Some mem ->
215
+
match Uring.Region.alloc mem with
216
+
| buf -> Some buf
217
+
| exception Uring.Region.No_space -> None
211
218
212
-
let alloc_fixed_or_wait () = Effect.perform Sched.Alloc_or_wait
219
+
let alloc_fixed_or_wait () =
220
+
let s = Sched.get () in
221
+
match s.mem with
222
+
| None -> failwith "No fixed buffer available"
223
+
| Some mem ->
224
+
match Uring.Region.alloc mem with
225
+
| buf -> buf
226
+
| exception Uring.Region.No_space ->
227
+
let id = Eio.Private.Trace.mint_id () in
228
+
let trigger = Eio.Private.Single_waiter.create () in
229
+
Queue.push trigger s.mem_q;
230
+
(* todo: remove protect; but needs to remove from queue on cancel *)
231
+
Eio.Private.Single_waiter.await_protect trigger "alloc_fixed_or_wait" id
213
232
214
-
let free_fixed buf = Effect.perform (Sched.Free buf)
233
+
let free_fixed buf =
234
+
let s = Sched.get () in
235
+
match Queue.take_opt s.mem_q with
236
+
| None -> Uring.Region.free buf
237
+
| Some k -> Eio.Private.Single_waiter.wake k (Ok buf)
215
238
216
239
let splice src ~dst ~len =
217
240
Fd.use_exn "splice-src" src @@ fun src ->
+6
-36
vendor/opam/eio/lib_eio_linux/sched.ml
+6
-36
vendor/opam/eio/lib_eio_linux/sched.ml
···
50
50
uring: io_job Uring.t;
51
51
mem: Uring.Region.t option;
52
52
io_q: (t -> unit) Queue.t; (* waiting for room on [uring] *)
53
-
mem_q : Uring.Region.chunk Suspended.t Queue.t;
53
+
mem_q : Uring.Region.chunk Eio.Private.Single_waiter.t Queue.t;
54
54
55
55
(* The queue of runnable fibers ready to be resumed. Note: other domains can also add work items here. *)
56
56
run_q : runnable Lf_queue.t;
···
74
74
type _ Effect.t +=
75
75
| Enter : (t -> 'a Suspended.t -> unit) -> 'a Effect.t
76
76
| Cancel : io_job Uring.job -> unit Effect.t
77
-
| Alloc : Uring.Region.chunk option Effect.t
78
-
| Alloc_or_wait : Uring.Region.chunk Effect.t
79
-
| Free : Uring.Region.chunk -> unit Effect.t
77
+
| Get : t Effect.t
78
+
79
+
let get () = Effect.perform Get
80
80
81
81
let wake_buffer =
82
82
let b = Bytes.create 8 in
···
339
339
| _, Exactly len -> Suspended.continue action len
340
340
| n, Upto _ -> Suspended.continue action n
341
341
342
-
let alloc_buf_or_wait st k =
343
-
match st.mem with
344
-
| None -> Suspended.discontinue k (Failure "No fixed buffer available")
345
-
| Some mem ->
346
-
match Uring.Region.alloc mem with
347
-
| buf -> Suspended.continue k buf
348
-
| exception Uring.Region.No_space ->
349
-
Queue.push k st.mem_q;
350
-
schedule st
351
-
352
-
let free_buf st buf =
353
-
match Queue.take_opt st.mem_q with
354
-
| None -> Uring.Region.free buf
355
-
| Some k -> enqueue_thread st k buf
356
-
357
342
let rec enqueue_poll_add fd poll_mask st action =
358
343
Trace.log "poll_add";
359
344
let retry = with_cancel_hook ~action st (fun () ->
···
411
396
Fiber_context.destroy fiber;
412
397
Printexc.raise_with_backtrace ex (Printexc.get_raw_backtrace ())
413
398
);
414
-
effc = fun (type a) (e : a Effect.t) ->
399
+
effc = fun (type a) (e : a Effect.t) : ((a, _) continuation -> _) option ->
415
400
match e with
401
+
| Get -> Some (fun k -> continue k st)
416
402
| Enter fn -> Some (fun k ->
417
403
match Fiber_context.get_error fiber with
418
404
| Some e -> discontinue k e
···
466
452
let enqueue x = enqueue_thread st k (x, st.thread_pool) in
467
453
Eio_unix.Private.Thread_pool.submit st.thread_pool ~ctx:fiber ~enqueue fn;
468
454
schedule st
469
-
)
470
-
| Alloc -> Some (fun k ->
471
-
match st.mem with
472
-
| None -> continue k None
473
-
| Some mem ->
474
-
match Uring.Region.alloc mem with
475
-
| buf -> continue k (Some buf)
476
-
| exception Uring.Region.No_space -> continue k None
477
-
)
478
-
| Alloc_or_wait -> Some (fun k ->
479
-
let k = { Suspended.k; fiber } in
480
-
alloc_buf_or_wait st k
481
-
)
482
-
| Free buf -> Some (fun k ->
483
-
free_buf st buf;
484
-
continue k ()
485
455
)
486
456
| e -> extra_effects.effc e
487
457
}