<commit_before>open Lwt
open Hostnet

let src =
  let src = Logs.Src.create "usernet" ~doc:"Mirage TCP/IP <-> socket proxy" in
  Logs.Src.set_level src (Some Logs.Debug);
  src

module Log = (val Logs.src_log src : Logs.LOG)

let log_exception_continue description f =
  Lwt.catch
    (fun () -> f ())
    (fun e ->
       Log.err (fun f -> f "%s: caught %s" description (Printexc.to_string e));
       Lwt.return ()
    )

let default d = function None -> d | Some x -> x

let ethernet_serviceid = "30D48B34-7D27-4B0B-AAAF-BBBED334DD59"
let ports_serviceid = "0B95756A-9985-48AD-9470-78E060895BE7"

(* These could be shared with datakit. Perhaps they should live in mirage/conduit? *)

let make_unix_socket path =
  Lwt.catch
    (fun () -> Lwt_unix.unlink path)
    (function
      | Unix.Unix_error(Unix.ENOENT, _, _) -> Lwt.return ()
      | e -> Lwt.fail e)
  >>= fun () ->
  let s = Lwt_unix.(socket PF_UNIX SOCK_STREAM 0) in
  Lwt_unix.bind s (Lwt_unix.ADDR_UNIX path);
  Lwt.return s

let unix_accept_forever url socket callback =
  Lwt_unix.listen socket 5;
  let rec aux () =
    Lwt_unix.accept socket >>= fun (client, _addr) ->
    let _ = (* background thread *)
      (* the callback will close the connection when its done *)
      callback client in
    aux () in
  Log.debug (fun l -> l "Waiting for connections on socket %S" url);
  aux ()

let rec named_pipe_accept_forever path callback =
  let open Lwt.Infix in
  let p = Named_pipe_lwt.Server.create path in
  Named_pipe_lwt.Server.connect p
  >>= function
  | false ->
    Log.err (fun f -> f "Named-pipe connection failed on %s" path);
    Lwt.return ()
  | true ->
    let _ = (* background thread *)
      let fd = Named_pipe_lwt.Server.to_fd p in
      callback fd in
    named_pipe_accept_forever path callback

let hvsock_connect_forever url sockaddr callback =
  Log.info (fun f -> f "connecting to %s:%s" (Hvsock.string_of_vmid sockaddr.Hvsock.vmid) sockaddr.Hvsock.serviceid);
  let rec aux () =
    let socket = Lwt_hvsock.create () in
    Lwt.catch
      (fun () ->
        Lwt_hvsock.connect socket sockaddr
        >>= fun () ->
        Log.info (fun f -> f "hvsock connected successfully");
        callback socket
      ) (function
        | Unix.Unix_error(e, _, _) ->
          Lwt_hvsock.close socket
          >>= fun () ->
          Log.debug (fun f -> f "hvsock connect got %s: retrying in 1s" (Win_error.error_message e));
          Lwt_unix.sleep 1.
        | e ->
          Log.err (fun f -> f "hvsock connect raised %s" (Printexc.to_string e));
          Lwt.fail e
      )
    >>= fun () ->
    aux () in<commit_msg>hvsock: don't fail even if the error is unexpected<commit_after>open Lwt
open Hostnet

let src =
  let src = Logs.Src.create "usernet" ~doc:"Mirage TCP/IP <-> socket proxy" in
  Logs.Src.set_level src (Some Logs.Debug);
  src

module Log = (val Logs.src_log src : Logs.LOG)

let log_exception_continue description f =
  Lwt.catch
    (fun () -> f ())
    (fun e ->
       Log.err (fun f -> f "%s: caught %s" description (Printexc.to_string e));
       Lwt.return ()
    )

let default d = function None -> d | Some x -> x

let ethernet_serviceid = "30D48B34-7D27-4B0B-AAAF-BBBED334DD59"
let ports_serviceid = "0B95756A-9985-48AD-9470-78E060895BE7"

(* These could be shared with datakit. Perhaps they should live in mirage/conduit? *)

let make_unix_socket path =
  Lwt.catch
    (fun () -> Lwt_unix.unlink path)
    (function
      | Unix.Unix_error(Unix.ENOENT, _, _) -> Lwt.return ()
      | e -> Lwt.fail e)
  >>= fun () ->
  let s = Lwt_unix.(socket PF_UNIX SOCK_STREAM 0) in
  Lwt_unix.bind s (Lwt_unix.ADDR_UNIX path);
  Lwt.return s

let unix_accept_forever url socket callback =
  Lwt_unix.listen socket 5;
  let rec aux () =
    Lwt_unix.accept socket >>= fun (client, _addr) ->
    let _ = (* background thread *)
      (* the callback will close the connection when its done *)
      callback client in
    aux () in
  Log.debug (fun l -> l "Waiting for connections on socket %S" url);
  aux ()

let rec named_pipe_accept_forever path callback =
  let open Lwt.Infix in
  let p = Named_pipe_lwt.Server.create path in
  Named_pipe_lwt.Server.connect p
  >>= function
  | false ->
    Log.err (fun f -> f "Named-pipe connection failed on %s" path);
    Lwt.return ()
  | true ->
    let _ = (* background thread *)
      let fd = Named_pipe_lwt.Server.to_fd p in
      callback fd in
    named_pipe_accept_forever path callback

let hvsock_connect_forever url sockaddr callback =
  Log.info (fun f -> f "connecting to %s:%s" (Hvsock.string_of_vmid sockaddr.Hvsock.vmid) sockaddr.Hvsock.serviceid);
  let rec aux () =
    let socket = Lwt_hvsock.create () in
    Lwt.catch
      (fun () ->
        Lwt_hvsock.connect socket sockaddr
        >>= fun () ->
        Log.info (fun f -> f "hvsock connected successfully");
        callback socket
      ) (function
        | Unix.Unix_error(e, _, _) ->
          Lwt_hvsock.close socket
          >>= fun () ->
          Log.debug (fun f -> f "hvsock connect got %s: retrying in 1s" (Win_error.error_message e));
          Lwt_unix.sleep 1.
        | e ->
          Log.err (fun f -> f "hvsock connect raised %s" (Printexc.to_string e));
          Lwt_unix.sleep 1.
      )
    >>= fun () ->
    aux () in
{'fraction_non_alphanumeric': 0.11888509670079636, 'fraction_numerical': 0.01725445582100872, 'mean_word_length': 2.5690121786197566, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 77, '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': '3852162', 'n_tokens_mistral': 1957, 'n_tokens_neox': 1898, 'n_words': 531}