pv.ml

open Lwt

let block_size = 256 * 4096
let ifd = Lwt_unix.stdin
let ofd = Lwt_unix.stdout

let print spd =
  try_lwt
    Lwt_io.eprintf "%s\r" (Speed.to_string spd)
  with Speed.Undefined -> return_unit

let program () =
  let open Bigarray in
  let open Lwt_bytes in
  let buf = create block_size in

  let rec loop (s) =
    let rec get n =
      if n < block_size
      then read ifd buf n (block_size - n) >>= fun m -> get (n+m)
      else return_unit in

    let rec put n =
      if n < block_size
      then write ofd buf n (block_size - n) >>= fun m -> put (n+m)
      else return_unit in

    get 0 >> put 0 >> print s >> loop Speed.(s + block_size) in

    Speed.create () |> loop

let () = Lwt_main.run (program ())

speed.ml

type t = (float * float) list
type speed = t -> float
exception Undefined


let create () = []

let cur = function
  | (b,t') :: (_,t) :: _ -> b /.  (t' -. t)
  | _ -> raise Not_found


let avg = function
  | [] -> raise Undefined
  | (b,t1) :: ss ->
    let b,t0 = List.fold_left
        (fun (b',t') (b,t) -> b +. b', t) (0.,t1) ss in
    if t0 = t1 then raise Undefined else b /.  (t1 -. t0)

let find_extremum f init = function
  | [] -> raise Undefined
  | (b,t1) :: ss ->
    let (_,spd) = List.fold_left
        (fun (t',spd') (b,t) ->
           let spd = b /. (t' -. t) in
           (t, f spd spd')) (t1,init) ss in
    spd

let sum = List.fold_left (fun total (b,_) -> total +. b) 0.

let max = find_extremum max 0.
let min = find_extremum min max_float

module S = struct
  let (+)  s t = (float t, Unix.gettimeofday ()) :: s
  let (+.) s t = (t, Unix.gettimeofday ()) :: s
end

include S


let to_string spd =
  let m = 1024.0 *. 1024.0 in
  let mbps sel = sel spd /. m in
  Printf.sprintf
    "%8.0f MB : %-4.2f/%-4.2f/%-4.2f/%-4.2f MB/s"
    (sum spd /. m) (mbps cur) (mbps avg) (mbps max) (mbps min)

let print spd = to_string spd |> print_endline

type 'a updater = {
  print: t -> 'a;
  mutable speed: t;
}


let create_updater print = {print; speed = create ();}

let update u wr =
  u.speed <- u.speed + wr;
  u.print u.speed

let inspect u = u.speed

speed.mli

type t
exception Undefined

val create: unit -> t
val (+) : t -> int -> t
val (+.): t -> float -> t

type speed = t -> float
val max: speed
val min: speed
val avg: speed
val cur: speed
val sum: t -> float

val to_string: t -> string
val print: t -> unit


type 'a updater

val create_updater: (t -> 'a) -> 'a updater

val update: 'a updater -> int -> 'a
val inspect: 'a updater -> t

To compile:

$ ocamlbuild -use-ocamlfind -syntax camlp4o -package lwt.unix -pkg lwt.syntax pv.native

Examples:

$ ./pv.native  < /dev/zero > /dev/null
3897 MB : 2438.55/3545.74/6668.21/736.88 MB/s

$ ./pv.native  < /dev/urandom > /dev/null
30 MB : 16.90/16.82/16.99/16.16 MB/s