CP-307933: avoid serializing/deserializing database fields in DB actions

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

Author: edwintorok

ocaml/database/database_server_main.ml

@@ -33,8 +33,9 @@ let remote_database_access_handler_v2 req bio =
     flush stdout ;
     raise e
 
+open Xapi_database
 module Local_tests =
-  Xapi_database.Database_test.Tests (Xapi_database.Db_cache_impl)
+  Database_test.Tests (Db_interface_compat.OfCached (Db_cache_impl))
 
 let schema = Test_schemas.schema
 

ocaml/database/database_test.ml

@@ -205,11 +205,13 @@ functor
       let db =
         db
         |> add_row "bar" "bar:1"
-             (Row.add 0L Db_names.ref (String "bar:1")
+             (Row.add 0L Db_names.ref
+                (Schema.Value.string "bar:1")
                 (Row.add 0L "foos" (Set []) Row.empty)
              )
         |> add_row "foo" "foo:1"
-             (Row.add 0L Db_names.ref (String "foo:1")
+             (Row.add 0L Db_names.ref
+                (Schema.Value.string "foo:1")
                 (Row.add 0L "bars" (Set []) Row.empty)
              )
         |> set_field "foo" "foo:1" "bars" (add_to_set "bar:1" (Set []))
@@ -219,7 +221,7 @@ functor
         Table.find "bar:1" (TableSet.find "bar" (Database.tableset db))
       in
       let bar_foos = Row.find "foos" bar_1 in
-      if bar_foos <> Set ["foo:1"] then
+      if bar_foos <> Schema.Value.set ["foo:1"] then
         failwith_fmt
           "check_many_to_many: bar(bar:1).foos expected ('foo:1') got %s"
           (Schema.Value.marshal bar_foos) ;
@@ -235,13 +237,13 @@ functor
         failwith_fmt "check_many_to_many: bar(bar:1).foos expected () got %s"
           (Schema.Value.marshal bar_foos) ;
       (* add 'bar' to foo.bars *)
-      let db = set_field "foo" "foo:1" "bars" (Set ["bar:1"]) db in
+      let db = set_field "foo" "foo:1" "bars" (Schema.Value.set ["bar:1"]) db in
       (* check that 'bar.foos' includes 'foo' *)
       let bar_1 =
         Table.find "bar:1" (TableSet.find "bar" (Database.tableset db))
       in
       let bar_foos = Row.find "foos" bar_1 in
-      if bar_foos <> Set ["foo:1"] then
+      if bar_foos <> Schema.Value.set ["foo:1"] then
         failwith_fmt
           "check_many_to_many: bar(bar:1).foos expected ('foo:1') got %s - 2"
           (Schema.Value.marshal bar_foos) ;
@@ -269,8 +271,9 @@ functor
                     let row = Db_cache_types.Table.find r table in
                     let s =
                       Db_cache_types.Row.fold_over_recent g
-                        (fun k _ (_, cached) acc ->
-                          Printf.sprintf "%s %s=%s" acc k cached
+                        (fun k _ cached acc ->
+                          Printf.sprintf "%s %s=%s" acc k
+                            (Schema.CachedValue.string_of cached)
                         )
                         row ""
                     in

ocaml/database/db_backend.ml

@@ -21,11 +21,11 @@ let db_FLUSH_TIMER = 2.0
 
 (* --------------------- Util functions on db datastructures *)
 
-let master_database = ref (Db_cache_types.Database.make Schema.empty)
+let master_database = Atomic.make (Db_cache_types.Database.make Schema.empty)
 
-let __test_set_master_database db = master_database := db
+let __test_set_master_database db = Atomic.set master_database db
 
-let make () = Db_ref.in_memory (ref master_database)
+let make () = Db_ref.in_memory master_database
 
 (* !!! Right now this is called at cache population time. It would probably be preferable to call it on flush time instead, so we
    don't waste writes storing non-persistent field values on disk.. At the moment there's not much to worry about, since there are
@@ -38,12 +38,15 @@ let blow_away_non_persistent_fields (schema : Schema.t) db =
   (* Generate a new row given a table schema *)
   let row schema row : Row.t * int64 =
     Row.fold
-      (fun name {Stat.created; modified; _} (v, _) (acc, max_upd) ->
+      (fun name {Stat.created; modified; _} v (acc, max_upd) ->
         try
           let col = Schema.Table.find name schema in
           let empty = col.Schema.Column.empty in
           let v', modified' =
-            if col.Schema.Column.persistent then (v, modified) else (empty, g)
+            if col.Schema.Column.persistent then
+              (Schema.CachedValue.value_of v, modified)
+            else
+              (empty, g)
           in
           ( Row.update modified' name empty
               (fun _ -> v')

ocaml/database/db_cache.ml

@@ -19,30 +19,32 @@ module D = Debug.Make (struct let name = "db_cache" end)
 open D
 
 (** Masters will use this to modify the in-memory cache directly *)
-module Local_db : DB_ACCESS = Db_cache_impl
+module Local_db : DB_ACCESS2 = Db_cache_impl
 
 (** Slaves will use this to call the master by XMLRPC *)
-module Remote_db : DB_ACCESS = Db_rpc_client_v1.Make (struct
+module Remote_db : DB_ACCESS2 =
+Db_interface_compat.OfCompat (Db_rpc_client_v1.Make (struct
   let initialise () =
     ignore (Master_connection.start_master_connection_watchdog ()) ;
     ignore (Master_connection.open_secure_connection ())
 
   let rpc request = Master_connection.execute_remote_fn request
-end)
+end))
 
 let get = function
   | Db_ref.In_memory _ ->
-      (module Local_db : DB_ACCESS)
+      (module Local_db : DB_ACCESS2)
   | Db_ref.Remote ->
-      (module Remote_db : DB_ACCESS)
+      (module Remote_db : DB_ACCESS2)
 
 let lifecycle_state_of ~obj fld =
   let open Datamodel in
   let {fld_states; _} = StringMap.find obj all_lifecycles in
   StringMap.find fld fld_states
 
+module DB = Db_interface_compat.OfCached (Local_db)
+
 let apply_delta_to_cache entry db_ref =
-  let module DB : DB_ACCESS = Local_db in
   match entry with
   | Redo_log.CreateRow (tblname, objref, kvs) ->
       debug "Redoing create_row %s (%s)" tblname objref ;

ocaml/database/db_cache.mli

@@ -12,6 +12,6 @@
  * GNU Lesser General Public License for more details.
  *)
 
-val get : Db_ref.t -> (module Db_interface.DB_ACCESS)
+val get : Db_ref.t -> (module Db_interface.DB_ACCESS2)
 
 val apply_delta_to_cache : Redo_log.t -> Db_ref.t -> unit