oxenstored: handling of domain conflict-credit

This commit gives each domain a conflict-credit variable, which will
later be used for limiting how often a domain can cause other domain's
transaction-commits to fail.

This commit also provides functions and data for manipulating domains
and their conflict-credit, and checking whether they have credit.

Reported-by: Juergen Gross <jgross@suse.com>
Signed-off-by: Thomas Sanders <thomas.sanders@citrix.com>
Reviewed-by: Jonathan Davies <jonathan.davies@citrix.com>
Reviewed-by: Christian Lindig <christian.lindig@citrix.com>

tools/ocaml/xenstored/connection.ml

@@ -258,3 +258,8 @@ let debug con =
 	let domid = get_domstr con in
 	let watches = List.map (fun (path, token) -> Printf.sprintf "watch %s: %s %s\n" domid path token) (list_watches con) in
 	String.concat "" watches
+
+let decr_conflict_credit doms con =
+	match con.dom with
+	| None -> () (* It's a socket connection. We don't know which domain we're in, so treat it as if it's free to conflict *)
+	| Some dom -> Domains.decr_conflict_credit doms dom

tools/ocaml/xenstored/define.ml

@@ -29,6 +29,9 @@ let maxwatch = ref (50)
 let maxtransaction = ref (20)
 let maxrequests = ref (-1)   (* maximum requests per transaction *)
 
+let conflict_burst_limit = ref 5.0
+let conflict_rate_limit_is_aggregate = ref true
+
 let domid_self = 0x7FF0
 
 exception Not_a_directory of string

tools/ocaml/xenstored/domain.ml

@@ -31,8 +31,12 @@ type t =
 	mutable io_credit: int; (* the rounds of ring process left to do, default is 0,
 	                           usually set to 1 when there is work detected, could
 	                           also set to n to give "lazy" clients extra credit *)
+	mutable conflict_credit: float; (* Must be positive to perform writes; a commit
+	                                   that later causes conflict with another
+	                                   domain's transaction costs credit. *)
 }
 
+let is_dom0 d = d.id = 0
 let get_path dom = "/local/domain/" ^ (sprintf "%u" dom.id)
 let get_id domain = domain.id
 let get_interface d = d.interface
@@ -48,6 +52,10 @@ let set_io_credit ?(n=1) domain = domain.io_credit <- max 0 n
 let incr_io_credit domain = domain.io_credit <- domain.io_credit + 1
 let decr_io_credit domain = domain.io_credit <- max 0 (domain.io_credit - 1)
 
+let is_paused_for_conflict dom = dom.conflict_credit <= 0.0
+
+let is_free_to_conflict = is_dom0
+
 let string_of_port = function
 | None -> "None"
 | Some x -> string_of_int (Xeneventchn.to_int x)
@@ -84,6 +92,5 @@ let make id mfn remote_port interface eventchn = {
 	port = None;
 	bad_client = false;
 	io_credit = 0;
+	conflict_credit = !Define.conflict_burst_limit;
 }
-
-let is_dom0 d = d.id = 0

tools/ocaml/xenstored/domains.ml

@@ -15,20 +15,58 @@
  *)
 
 let debug fmt = Logging.debug "domains" fmt
+let error fmt = Logging.error "domains" fmt
+let warn fmt  = Logging.warn  "domains" fmt
 
 type domains = {
 	eventchn: Event.t;
 	table: (Xenctrl.domid, Domain.t) Hashtbl.t;
+
+	(* N.B. the Queue module is not thread-safe but oxenstored is single-threaded. *)
+	(* Domains queue up to regain conflict-credit; we have a queue for
+	   domains that are carrying some penalty and so are below the
+	   maximum credit, and another queue for domains that have run out of
+	   credit and so have had their access paused. *)
+	doms_conflict_paused: (Domain.t option ref) Queue.t;
+	doms_with_conflict_penalty: (Domain.t option ref) Queue.t;
+
+	(* A callback function to be called when we go from zero to one paused domain.
+	   This will be to reset the countdown until the next unit of credit is issued. *)
+	on_first_conflict_pause: unit -> unit;
+
+	(* If config is set to use individual instead of aggregate conflict-rate-limiting,
+	   we use this instead of the queues. *)
+	mutable n_paused: int;
 }
 
-let init eventchn =
-	{ eventchn = eventchn; table = Hashtbl.create 10 }
+let init eventchn = {
+	eventchn = eventchn;
+	table = Hashtbl.create 10;
+	doms_conflict_paused = Queue.create ();
+	doms_with_conflict_penalty = Queue.create ();
+	on_first_conflict_pause = (fun () -> ()); (* Dummy value for now, pending subsequent commit. *)
+	n_paused = 0;
+}
 let del doms id = Hashtbl.remove doms.table id
 let exist doms id = Hashtbl.mem doms.table id
 let find doms id = Hashtbl.find doms.table id
 let number doms = Hashtbl.length doms.table
 let iter doms fct = Hashtbl.iter (fun _ b -> fct b) doms.table
 
+(* Functions to handle queues of domains given that the domain might be deleted while in a queue. *)
+let push dom queue =
+	Queue.push (ref (Some dom)) queue
+
+let rec pop queue =
+	match !(Queue.pop queue) with
+	| None -> pop queue
+	| Some x -> x
+
+let remove_from_queue dom queue =
+	Queue.iter (fun d -> match !d with
+		| None -> ()
+		| Some x -> if x=dom then d := None) queue
+
 let cleanup xc doms =
 	let notify = ref false in
 	let dead_dom = ref [] in
