Source file call.ml

(*
 * Copyright (C) Citrix Systems Inc.
 *
 * 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.
 *)

open Xs_protocol
open Junk

let ( |> ) a b = b a
let debug fmt = Logging.debug "call" fmt
let error fmt = Logging.error "call" fmt

exception Parse_failure
exception Transaction_again
exception Transaction_nested

let get_namespace_implementation path =
  match Store.Path.to_string_list path with
  | "tool" :: "xenstored" :: "quota" :: rest ->
      (Store.Path.of_string_list rest, (module Quota_interface : Namespace.IO))
  | "tool" :: "xenstored" :: "connection" :: rest ->
      ( Store.Path.of_string_list rest
      , (module Connection.Interface : Namespace.IO) )
  | "tool" :: "xenstored" :: "log" :: rest ->
      (Store.Path.of_string_list rest, (module Logging_interface : Namespace.IO))
  | "tool" :: "xenstored" :: "memory" :: rest ->
      ( Store.Path.of_string_list rest
      , (module Heap_debug_interface : Namespace.IO) )
  | _ -> (path, (module Transaction : Namespace.IO))

(* Perform a 'simple' operation (not a Transaction_start or Transaction_end)
   and create a response. *)
let op_exn _store c t (payload : Request.payload) : Response.payload =
  let connection_path = c.Connection.domainpath in
  let resolve data = Store.Path.create data connection_path in

  let open Request in
  match payload with
  | Transaction_start | Transaction_end _
  | Watch (_, _)
  | Unwatch (_, _)
  | Debug _
  | Introduce (_, _, _)
  | Resume _ | Release _
  | Set_target (_, _)
  | Isintroduced _ | Error _ | Watchevent _ ->
      assert false
  | Getdomainpath domid ->
      let v = Store.Path.getdomainpath domid |> Store.Path.to_string in
      Response.Getdomainpath v
  | PathOp (path, op) -> (
      let path = resolve path in

      let path, m = get_namespace_implementation path in
      let module Impl = (val m : Namespace.IO) in
      let mkdir_p t creator perm path =
        let dirname = Store.Path.get_parent path in
        if not (Impl.exists t perm dirname) then
          let rec check_path p =
            match p with
            | [] -> []
            | h :: l -> if Impl.exists t perm h then check_path l else p
          in
          let ret = check_path (List.tl (Store.Path.get_hierarchy dirname)) in
          List.iter (fun s -> Impl.mkdir ~with_watch:false t creator perm s) ret
      in
      match op with
      | Read ->
          let v = Impl.read t c.Connection.perm path in
          Response.Read v
      | Directory ->
          let entries = Impl.list t c.Connection.perm path in
          Response.Directory entries
      | Directory_part _ -> raise Parse_failure
      | Getperms ->
          let v = Impl.getperms t c.Connection.perm path in
          Response.Getperms v
      | Write value ->
          mkdir_p t c.Connection.domid c.Connection.perm path;
          Impl.write t c.Connection.domid c.Connection.perm path value;
          Response.Write
      | Mkdir ->
          mkdir_p t c.Connection.domid c.Connection.perm path;
          Impl.mkdir t c.Connection.domid c.Connection.perm path;
          Response.Mkdir
      | Rm ->
          Impl.rm t c.Connection.perm path;
          Response.Rm
      | Setperms perms ->
          Impl.setperms t c.Connection.perm path perms;
          Response.Setperms)

(* Replay a stored transaction against a fresh store, check the responses are
   all equivalent: if so, commit the transaction. Otherwise send the abort to
   the client. *)
let transaction_replay store c t =
  let ops = Transaction.get_operations t in
  let tid = Connection.register_transaction c store in
  let t = Transaction.make tid store in
  let con = "replay request:" ^ c.Connection.domstr in
  let perform_exn (request, response) =
    Logging.request ~tid ~con:("replay request:" ^ c.Connection.domstr) request;
    Logging.response ~tid
      ~con:("replay reply1: " ^ c.Connection.domstr)
      response;
    let response' = op_exn store c t request in
    Logging.response ~tid
      ~con:("replay reply2: " ^ c.Connection.domstr)
      response';
    Logging.response ~tid ~con response';
    if response <> response' then raise Transaction_again
  in
  try
    Logging.start_transaction ~con ~tid;
    List.iter perform_exn ops;
    Logging.end_transaction ~tid ~con;

    Transaction.commit ~con t
  with e ->
    error "transaction_replay caught: %s" (Printexc.to_string e);
    false

