Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

wakeup-cleanup: only allow one wakeup calblack per I/O operation #161

Merged
merged 1 commit into from
Jan 13, 2020
Merged

wakeup-cleanup: only allow one wakeup calblack per I/O operation #161

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 31 additions & 31 deletions lib/server_connection.ml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,10 +64,12 @@ type t =
; request_queue : Reqd.t Queue.t
(* invariant: If [request_queue] is not empty, then the head of the queue
has already had [request_handler] called on it. *)
; wakeup_writer : (unit -> unit) list ref
; wakeup_reader : (unit -> unit) list ref
; mutable wakeup_writer : (unit -> unit)
; mutable wakeup_reader : (unit -> unit)
}

let default_wakeup = Sys.opaque_identity (fun () -> ())

let is_closed t =
Reader.is_closed t.reader && Writer.is_closed t.writer

Expand All @@ -80,37 +82,39 @@ let is_active t =
let current_reqd_exn t =
Queue.peek_exn t.request_queue

let on_wakeup_reader t k =
let yield_reader t k =
if is_closed t
then failwith "on_wakeup_reader on closed conn"
else t.wakeup_reader := k::!(t.wakeup_reader)
else if not (t.wakeup_reader == default_wakeup);
then failwith "yield_reader: only one callback can be registered at a time"
else t.wakeup_reader <- k
;;

let wakeup_reader t =
let f = t.wakeup_reader in
t.wakeup_reader <- default_wakeup;
f ()
;;

let on_wakeup_writer t k =
if is_closed t
then failwith "on_wakeup_writer on closed conn"
else t.wakeup_writer := k::!(t.wakeup_writer)
else if not (t.wakeup_writer == default_wakeup)
then failwith "yield_writer: only one callback can be registered at a time"
else t.wakeup_writer <- k
;;

let wakeup_writer t =
let fs = !(t.wakeup_writer) in
t.wakeup_writer := [];
List.iter (fun f -> f ()) fs

let rec _transfer_writer_callbacks fs reqd =
match fs with
| [] -> ()
| f :: fs ->
Reqd.on_more_output_available reqd f;
_transfer_writer_callbacks fs reqd

let transfer_writer_callbacks t reqd =
let fs = !(t.wakeup_writer) in
t.wakeup_writer := [];
_transfer_writer_callbacks fs reqd
let f = t.wakeup_writer in
t.wakeup_writer <- default_wakeup;
f ()
;;

let wakeup_reader t =
let fs = !(t.wakeup_reader) in
t.wakeup_reader := [];
List.iter (fun f -> f ()) fs
let transfer_writer_callback t reqd =
let f = t.wakeup_writer in
t.wakeup_writer <- default_wakeup;
Reqd.on_more_output_available reqd f
;;

let default_error_handler ?request:_ error handle =
let message =
Expand Down Expand Up @@ -145,8 +149,8 @@ let create ?(config=Config.default) ?(error_handler=default_error_handler) reque
; request_handler = request_handler
; error_handler = error_handler
; request_queue
; wakeup_writer = ref []
; wakeup_reader = ref []
; wakeup_writer = default_wakeup
; wakeup_reader = default_wakeup
}

let shutdown_reader t =
Expand Down Expand Up @@ -244,7 +248,7 @@ let read_with_more t bs ~off ~len more =
let reqd = current_reqd_exn t in
if call_handler
then (
transfer_writer_callbacks t reqd;
transfer_writer_callback t reqd;
t.request_handler reqd
);
Reqd.flush_request_body reqd;
Expand All @@ -258,10 +262,6 @@ let read t bs ~off ~len =
let read_eof t bs ~off ~len =
read_with_more t bs ~off ~len Complete

let yield_reader t k =
on_wakeup_reader t k
;;

let flush_response_body t =
if is_active t then
let reqd = current_reqd_exn t in
Expand Down