<commit_before>type +'a t = ('a -> unit) -> unit

let return x k = k x
let bind t f k = t (fun x -> f x k)
let ( >>> ) a b k = a (fun () -> b k)
let ( >>= ) t f k = t (fun x -> f x k)
let ( >>| ) t f k = t (fun x -> k (f x))

let both a b =
  a >>= fun a ->
  b >>= fun b -> return (a, b)

module Ivar = struct
  type 'a state = Full of 'a | Empty of ('a -> unit) Queue.t
  type 'a t = { mutable state : 'a state }

  let create () = { state = Empty (Queue.create ()) }

  let fill t x =
    match t.state with
    | Full _ -> failwith "Ivar.fill"
    | Empty q ->
        t.state <- Full x;
        Queue.iter (fun f -> f x) q

  let read t k = match t.state with Full x -> k x | Empty q -> Queue.push k q
end

let fork f k =
  let ivar = Ivar.create () in
  f () (fun x -> Ivar.fill ivar x);
  k ivar

let fork_and_join f g =
  fork f >>= fun a ->
  fork g >>= fun b -> both (Ivar.read a) (Ivar.read b)

let fork_and_join_unit f g =
  fork f >>= fun a ->
  fork g >>= fun b -> Ivar.read a >>> Ivar.read b

type seq = { mutable prev : seq; mutable next : seq }
type prgn = Prgn : ('a -> unit) * 'a -> prgn

type node = {
  mutable node_prev : seq;
  mutable node_next : seq;
  prgn : prgn;
  mutable active : bool;
}

external node_of_seq : seq -> node = "%identity"
external seq_of_node : node -> seq = "%identity"

let is_empty seq = seq.next == seq

let remove node =
  if node.active then (
    node.active <- true;
    let seq = seq_of_node node in
    seq.prev.next <- seq.next;
    seq.next.prev <- seq.prev)

let pop seq =
  if is_empty seq then None
  else
    let res = node_of_seq seq.next in
    remove res;
    Some res.prgn

let add seq prgn =
  let node = { node_prev = seq.prev; node_next = seq; prgn; active = true } in
  seq.prev.next <- seq_of_node node;
  seq.prev <- seq_of_node node

let make_seq () =
  let rec seq = { prev = seq; next = seq } in
  seq

type pool = {
  seq : seq;
  work_mutex : Mutex.t;
  work_cond : Condition.t;
  working_cond : Condition.t;
  mutable working_cnt : int;
  mutable thread_cnt : int;
  mutable stop : bool;
}

let wait pool =
  Mutex.lock pool.work_mutex;
  let rec loop () =
    if
      ((not pool.stop) && pool.working_cnt <> 0)
      || (pool.stop && pool.thread_cnt <> 0)
    then (
      Condition.wait pool.working_cond pool.work_mutex;
      loop ())
  in
  loop ();
  Mutex.unlock pool.work_mutex

let worker pool =
  let rec loop () =
    Mutex.lock pool.work_mutex;
    while is_empty pool.seq && not pool.stop do
      Condition.wait pool.work_cond pool.work_mutex
    done;
    if not pool.stop then (
      let res = pop pool.seq in
      pool.working_cnt <- pool.working_cnt + 1;
      Mutex.unlock pool.work_mutex;
      (try Option.iter (fun (Prgn (f, a)) -> f a) res with _exn -> ());
      Mutex.lock pool.work_mutex;
      pool.working_cnt <- pool.working_cnt - 1;<commit_msg>Revert "Fix a bit fiber"<commit_after>type +'a t = ('a -> unit) -> unit

let return x k = k x
let bind t f k = t (fun x -> f x k)
let ( >>> ) a b k = a (fun () -> b k)
let ( >>= ) t f k = t (fun x -> f x k)
let ( >>| ) t f k = t (fun x -> k (f x))

let both a b =
  a >>= fun a ->
  b >>= fun b -> return (a, b)

module Ivar = struct
  type 'a state = Full of 'a | Empty of ('a -> unit) Queue.t
  type 'a t = { mutable state : 'a state }

  let create () = { state = Empty (Queue.create ()) }

  let fill t x =
    match t.state with
    | Full _ -> failwith "Ivar.fill"
    | Empty q ->
        t.state <- Full x;
        Queue.iter (fun f -> f x) q

  let read t k = match t.state with Full x -> k x | Empty q -> Queue.push k q
end

let fork f k =
  let ivar = Ivar.create () in
  f () (fun x -> Ivar.fill ivar x);
  k ivar

let fork_and_join f g =
  fork f >>= fun a ->
  fork g >>= fun b -> both (Ivar.read a) (Ivar.read b)

let fork_and_join_unit f g =
  fork f >>= fun a ->
  fork g >>= fun b -> Ivar.read a >>> Ivar.read b

type seq = { mutable prev : seq; mutable next : seq }
type prgn = Prgn : ('a -> unit) * 'a -> prgn

type node = {
  mutable node_prev : seq;
  mutable node_next : seq;
  prgn : prgn;
  mutable active : bool;
}

external node_of_seq : seq -> node = "%identity"
external seq_of_node : node -> seq = "%identity"

let is_empty seq = seq.next == seq

let remove node =
  if node.active then (
    node.active <- true;
    let seq = seq_of_node node in
    seq.prev.next <- seq.next;
    seq.next.prev <- seq.prev)

let pop seq =
  if is_empty seq then None
  else
    let res = node_of_seq seq.next in
    remove res;
    Some res.prgn

let add seq prgn =
  let node = { node_prev = seq.prev; node_next = seq; prgn; active = true } in
  seq.prev.next <- seq_of_node node;
  seq.prev <- seq_of_node node

let make_seq () =
  let rec seq = { prev = seq; next = seq } in
  seq

type pool = {
  seq : seq;
  work_mutex : Mutex.t;
  work_cond : Condition.t;
  working_cond : Condition.t;
  mutable working_cnt : int;
  mutable thread_cnt : int;
  mutable stop : bool;
}

let wait pool =
  Mutex.lock pool.work_mutex;
  let rec loop () =
    if
      ((not pool.stop) && (pool.working_cnt <> 0 || not (is_empty pool.seq)))
      || (pool.stop && pool.thread_cnt <> 0)
    then (
      Condition.wait pool.working_cond pool.work_mutex;
      loop ())
  in
  loop ();
  Mutex.unlock pool.work_mutex

let worker pool =
  let rec loop () =
    Mutex.lock pool.work_mutex;
    while is_empty pool.seq && not pool.stop do
      Condition.wait pool.work_cond pool.work_mutex
    done;
    if not pool.stop then (
      let res = pop pool.seq in
      pool.working_cnt <- pool.working_cnt + 1;
      Mutex.unlock pool.work_mutex;
      (try Option.iter (fun (Prgn (f, a)) -> f a) res with _exn -> ());
      Mutex.lock pool.work_mutex;
      pool.working_cnt <- pool.working_cnt - 1;
{'fraction_non_alphanumeric': 0.13443478260869565, 'fraction_numerical': 0.001391304347826087, 'mean_word_length': 2.2938144329896906, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 101, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3664082', 'n_tokens_mistral': 2187, 'n_tokens_neox': 2141, 'n_words': 762}