xen/tools/ocaml/xenstored/connections.ml

(*
 * Copyright (C) 2006-2007 XenSource Ltd.
 * Copyright (C) 2008      Citrix Ltd.
 * Author Vincent Hanquez <vincent.hanquez@eu.citrix.com>
 * Author Thomas Gazagnaire <thomas.gazagnaire@eu.citrix.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; version 2.1 only. with the special
 * exception on linking described in file LICENSE.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *)

let debug fmt = Logging.debug "connections" fmt

type t = {
  anonymous: (Unix.file_descr, Connection.t) Hashtbl.t;
  domains: (int, Connection.t) Hashtbl.t;
  ports: (Xeneventchn.t, Connection.t) Hashtbl.t;
  mutable watches: Connection.watch list Trie.t;
  mutable has_pending_watchevents: Connection.Watch.Set.t
}

let create () = {
  anonymous = Hashtbl.create 37;
  domains = Hashtbl.create 37;
  ports = Hashtbl.create 37;
  watches = Trie.create ();
  has_pending_watchevents = Connection.Watch.Set.empty;
}

let get_capacity () =
  (* not multiplied by maxwatch on purpose: 2nd queue in watch itself! *)
  { Xenbus.Xb.maxoutstanding = !Define.maxoutstanding; maxwatchevents = !Define.maxwatchevents }

let add_anonymous cons fd =
  let capacity = get_capacity () in
  let xbcon = Xenbus.Xb.open_fd fd ~capacity in
  let con = Connection.create xbcon None in
  Hashtbl.add cons.anonymous (Xenbus.Xb.get_fd xbcon) con

let add_domain cons dom =
  let capacity = get_capacity () in
  let xbcon = Xenbus.Xb.open_mmap ~capacity (Domain.get_interface dom) (fun () -> Domain.notify dom) in
  let con = Connection.create xbcon (Some dom) in
  Hashtbl.add cons.domains (Domain.get_id dom) con;
  Hashtbl.add cons.ports (Domain.get_local_port dom) con

let select ?(only_if = (fun _ -> true)) cons =
  Hashtbl.fold (fun _ con (ins, outs) ->
      if (only_if con) then (
        let fd = Connection.get_fd con in
        let in_fds = if Connection.can_input con then fd :: ins else ins in
        let out_fds = if Connection.has_output con then fd :: outs else outs in
        in_fds, out_fds
      ) else (ins, outs)
    )
    cons.anonymous ([], [])

let find cons =
  Hashtbl.find cons.anonymous

let find_domain cons =
  Hashtbl.find cons.domains

let find_domain_by_port cons port =
  Hashtbl.find cons.ports port

let del_watches_of_con con watches =
  match List.filter (fun w -> Connection.get_con w != con) watches with
  | [] -> None
  | ws -> Some ws

let del_watches cons con =
  Connection.del_watches con;
  cons.watches <- Trie.map (del_watches_of_con con) cons.watches;
  cons.has_pending_watchevents <-
    cons.has_pending_watchevents |> Connection.Watch.Set.filter @@ fun w ->
    Connection.get_con w != con

let del_anonymous cons con =
  try
    Hashtbl.remove cons.anonymous (Connection.get_fd con);
    del_watches cons con;
    Connection.close con
  with exn ->
    debug "del anonymous %s" (Printexc.to_string exn)

let del_domain cons id =
  try
    let con = find_domain cons id in
    Hashtbl.remove cons.domains id;
    (match Connection.get_domain con with
     | Some d -> Hashtbl.remove cons.ports (Domain.get_local_port d)
     | None -> ());
    del_watches cons con;
    Connection.close con
  with exn ->
    debug "del domain %u: %s" id (Printexc.to_string exn)

let iter_domains cons fct =
  Hashtbl.iter (fun _ c -> fct c) cons.domains

let iter_anonymous cons fct =
  Hashtbl.iter (fun _ c -> fct c) cons.anonymous

let iter cons fct =
  iter_domains cons fct; iter_anonymous cons fct

let has_more_work cons =
  Hashtbl.fold
    (fun _id con acc ->
       if Connection.has_more_work con then con :: acc else acc)
    cons.domains []

let key_of_str path =
  if path.[0] = '@'
  then [path]
  else "" :: Store.Path.to_string_list (Store.Path.of_string path)

