CP-307958: Replace mutexes with Atomic where possible

Atomic has lower overhead, and can be used for simple situations (incrementing an integer, setting a boolean).

```
Mutex+incr (ns):
 { monotonic-clock per run = 32.272091 (confidence: 32.365531 to 32.170138);
   r² = Some 0.999441 }
Atomic.incr (ns):
 { monotonic-clock per run = 2.938678 (confidence: 2.944486 to 2.933688);
   r² = Some 0.999857 }
```


No functional change

Signed-off-by: Edwin Török <[email protected]>

Author: edwintorok

ocaml/database/db_connections.ml

@@ -62,22 +62,12 @@ let preferred_write_db () = List.hd (Db_conn_store.read_db_connections ())
 let exit_on_next_flush = ref false
 
 (* db flushing thread refcount: the last thread out of the door does the exit(0) when flush_on_exit is true *)
-let with_lock = Xapi_stdext_threads.Threadext.Mutex.execute
+let db_flush_thread_refcount = Atomic.make 0
 
-let db_flush_thread_refcount_m = Mutex.create ()
-
-let db_flush_thread_refcount = ref 0
-
-let inc_db_flush_thread_refcount () =
-  with_lock db_flush_thread_refcount_m (fun () ->
-      db_flush_thread_refcount := !db_flush_thread_refcount + 1
-  )
+let inc_db_flush_thread_refcount () = Atomic.incr db_flush_thread_refcount
 
 let dec_and_read_db_flush_thread_refcount () =
-  with_lock db_flush_thread_refcount_m (fun () ->
-      db_flush_thread_refcount := !db_flush_thread_refcount - 1 ;
-      !db_flush_thread_refcount
-  )
+  Atomic.fetch_and_add db_flush_thread_refcount (-1)
 
 let pre_exit_hook () =
   (* We're about to exit. Close the active redo logs. *)

ocaml/database/db_remote_cache_access_v1.ml

@@ -6,9 +6,7 @@ module DBCacheRemoteListener = struct
 
   exception DBCacheListenerUnknownMessageName of string
 
-  let ctr_mutex = Mutex.create ()
-
-  let calls_processed = ref 0
+  let calls_processed = Atomic.make 0
 
   let success xml =
     let resp = XMLRPC.To.array [XMLRPC.To.string "success"; xml] in
@@ -34,8 +32,7 @@ module DBCacheRemoteListener = struct
       		Note that, although the messages still contain the pool_secret for historical reasons,
       		access has already been applied by the RBAC code in Xapi_http.add_handler. *)
   let process_xmlrpc xml =
-    let with_lock = Xapi_stdext_threads.Threadext.Mutex.execute in
-    with_lock ctr_mutex (fun () -> calls_processed := !calls_processed + 1) ;
+    Atomic.incr calls_processed ;
     let fn_name, args =
       match XMLRPC.From.array (fun x -> x) xml with
       | [fn_name; _; args] ->

ocaml/database/redo_log.ml

@@ -77,8 +77,7 @@ type redo_log_conf = {
   ; backoff_delay: int ref
   ; sock: Unix.file_descr option ref
   ; pid: (Forkhelpers.pidty * string * string) option ref
-  ; dying_processes_mutex: Mutex.t
-  ; num_dying_processes: int ref
+  ; num_dying_processes: int Atomic.t
   ; mutex: Mutex.t  (** exclusive access to this configuration *)
 }
 
@@ -585,14 +584,10 @@ let shutdown log =
             (Thread.create
                (fun () ->
                  D.debug "Waiting for I/O process with pid %d to die..." ipid ;
-                 with_lock log.dying_processes_mutex (fun () ->
-                     log.num_dying_processes := !(log.num_dying_processes) + 1
-                 ) ;
+                 Atomic.incr log.num_dying_processes ;
                  ignore (Forkhelpers.waitpid p) ;
                  D.debug "Finished waiting for process with pid %d" ipid ;
-                 with_lock log.dying_processes_mutex (fun () ->
-                     log.num_dying_processes := !(log.num_dying_processes) - 1
-                 )
+                 Atomic.decr log.num_dying_processes
                )
                ()
             ) ;
@@ -633,13 +628,11 @@ let startup log =
           () (* We're already started *)
       | None -> (
           (* Don't start if there are already some processes hanging around *)
-          with_lock log.dying_processes_mutex (fun () ->
-              if
-                !(log.num_dying_processes)
-                >= Db_globs.redo_log_max_dying_processes
-              then
-                raise TooManyProcesses
-          ) ;
+          if
+            Atomic.get log.num_dying_processes
+            >= Db_globs.redo_log_max_dying_processes
+          then
+            raise TooManyProcesses ;
           match !(log.device) with
           | None ->
               D.info "Could not find block device" ;
@@ -793,8 +786,7 @@ let create ~name ~state_change_callback ~read_only =
     ; backoff_delay= ref Db_globs.redo_log_initial_backoff_delay
     ; sock= ref None
     ; pid= ref None
-    ; dying_processes_mutex= Mutex.create ()
-    ; num_dying_processes= ref 0
+    ; num_dying_processes= Atomic.make 0
     ; mutex= Mutex.create ()
     }
   in