Source file transaction.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.
 *)

let debug fmt = Logging.debug "transaction" fmt

open Junk

let none = 0l
let test_eagain = ref false

let check_parents_perms_identical root1 root2 path =
  let hierarch = Store.Path.get_hierarchy path in
  let permdiff =
    List.fold_left
      (fun acc path ->
        let n1 = Store.lookup root1 path and n2 = Store.lookup root2 path in
        match (n1, n2) with
        | Some n1, Some n2 ->
            Store.Node.get_perms n1 <> Store.Node.get_perms n2 || acc
        | _ -> true || acc)
      false hierarch
  in
  not permdiff

let get_lowest path1 path2 =
  match path2 with
  | None -> Some path1
  | Some path2 -> Some (Store.Path.get_common_prefix path1 path2)

let test_coalesce oldroot currentroot path =
  let oldnode = Store.lookup oldroot path
  and currentnode = Store.lookup currentroot path in

  match (oldnode, currentnode) with
  | Some oldnode, Some currentnode ->
      if oldnode == currentnode then
        check_parents_perms_identical oldroot currentroot path
      else false
  | None, None -> (
      (* ok then it doesn't exists in the old version and the current version,
         			   just sneak it in as a child of the parent node if it exists, or else fail *)
      let pnode = Store.lookup currentroot (Store.Path.get_parent path) in
      match pnode with
      | None -> false (* ok it doesn't exists, just bail out. *)
      | Some _pnode -> true)
  | _ -> false

let can_coalesce oldroot currentroot path =
  try test_coalesce oldroot currentroot path with _ -> false

type ty = No | Full of (int32 * Store.Node.t * Store.t)

type t = {
    ty : ty
  ; store : Store.t
  ; quota : Quota.t
  ; mutable paths : (Xs_protocol.Op.t * Store.Name.t) list
  ; mutable operations :
      (Xs_protocol.Request.payload * Xs_protocol.Response.payload) list
  ; mutable read_lowpath : Store.Path.t option
  ; mutable write_lowpath : Store.Path.t option
}

let make id store =
  let ty = if id = none then No else Full (id, store.Store.root, store) in
  {
    ty
  ; store = (if id = none then store else Store.copy store)
  ; quota = Quota.copy store.Store.quota
  ; paths = []
  ; operations = []
  ; read_lowpath = None
  ; write_lowpath = None
  }

let get_id t = match t.ty with No -> none | Full (id, _, _) -> id
let get_store t = t.store
let get_paths t = t.paths
let add_wop t ty path = t.paths <- (ty, Store.Path.to_name path) :: t.paths

let add_operation t request response =
  t.operations <- (request, response) :: t.operations

let get_operations t = List.rev t.operations
let set_read_lowpath t path = t.read_lowpath <- get_lowest path t.read_lowpath

let set_write_lowpath t path =
  t.write_lowpath <- get_lowest path t.write_lowpath

let exists t _perms path = Store.exists t.store path

let write t creator perm path value =
  let path_existed = exists t perm path in
  Store.write t.store creator perm path value;
  if path_existed then set_write_lowpath t path
  else set_write_lowpath t (Store.Path.get_parent path);
  add_wop t Xs_protocol.Op.Write path

let mkdir ?(with_watch = true) t creator perm path =
  Store.mkdir t.store creator perm path;
  set_write_lowpath t path;
  if with_watch then add_wop t Xs_protocol.Op.Mkdir path

let setperms t perm path perms =
  Store.setperms t.store perm path perms;
  set_write_lowpath t path;
  add_wop t Xs_protocol.Op.Setperms path

let rm t perm path =
  Store.rm t.store perm path;
  set_write_lowpath t (Store.Path.get_parent path);
  add_wop t Xs_protocol.Op.Rm path

let list t perm path =
  let r = Store.ls t.store perm path in
  set_read_lowpath t path;
  r

let read t perm path =
  let r = Store.read t.store perm path in
  set_read_lowpath t path;
  r

let getperms t perm path =
  let r = Store.getperms t.store perm path in
  set_read_lowpath t path;
  r

let commit ~con t =
  let has_write_ops = List.length t.paths > 0 in
  let has_coalesced = ref false in
  let has_commited =
    match t.ty with
    | No -> true
    | Full (_id, oldroot, cstore) ->
        let commit_partial oldroot cstore store =
          (* get the lowest path of the query and verify that it hasn't
             			   been modified by others transactions. *)
          let readpath_ok =
            match t.read_lowpath with
            | None -> true (* no reads recorded *)
            | Some path -> can_coalesce oldroot cstore.Store.root path
          in
          let writepath_ok =
            match t.write_lowpath with
            | None -> true (* no writes recorded *)
            | Some path -> can_coalesce oldroot cstore.Store.root path
          in
          if readpath_ok && writepath_ok then (
            maybe
              (fun p ->
                let n = Store.lookup store.Store.root p in

                (* it has to be in the store, otherwise it means bugs
                   					   in the lowpath registration. we don't need to handle none. *)
                maybe
                  (fun n -> Store.set_node cstore p n t.quota store.Store.quota)
                  n;
                Logging.write_coalesce ~tid:(get_id t) ~con
                  (Store.Path.to_string p))
              t.write_lowpath;
            maybe
              (fun p ->
                Logging.read_coalesce ~tid:(get_id t) ~con
                  (Store.Path.to_string p))
              t.read_lowpath;
            has_coalesced := true;
            cstore.Store.stat_transaction_coalesce <-
              cstore.Store.stat_transaction_coalesce + 1;
            true)
          else (
            (* cannot do anything simple, just discard the queries,
               				   and the client need to redo it later *)
            cstore.Store.stat_transaction_abort <-
              cstore.Store.stat_transaction_abort + 1;
            false)
        in
        let try_commit oldroot cstore store =
          if oldroot == cstore.Store.root then (
            (* move the new root to the current store, if the oldroot
               				   has not been modified *)
            if has_write_ops then (
              Store.set_root cstore store.Store.root;
              Store.set_quota cstore store.Store.quota);
            true)
          else
            (* we try a partial commit if possible *)
            commit_partial oldroot cstore store
        in
        if !test_eagain && Random.int 3 = 0 then false
        else try_commit oldroot cstore t.store
  in
  (*
	if has_commited && has_write_ops then
		Disk.write t.store;
*)
  if not has_commited then Logging.conflict ~tid:(get_id t) ~con
  else if not !has_coalesced then Logging.commit ~tid:(get_id t) ~con;
  has_commited