Skip to content

Commit b1dfdf1

Browse files
c-cubec-cube
committed
feat: tiny_httpd_eio library
provides a tiny_httpd server that relies on Eio for non-blocking sockets and for concurrency using eio fibers.
1 parent 3658895 commit b1dfdf1

File tree

5 files changed

+286
-0
lines changed

5 files changed

+286
-0
lines changed

‎dune-project‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,3 +39,12 @@
3939
(iostream-camlzip (>= 0.2.1))
4040
(logs :with-test)
4141
(odoc :with-doc)))
42+
43+
(package
44+
(name tiny_httpd_eio)
45+
(synopsis "Use eio for tiny_httpd")
46+
(depends
47+
(tiny_httpd (= :version))
48+
eio
49+
(logs :with-test)
50+
(odoc :with-doc)))

‎src/eio/dune‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
2+
(library
3+
(name tiny_httpd_eio)
4+
(public_name tiny_httpd_eio)
5+
(synopsis "An EIO-based backend for Tiny_httpd")
6+
(flags :standard -safe-string -warn-error -a+8)
7+
(libraries tiny_httpd eio eio.unix))
8+

‎src/eio/tiny_httpd_eio.ml‎

Lines changed: 207 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,207 @@
1+
module IO = Tiny_httpd.IO
2+
module H = Tiny_httpd.Server
3+
module Pool = Tiny_httpd.Pool
4+
module Slice = IO.Slice
5+
module Log = Tiny_httpd.Log
6+
7+
let ( let@ ) = ( @@ )
8+
9+
type 'a with_args =
10+
?addr:string ->
11+
?port:int ->
12+
?unix_sock:string ->
13+
?max_connections:int ->
14+
?max_buf_pool_size:int ->
15+
stdenv:Eio_unix.Stdenv.base ->
16+
sw:Eio.Switch.t ->
17+
'a
18+
19+
let get_max_connection_ ?(max_connections = 64) () : int =
20+
let max_connections = max 4 max_connections in
21+
max_connections
22+
23+
let buf_size = 16 * 1024
24+
25+
let eio_ipaddr_to_unix (a : _ Eio.Net.Ipaddr.t) : Unix.inet_addr =
26+
(* TODO: for ipv4 we really could do it faster via sprintf 🙄 *)
27+
Unix.inet_addr_of_string (Format.asprintf "%a" Eio.Net.Ipaddr.pp a)
28+
29+
let eio_sock_addr_to_unix (a : Eio.Net.Sockaddr.stream) : Unix.sockaddr =
30+
match a with
31+
| `Tcp (h, p) -> Unix.ADDR_INET (eio_ipaddr_to_unix h, p)
32+
| `Unix s -> Unix.ADDR_UNIX s
33+
34+
let ic_of_flow ~buf_pool:ic_pool (flow : _ Eio.Net.stream_socket) : IO.Input.t =
35+
let cstruct = Pool.Raw.acquire ic_pool in
36+
37+
object
38+
inherit Iostream.In_buf.t_from_refill ()
39+
40+
method private refill (sl : Slice.t) =
41+
assert (sl.len = 0);
42+
let cap = min (Bytes.length sl.bytes) (Cstruct.length cstruct) in
43+
44+
match Eio.Flow.single_read flow (Cstruct.sub cstruct 0 cap) with
45+
| exception End_of_file ->
46+
Log.debug (fun k -> k "read: eof");
47+
()
48+
| n ->
49+
Log.debug (fun k -> k "read %d bytes..." n);
50+
Cstruct.blit_to_bytes cstruct 0 sl.bytes 0 n;
51+
sl.off <- 0;
52+
sl.len <- n
53+
54+
method close () =
55+
Pool.Raw.release ic_pool cstruct;
56+
Eio.Flow.shutdown flow `Receive
57+
end
58+
59+
let oc_of_flow ~buf_pool:oc_pool (flow : _ Eio.Net.stream_socket) : IO.Output.t
60+
=
61+
(* write buffer *)
62+
let wbuf : Cstruct.t = Pool.Raw.acquire oc_pool in
63+
let offset = ref 0 in
64+
65+
object (self)
66+
method flush () : unit =
67+
if !offset > 0 then (
68+
Eio.Flow.write flow [ Cstruct.sub wbuf 0 !offset ];
69+
offset := 0
70+
)
71+
72+
method output buf i len =
73+
let i = ref i in
74+
let len = ref len in
75+
76+
while !len > 0 do
77+
let available = Cstruct.length wbuf - !offset in
78+
let n = min !len available in
79+
Cstruct.blit_from_bytes buf !i wbuf !offset n;
80+
offset := !offset + n;
81+
i := !i + n;
82+
len := !len - n;
83+
84+
if !offset = Cstruct.length wbuf then self#flush ()
85+
done
86+
87+
method output_char c =
88+
if !offset = Cstruct.length wbuf then self#flush ();
89+
Cstruct.set_char wbuf !offset c;
90+
incr offset;
91+
if !offset = Cstruct.length wbuf then self#flush ()
92+
93+
method close () =
94+
Pool.Raw.release oc_pool wbuf;
95+
Eio.Flow.shutdown flow `Send
96+
end
97+
98+
let io_backend ?addr ?port ?unix_sock ?max_connections ?max_buf_pool_size
99+
~(stdenv : Eio_unix.Stdenv.base) ~(sw : Eio.Switch.t) () :
100+
(module H.IO_BACKEND) =
101+
let addr, port, (sockaddr : Eio.Net.Sockaddr.stream) =
102+
match addr, port, unix_sock with
103+
| _, _, Some s -> Printf.sprintf "unix:%s" s, 0, `Unix s
104+
| addr, port, None ->
105+
let addr = Option.value ~default:"127.0.0.1" addr in
106+
let sockaddr, port =
107+
match Eio.Net.getaddrinfo stdenv#net addr, port with
108+
| `Tcp (h, _) :: _, None ->
109+
let p = 8080 in
110+
`Tcp (h, p), p
111+
| `Tcp (h, _) :: _, Some p -> `Tcp (h, p), p
112+
| _ ->
113+
failwith @@ Printf.sprintf "Could not parse TCP address from %S" addr
114+
in
115+
addr, port, sockaddr
116+
in
117+
118+
let module M = struct
119+
let init_addr () = addr
120+
let init_port () = port
121+
let get_time_s () = Unix.gettimeofday ()
122+
let max_connections = get_max_connection_ ?max_connections ()
123+
124+
let pool_size =
125+
match max_buf_pool_size with
126+
| Some n -> n
127+
| None -> min 4096 (max_connections * 2)
128+
129+
let cstruct_pool =
130+
Pool.create ~max_size:max_connections
131+
~mk_item:(fun () -> Cstruct.create buf_size)
132+
()
133+
134+
let tcp_server () : IO.TCP_server.builder =
135+
{
136+
IO.TCP_server.serve =
137+
(fun ~after_init ~handle () : unit ->
138+
let running = Atomic.make true in
139+
let active_conns = Atomic.make 0 in
140+
141+
Eio.Switch.on_release sw (fun () -> Atomic.set running false);
142+
let net = Eio.Stdenv.net stdenv in
143+
144+
(* main server socket *)
145+
let sock =
146+
let backlog = max_connections in
147+
Eio.Net.listen ~reuse_addr:true ~reuse_port:true ~backlog ~sw net
148+
sockaddr
149+
in
150+
151+
let tcp_server : IO.TCP_server.t =
152+
{
153+
running = (fun () -> Atomic.get running);
154+
stop =
155+
(fun () ->
156+
Atomic.set running false;
157+
Eio.Switch.fail sw Exit);
158+
endpoint =
159+
(fun () ->
160+
(* TODO: find the real port *)
161+
addr, port);
162+
active_connections = (fun () -> Atomic.get active_conns);
163+
}
164+
in
165+
166+
after_init tcp_server;
167+
168+
while Atomic.get running do
169+
Eio.Net.accept_fork ~sw
170+
~on_error:(fun exn ->
171+
Log.error (fun k ->
172+
k "error in client handler: %s" (Printexc.to_string exn)))
173+
sock
174+
(fun flow client_addr ->
175+
Atomic.incr active_conns;
176+
let@ () =
177+
Fun.protect ~finally:(fun () ->
178+
Log.debug (fun k ->
179+
k "Tiny_httpd_eio: client handler returned");
180+
Atomic.decr active_conns)
181+
in
182+
let ic = ic_of_flow ~buf_pool:cstruct_pool flow in
183+
let oc = oc_of_flow ~buf_pool:cstruct_pool flow in
184+
185+
Log.debug (fun k ->
186+
k "handling client on %a…" Eio.Net.Sockaddr.pp client_addr);
187+
let client_addr_unix = eio_sock_addr_to_unix client_addr in
188+
try handle.handle ~client_addr:client_addr_unix ic oc
189+
with exn ->
190+
let bt = Printexc.get_raw_backtrace () in
191+
Log.error (fun k ->
192+
k "Client handler for %a failed with %s\n%s"
193+
Eio.Net.Sockaddr.pp client_addr
194+
(Printexc.to_string exn)
195+
(Printexc.raw_backtrace_to_string bt)))
196+
done);
197+
}
198+
end in
199+
(module M)
200+
201+
let create ?addr ?port ?unix_sock ?max_connections ?max_buf_pool_size ~stdenv
202+
~sw ?buf_size ?middlewares () : H.t =
203+
let backend =
204+
io_backend ?addr ?port ?unix_sock ?max_buf_pool_size ?max_connections
205+
~stdenv ~sw ()
206+
in
207+
H.create_from ?buf_size ?middlewares ~backend ()

‎src/eio/tiny_httpd_eio.mli‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
(** Tiny httpd EIO backend.
2+
3+
This replaces the threads + Unix blocking syscalls of {!Tiny_httpd_server}
4+
with an Eio-based cooperative system.
5+
6+
{b NOTE}: this is very experimental and will absolutely change over time,
7+
especially since Eio itself is also subject to change.
8+
@since NEXT_RELEASE *)
9+
10+
(* TODO: pass in a switch *)
11+
12+
type 'a with_args =
13+
?addr:string ->
14+
?port:int ->
15+
?unix_sock:string ->
16+
?max_connections:int ->
17+
?max_buf_pool_size:int ->
18+
stdenv:Eio_unix.Stdenv.base ->
19+
sw:Eio.Switch.t ->
20+
'a
21+
22+
val io_backend : (unit -> (module Tiny_httpd.Server.IO_BACKEND)) with_args
23+
(** Create a server *)
24+
25+
val create :
26+
(?buf_size:int ->
27+
?middlewares:([ `Encoding | `Stage of int ] * Tiny_httpd.Middleware.t) list ->
28+
unit ->
29+
Tiny_httpd.Server.t)
30+
with_args
31+
(** Create a server *)

‎tiny_httpd_eio.opam‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
# This file is generated by dune, edit dune-project instead
2+
opam-version: "2.0"
3+
version: "0.19"
4+
synopsis: "Use eio for tiny_httpd"
5+
maintainer: ["c-cube"]
6+
authors: ["c-cube"]
7+
license: "MIT"
8+
homepage: "https://github.com/c-cube/tiny_httpd/"
9+
bug-reports: "https://github.com/c-cube/tiny_httpd/issues"
10+
depends: [
11+
"dune" {>= "3.2"}
12+
"tiny_httpd" {= version}
13+
"eio"
14+
"logs" {with-test}
15+
"odoc" {with-doc}
16+
]
17+
build: [
18+
["dune" "subst"] {dev}
19+
[
20+
"dune"
21+
"build"
22+
"-p"
23+
name
24+
"-j"
25+
jobs
26+
"@install"
27+
"@runtest" {with-test}
28+
"@doc" {with-doc}
29+
]
30+
]
31+
dev-repo: "git+https://github.com/c-cube/tiny_httpd.git"

0 commit comments

Comments
 (0)