let reply_exn store c (request : t) : Response.payload =
  let tid = get_tid request in
  let t =
    if tid = Transaction.none then Transaction.make tid store
    else Connection.get_transaction c tid
  in
  let payload : Xs_protocol.Request.payload =
    match Xs_protocol.Request.parse (request : t) with
    | None ->
        error "Failed to parse request: got %s"
          (hexify (Bytes.to_string @@ Xs_protocol.to_bytes request));
        raise Parse_failure
    | Some x -> x
  in

  Logging.request ~tid ~con:c.Connection.domstr payload;

  let response_payload =
    match payload with
    | Request.Transaction_start ->
        if tid <> Transaction.none then raise Transaction_nested;
        let tid = Connection.register_transaction c store in
        Response.Transaction_start tid
    | Request.Transaction_end commit ->
        Connection.unregister_transaction c tid;
        if commit then (
          Logging.end_transaction ~tid ~con:c.Connection.domstr;
          if
            true
            && (not (Transaction.commit ~con:c.Connection.domstr t))
            && not (transaction_replay store c t)
          then raise Transaction_again;
          Transaction.get_paths t |> List.rev |> List.iter Connection.fire;
          Response.Transaction_end)
        else (* Don't log an explicit abort *)
          Response.Transaction_end
    | Request.Watch (path, token) ->
        let watch = Connection.add_watch c (Store.Name.of_string path) token in
        Connection.fire_one None watch;
        Response.Watch
    | Request.Unwatch (path, token) ->
        Connection.del_watch c (Store.Name.of_string path) token;
        Response.Unwatch
    | Request.Debug cmd ->
        Perms.has c.Connection.perm Perms.DEBUG;
        Response.Debug
          (try
             match cmd with
             | "print" :: msg :: _ ->
                 Logging.debug_print ~tid:0l ~con:c.Connection.domstr msg;
                 []
             | _ -> []
           with _ -> [])
    | Request.Introduce (domid, mfn, remote_port) ->
        Perms.has c.Connection.perm Perms.INTRODUCE;
        Introduce.(introduce { domid; mfn; remote_port });
        Connection.fire (Xs_protocol.Op.Write, Store.Name.introduceDomain);
        Response.Introduce
    | Request.Resume _ ->
        Perms.has c.Connection.perm Perms.RESUME;
        (* register domain *)
        Response.Resume
    | Request.Release _ ->
        Perms.has c.Connection.perm Perms.RELEASE;
        (* unregister domain *)
        Connection.fire (Xs_protocol.Op.Write, Store.Name.releaseDomain);
        Response.Release
    | Request.Set_target (mine, yours) ->
        Perms.has c.Connection.perm Perms.SET_TARGET;
        Hashtbl.iter
          (fun address c ->
            if Xs_protocol.domain_of_address address = mine then
              c.Connection.perm <- Perms.set_target c.Connection.perm yours)
          Connection.by_address;
        Response.Set_target
    | Request.Isintroduced _ ->
        Perms.has c.Connection.perm Perms.ISINTRODUCED;
        Response.Isintroduced false
    | Request.Error msg ->
        error "client sent us an error: %s" (hexify msg);
        raise Parse_failure
    | Request.Watchevent msg ->
        error "client sent us a watch event: %s" (hexify msg);
        raise Parse_failure
    | op ->
        let reply = op_exn store c t op in
        if tid <> Transaction.none then Transaction.add_operation t op reply;
        reply
  in

  if tid = Transaction.none then
    Transaction.get_paths t |> List.rev |> List.iter Connection.fire;
  response_payload

let gc store =
  if Symbol.created () > 1000 || Symbol.used () > 20000 then (
    debug "Started symbol GC";
    Symbol.mark_all_as_unused ();
    Store.mark_symbols store;
    Hashtbl.iter (fun _ c -> Connection.mark_symbols c) Connection.by_address;
    Symbol.garbage ())

let reply store c request =
  gc store;
  c.Connection.stat_nb_ops <- c.Connection.stat_nb_ops + 1;
  let tid = get_tid request in
  let rid = get_rid request in
  let response_payload, info =
    try (reply_exn store c request, None)
    with e -> (
      let default = Some (Printexc.to_string e) in
      let reply code = Response.Error code in
      match e with
      | Store.Already_exists p -> (reply "EEXIST", Some p)
      | Store.Path.Doesnt_exist p -> (reply "ENOENT", Some p)
      | Perms.Permission_denied -> (reply "EACCES", default)
      | Not_found -> (reply "ENOENT", default)
      | Parse_failure -> (reply "EINVAL", default)
      | Invalid_argument i -> (reply "EINVAL", Some i)
      | Transaction_again -> (reply "EAGAIN", default)
      | Transaction_nested -> (reply "EBUSY", default)
      | Quota.Limit_reached -> (reply "EQUOTA", default)
      | Quota.Data_too_big -> (reply "E2BIG", default)
      | Quota.Transaction_opened -> (reply "EQUOTA", default)
      | Failure _ -> (reply "EINVAL", default)
      | Namespace.Unsupported -> (reply "ENOTSUP", default)
      | _ -> (reply "EIO", default))
  in
  Logging.response ~tid ~con:c.Connection.domstr ?info response_payload;

  Response.print response_payload tid rid