@@ -52,6 +90,11 @@ let cleanup xc doms =
 		let dom = Hashtbl.find doms.table id in
 		Domain.close dom;
 		Hashtbl.remove doms.table id;
+		if dom.Domain.conflict_credit <= !Define.conflict_burst_limit
+		then (
+			remove_from_queue dom doms.doms_with_conflict_penalty;
+			if (dom.Domain.conflict_credit <= 0.) then remove_from_queue dom doms.doms_conflict_paused
+		)
 	) !dead_dom;
 	!notify, !dead_dom
 
@@ -84,3 +127,59 @@ let create0 fake doms =
 	Domain.bind_interdomain dom;
 	Domain.notify dom;
 	dom
+
+let decr_conflict_credit doms dom =
+	let before = dom.Domain.conflict_credit in
+	let after = max (-1.0) (before -. 1.0) in
+	dom.Domain.conflict_credit <- after;
+	if !Define.conflict_rate_limit_is_aggregate then (
+		if before >= !Define.conflict_burst_limit
+		&& after < !Define.conflict_burst_limit
+		&& after > 0.0
+		then (
+			push dom doms.doms_with_conflict_penalty
+		) else if before > 0.0 && after <= 0.0
+		then (
+			let first_pause = Queue.is_empty doms.doms_conflict_paused in
+			push dom doms.doms_conflict_paused;
+			if first_pause then doms.on_first_conflict_pause ()
+		) else (
+			(* The queues are correct already: no further action needed. *)
+		)
+	) else if before > 0.0 && after <= 0.0 then (
+		doms.n_paused <- doms.n_paused + 1;
+		if doms.n_paused = 1 then doms.on_first_conflict_pause ()
+	)
+
+(* Give one point of credit to one domain, and update the queues appropriately. *)
+let incr_conflict_credit_from_queue doms =
+	let process_queue q requeue_test =
+		let d = pop q in
+		d.Domain.conflict_credit <- min (d.Domain.conflict_credit +. 1.0) !Define.conflict_burst_limit;
+		if requeue_test d.Domain.conflict_credit then (
+			push d q (* Make it queue up again for its next point of credit. *)
+		)
+	in
+	let paused_queue_test cred = cred <= 0.0 in
+	let penalty_queue_test cred = cred < !Define.conflict_burst_limit in
+	try process_queue doms.doms_conflict_paused paused_queue_test
+	with Queue.Empty -> (
+		try process_queue doms.doms_with_conflict_penalty penalty_queue_test
+		with Queue.Empty -> () (* Both queues are empty: nothing to do here. *)
+	)
+
+let incr_conflict_credit doms =
+	if !Define.conflict_rate_limit_is_aggregate
+	then incr_conflict_credit_from_queue doms
+	else (
+		(* Give a point of credit to every domain, subject only to the cap. *)
+		let inc dom =
+			let before = dom.Domain.conflict_credit in
+			let after = min (before +. 1.0) !Define.conflict_burst_limit in
+			dom.Domain.conflict_credit <- after;
+			if before <= 0.0 && after > 0.0
+			then doms.n_paused <- doms.n_paused - 1
+		in
+		(* Scope for optimisation (probably tiny): avoid iteration if all domains are at max credit *)
+		iter doms inc
+	)

tools/ocaml/xenstored/oxenstored.conf

@@ -9,6 +9,38 @@ test-eagain = false
 # Activate transaction merge support
 merge-activate = true
 
+# Limits applied to domains whose writes cause other domains' transaction
+# commits to fail. Must include decimal point.
+
+# The burst limit is the number of conflicts a domain can cause to
+# fail in a short period; this value is used for both the initial and
+# the maximum value of each domain's conflict-credit, which falls by
+# one point for each conflict caused, and when it reaches zero the
+# domain's requests are ignored.
+conflict-burst-limit = 5.0
+
+# The conflict-credit is replenished over time:
+# one point is issued after each conflict-max-history-seconds, so this
+# is the minimum pause-time during which a domain will be ignored.
+# conflict-max-history-seconds = 0.05
+
+# If the conflict-rate-limit-is-aggregate flag is true then after each
+# tick one point of conflict-credit is given to just one domain: the
+# one at the front of the queue. If false, then after each tick each
+# domain gets a point of conflict-credit.
+# 
+# In environments where it is known that every transaction will
+# involve a set of nodes that is writable by at most one other domain,
+# then it is safe to set this aggregate-limit flag to false for better
+# performance. (This can be determined by considering the layout of
+# the xenstore tree and permissions, together with the content of the
+# transactions that require protection.)
+# 
+# A transaction which involves a set of nodes which can be modified by
+# multiple other domains can suffer conflicts caused by any of those
+# domains, so the flag must be set to true.
+conflict-rate-limit-is-aggregate = true
+
 # Activate node permission system
 perms-activate = true
 

tools/ocaml/xenstored/transaction.ml

@@ -14,6 +14,8 @@
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU Lesser General Public License for more details.
  *)
+let error fmt = Logging.error "transaction" fmt
+
 open Stdext
 
 let none = 0

tools/ocaml/xenstored/xenstored.ml

@@ -89,6 +89,8 @@ let parse_config filename =
 	let pidfile = ref default_pidfile in
 	let options = [
 		("merge-activate", Config.Set_bool Transaction.do_coalesce);
+		("conflict-burst-limit", Config.Set_float Define.conflict_burst_limit);
+		("conflict-rate-limit-is-aggregate", Config.Set_bool Define.conflict_rate_limit_is_aggregate);
 		("perms-activate", Config.Set_bool Perms.activate);
 		("quota-activate", Config.Set_bool Quota.activate);
 		("quota-maxwatch", Config.Set_int Define.maxwatch);