package eio

  1. Overview
  2. Docs
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file buf_read.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
exception Buffer_limit_exceeded

open Std

type t = {
  mutable buf : Cstruct.buffer;
  mutable pos : int;
  mutable len : int;
  mutable flow : Flow.source_ty r option;    (* None if we've seen eof *)
  mutable consumed : int;                    (* Total bytes consumed so far *)
  max_size : int;
}

type 'a parser = t -> 'a

let return = Fun.const

let map f x r = f (x r)

let pair x y r =
  let a = x r in
  let b = y r in
  a, b

let bind x f r = f (x r) r

module Syntax = struct
  let ( let+ ) x f r = f (x r)
  let ( let* ) = bind
  let ( and* ) = pair
  let ( and+ ) = pair

  let ( <*> ) = pair

  let ( <* ) a b t =
    let x = a t in
    ignore (b t);
    x

  let ( *> ) a b t =
    ignore (a t);
    b t
end

open Syntax

let capacity t = Bigarray.Array1.dim t.buf

let of_flow ?initial_size ~max_size flow =
  let flow = (flow :> Flow.source_ty r) in
  if max_size <= 0 then Fmt.invalid_arg "Max size %d should be positive!" max_size;
  let initial_size = Option.value initial_size ~default:(min 4096 max_size) in
  let buf = Bigarray.(Array1.create char c_layout initial_size) in
  { buf; pos = 0; len = 0; flow = Some flow; max_size; consumed = 0 }

let of_buffer buf =
  let len = Bigarray.Array1.dim buf in
  { buf; pos = 0; len; flow = None; max_size = max_int; consumed = 0 }

let of_string s =
  let len = String.length s in
  let buf = Bigarray.(Array1.create char c_layout) len in
  Cstruct.blit_from_string s 0 (Cstruct.of_bigarray buf) 0 len;
  of_buffer buf

let peek t =
  Cstruct.of_bigarray ~off:t.pos ~len:t.len t.buf

let consume_err t n =
  Fmt.invalid_arg "Can't consume %d bytes of a %d byte buffer!" n t.len

let [@inline] consume t n =
  if n < 0 || n > t.len then consume_err t n;
  t.pos <- t.pos + n;
  t.len <- t.len - n;
  t.consumed <- t.consumed + n

let consume_all t =
  t.consumed <- t.consumed + t.len;
  t.len <- 0

let buffered_bytes t = t.len

let consumed_bytes t = t.consumed

let eof_seen t = t.flow = None

let ensure_slow_path t n =
  assert (n >= 0);
  if n > t.max_size then raise Buffer_limit_exceeded;
  (* We don't have enough data yet, so we'll need to do a read. *)
  match t.flow with
  | None -> raise End_of_file
  | Some flow ->
    (* If the buffer is empty, we might as well use all of it: *)
    if t.len = 0 then t.pos <- 0;
    let () =
      let cap = capacity t in
      if n > cap then (
        (* [n] bytes won't fit. We need to resize the buffer. *)
        let new_size = max n (min t.max_size (cap * 2)) in
        let new_buf = Bigarray.(Array1.create char c_layout new_size) in
        Cstruct.blit
          (peek t) 0
          (Cstruct.of_bigarray new_buf) 0
          t.len;
        t.pos <- 0;
        t.buf <- new_buf
      ) else if t.pos + n > cap then (
        (* [n] bytes will fit in the existing buffer, but we need to compact it first. *)
        Cstruct.blit
          (peek t) 0
          (Cstruct.of_bigarray t.buf) 0
          t.len;
        t.pos <- 0
      )
    in
    try
      while t.len < n do
        let free_space = Cstruct.of_bigarray t.buf ~off:(t.pos + t.len) in
        assert (t.len + Cstruct.length free_space >= n);
        let got = Flow.single_read flow free_space in
        t.len <- t.len + got
      done;
      assert (buffered_bytes t >= n)
    with End_of_file ->
      t.flow <- None;
      raise End_of_file

let ensure t n =
  if t.len < n then ensure_slow_path t n

module F = struct
  type nonrec t = t

  let single_read t dst =
    ensure t 1;
    let len = min (buffered_bytes t) (Cstruct.length dst) in
    Cstruct.blit (peek t) 0 dst 0 len;
    consume t len;
    len

  let rsb t fn =
    ensure t 1;
    let data = peek t in
    let sent = fn [data] in
    consume t sent

  let read_methods = [Flow.Read_source_buffer rsb]
end

let as_flow =
  let ops = Flow.Pi.source (module F) in
  fun t -> Resource.T (t, ops)

let get t i =
  Bigarray.Array1.get t.buf (t.pos + i)

module BE = struct
  let uint16 t = 
    ensure t 2;
    let data = Bigstringaf.get_int16_be t.buf t.pos in
    consume t 2;
    data

  let uint32 t = 
    ensure t 4;
    let data = Bigstringaf.get_int32_be t.buf t.pos in
    consume t 4;
    data

  let uint48 t = 
    ensure t 6;
    let upper_32 = Bigstringaf.get_int32_be t.buf t.pos |> Int64.of_int32 |> Int64.logand 0xffffffffL in
    let lower_16 = Bigstringaf.get_int16_be t.buf (t.pos + 4) |> Int64.of_int in
    consume t 6;
    Int64.(
      logor 
        (lower_16)
        (shift_left upper_32 16) 
    )

  let uint64 t =
    ensure t 8;
    let data = Bigstringaf.get_int64_be t.buf t.pos in
    consume t 8;
    data

  let float t =
    ensure t 4;
    let data = Bigstringaf.unsafe_get_int32_be t.buf t.pos in
    consume t 4;
    Int32.float_of_bits data

  let double t =
    ensure t 8;
    let data = Bigstringaf.unsafe_get_int64_be t.buf t.pos in
    consume t 8;
    Int64.float_of_bits data
end

module LE = struct
  let uint16 t = 
    ensure t 2;
    let data = Bigstringaf.get_int16_le t.buf t.pos in
    consume t 2;
    data

  let uint32 t = 
    ensure t 4;
    let data = Bigstringaf.get_int32_le t.buf t.pos in
    consume t 4;
    data

  let uint48 t = 
    ensure t 6;
    let lower_32 = Bigstringaf.get_int32_le t.buf t.pos |> Int64.of_int32 |> Int64.logand 0xffffffffL in
    let upper_16 = Bigstringaf.get_int16_le t.buf (t.pos + 4) |> Int64.of_int in
    consume t 6;
    Int64.(
      logor 
        (shift_left upper_16 32)
        lower_32
    )

  let uint64 t =
    ensure t 8;
    let data = Bigstringaf.get_int64_le t.buf t.pos in
    consume t 8;
    data

  let float t =
    ensure t 4;
    let data = Bigstringaf.unsafe_get_int32_le t.buf t.pos in
    consume t 4;
    Int32.float_of_bits data

  let double t =
    ensure t 8;
    let data = Bigstringaf.unsafe_get_int64_le t.buf t.pos in
    consume t 8;
    Int64.float_of_bits data
end

let char c t =
  ensure t 1;
  let c2 = get t 0 in
  if c <> c2 then Fmt.failwith "Expected %C but got %C" c c2;
  consume t 1

let any_char t =
  ensure t 1;
  let c = get t 0 in
  consume t 1;
  c

let uint8 t = Char.code (any_char t)

let peek_char t =
  match ensure t 1 with
  | () -> Some (get t 0)
  | exception End_of_file -> None

let take len t =
  if len < 0 then Fmt.invalid_arg "take: %d is negative!" len;
  ensure t len;
  let data = Cstruct.to_string (Cstruct.of_bigarray t.buf ~off:t.pos ~len) in
  consume t len;
  data

let string s t =
  let rec aux i =
    if i = String.length s then (
      consume t i
    ) else if i < t.len then (
      if get t i = s.[i] then aux (i + 1)
      else (
        let buf = peek t in
        let len = min (String.length s) (Cstruct.length buf) in
        Fmt.failwith "Expected %S but got %S"
          s
          (Cstruct.to_string buf ~off:0 ~len)
      )
    ) else (
      ensure t (t.len + 1);
      aux i
    )
  in
  aux 0

let take_all t =
  try
    while true do ensure t (t.len + 1) done;
    assert false
  with End_of_file ->
    let data = Cstruct.to_string (peek t) in
    consume t t.len;
    data

let count_while p t =
  let rec aux i =
    if i < t.len then (
      if p (get t i) then aux (i + 1)
      else i
    ) else (
      ensure t (t.len + 1);
      aux i
    )
  in
  try aux 0
  with End_of_file -> t.len

let take_while p t =
  let len = count_while p t in
  let data = Cstruct.to_string (Cstruct.of_bigarray t.buf ~off:t.pos ~len) in
  consume t len;
  data

let take_while1 p t =
  let len = count_while p t in
  if len < 1 then Fmt.failwith "take_while1"
  else
    let data = Cstruct.to_string (Cstruct.of_bigarray t.buf ~off:t.pos ~len) in
    consume t len;
    data

let skip_while p t =
  let rec aux i =
    if i < t.len then (
      if p (get t i) then aux (i + 1)
      else consume t i
    ) else (
      consume t t.len;
      ensure t 1;
      aux 0
    )
  in
  try aux 0
  with End_of_file -> ()

let skip_while1 p t =
  let len = count_while p t in
  if len < 1 then Fmt.failwith "skip_while1"
  else consume t len

let rec skip n t =
  if n <= t.len then (
    consume t n
  ) else (
    let n = n - t.len in
    consume_all t;
    ensure t (min n (capacity t));
    skip n t
  )

let skip n t =
  if n < 0 then Fmt.invalid_arg "skip: %d is negative!" n;
  try skip n t
  with End_of_file ->
    (* Skip isn't atomic, so discard everything in this case for consistency. *)
    consume t t.len;
    raise End_of_file

let line t =
  (* Return the index of the first '\n', reading more data as needed. *)
  let rec aux i =
    if i = t.len then (
      ensure t (t.len + 1);
      aux i
    ) else if get t i = '\n' then (
      i
    ) else (
      aux (i + 1)
    )
  in
  match aux 0 with
  | exception End_of_file when t.len > 0 -> take_all t
  | nl ->
    let len =
      if nl > 0 && get t (nl - 1) = '\r' then nl - 1
      else nl
    in
    let line = Cstruct.to_string (Cstruct.of_bigarray t.buf ~off:t.pos ~len) in
    consume t (nl + 1);
    line

let at_end_of_input t =
  if t.len = 0 && eof_seen t then true
  else (
    match ensure t 1 with
    | () -> false
    | exception End_of_file -> true
  )

let end_of_input t =
  if not (at_end_of_input t) then
    failwith "Unexpected data after parsing"

let pp_pos f t =
  Fmt.pf f "at offset %d" (consumed_bytes t)

let format_errors p t =
  match p t with
  | v -> Ok v
  | exception Failure msg -> Fmt.error_msg "%s (%a)" msg pp_pos t
  | exception End_of_file -> Fmt.error_msg "Unexpected end-of-file at offset %d" (t.consumed + t.len)
  | exception Buffer_limit_exceeded -> Fmt.error_msg "Buffer size limit exceeded when reading %a" pp_pos t

let parse ?initial_size ~max_size p flow =
  let buf = of_flow flow ?initial_size ~max_size in
  format_errors (p <* end_of_input) buf

let parse_exn ?initial_size ~max_size p flow =
  match parse ?initial_size ~max_size p flow with
  | Ok x -> x
  | Error (`Msg m) -> failwith m

let parse_string p s =
  format_errors (p <* end_of_input) (of_string s)

let parse_string_exn p s =
  match parse_string p s with
  | Ok x -> x
  | Error (`Msg m) -> failwith m

[@@inline never]
let bad_offset ~expected actual =
  Fmt.invalid_arg "Sequence is stale (expected to be used at offset %d, but stream is now at %d)"
    expected actual

let seq ?(stop=at_end_of_input) p t =
  let rec aux offset () =
    if offset <> t.consumed then bad_offset ~expected:offset t.consumed;
    if stop t then Seq.Nil
    else (
      let item = p t in
      Seq.Cons (item, aux t.consumed)
    )
  in
  aux t.consumed

let lines t = seq line t
OCaml

Innovation. Community. Security.