let key_of_path path =
  "" :: Store.Path.to_string_list path

let add_watch cons con path token =
  let apath = Connection.get_watch_path con path in
  (* fail on invalid paths early by calling key_of_str before adding watch *)
  let key = key_of_str apath in
  let watch = Connection.add_watch con (path, apath) token in
  let watches =
    if Trie.mem cons.watches key
    then Trie.find cons.watches key
    else []
  in
  cons.watches <- Trie.set cons.watches key (watch :: watches);
  watch

let del_watch cons con path token =
  let apath, watch = Connection.del_watch con path token in
  let key = key_of_str apath in
  let watches = Utils.list_remove watch (Trie.find cons.watches key) in
  if watches = [] then
    cons.watches <- Trie.unset cons.watches key
  else
    cons.watches <- Trie.set cons.watches key watches;
  watch

(* path is absolute *)
let fire_watches ?oldroot source root cons path recurse =
  let key = key_of_path path in
  let path = Store.Path.to_string path in
  let roots = oldroot, root in
  let fire_watch _ = function
    | None         -> ()
    | Some watches -> List.iter (fun w -> Connection.fire_watch source roots w path) watches
  in
  let fire_rec _x = function
    | None         -> ()
    | Some watches ->
      List.iter (Connection.fire_single_watch source roots) watches
  in
  Trie.iter_path fire_watch cons.watches key;
  if recurse then
    Trie.iter fire_rec (Trie.sub cons.watches key)

let send_watchevents cons con =
  cons.has_pending_watchevents <-
    cons.has_pending_watchevents |> Connection.Watch.Set.filter Connection.Watch.flush_events;
  Connection.source_flush_watchevents con

let fire_spec_watches root cons specpath =
  let source = find_domain cons 0 in
  iter cons (fun con ->
      List.iter (Connection.fire_single_watch source (None, root)) (Connection.get_watches con specpath))

let set_target cons domain target_domain =
  let con = find_domain cons domain in
  Connection.set_target con target_domain

let number_of_transactions cons =
  let res = ref 0 in
  let aux con =
    res := Connection.number_of_transactions con + !res
  in
  iter cons aux;
  !res

let stats cons =
  let nb_ops_anon = ref 0
  and nb_watchs_anon = ref 0
  and nb_ops_dom = ref 0
  and nb_watchs_dom = ref 0 in
  iter_anonymous cons (fun con ->
      let con_watchs, con_ops = Connection.stats con in
      nb_ops_anon := !nb_ops_anon + con_ops;
      nb_watchs_anon := !nb_watchs_anon + con_watchs;
    );
  iter_domains cons (fun con ->
      let con_watchs, con_ops = Connection.stats con in
      nb_ops_dom := !nb_ops_dom + con_ops;
      nb_watchs_dom := !nb_watchs_dom + con_watchs;
    );
  (Hashtbl.length cons.anonymous, !nb_ops_anon, !nb_watchs_anon,
   Hashtbl.length cons.domains, !nb_ops_dom, !nb_watchs_dom)

let debug cons =
  let anonymous = Hashtbl.fold (fun _ con accu -> Connection.debug con :: accu) cons.anonymous [] in
  let domains = Hashtbl.fold (fun _ con accu -> Connection.debug con :: accu) cons.domains [] in
  String.concat "" (domains @ anonymous)

let debug_watchevents cons con =
  (* == (physical equality)
     	   has to be used here because w.con.xb.backend might contain a [unit->unit] value causing regular
     	   comparison to fail due to having a 'functional value' which cannot be compared.
     	 *)
  let s = cons.has_pending_watchevents |> Connection.Watch.Set.filter (fun w -> w.con == con) in
  let pending = s |> Connection.Watch.Set.elements
                |> List.map (fun w -> Connection.Watch.pending_watchevents w) |> List.fold_left (+) 0 in
  Printf.sprintf "Watches with pending events: %d, pending events total: %d" (Connection.Watch.Set.cardinal s) pending

let filter ~f cons =
  let fold _ v acc = if f v then v :: acc else acc in
  []
  |> Hashtbl.fold fold cons.anonymous
  |> Hashtbl.fold fold cons.domains

let prevents_quit cons = filter ~f:Connection.prevents_live_update cons