Convert queries to asynchronous queries

Using the ppx syntax extension:
`let%map y = f x in ...` is equivalent to `f x >>| fun y -> ..`
`let%bind y = f x in ...` is equivalent to `f x >>= fun y -> ...`

Except these also allow you to do parallel binds, which is tedious to
accomplish "manually".

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

src/create_tiny_url_handler.ml

@@ -1,11 +1,12 @@
 open Core
+open Async
 
 let t ~args = object (self)
   inherit Json_handler.t ~args
 
   method private write_body =
     let url = self#get_param_exn "url" in
     let tuples = [("url", url)] in
-    let id = Sql.ensure_inserted_get_id ~conn ~tbl:"tiny_urls" ~tuples in
+    let%map id = Postgresql_async.wrap_sql ~conn (Sql.ensure_inserted_get_id ~tbl:"tiny_urls" ~tuples) in
     printf "{\"id\":%d}" id
 end

src/default_handler.ml

@@ -1,4 +1,5 @@
 open Core
+open Async
 
 let t ~args = object (self)
   inherit Html_handler.t ~args
@@ -8,4 +9,5 @@ let t ~args = object (self)
     printf "<li><a href='?p=soms_by_tc'>Scales of Measure</a></li>\n";
     printf "<li><a href='?p=import_page'>Import Jobs</a></li>\n";
     printf "</ul>\n";
+    return ()
 end

src/handler.ml

@@ -1,8 +1,9 @@
 open Core
+open Async
 open Utils
 
 type args = {
-  conn : Postgresql.connection;
+  conn : Postgresql_async.t;
   params : (string * string) list;
 }
 
@@ -19,13 +20,13 @@ object (self)
 
   method private write_header = ()
 
-  method private write_body = ()
+  method private write_body = return ()
 
   method private write_footer = ()
 
   method handle =
     self#write_header;
-    self#write_body;
+    let%map () = self#write_body in
     self#write_footer
 
   method private write_html_header =

src/import_jobs_handler.ml

@@ -41,6 +41,7 @@ let t ~args = object (self)
     Printf.printf "<pre>";
     import_job job_ids;
     Printf.printf "</pre>";
-    Printf.printf "Finished."
+    Printf.printf "Finished.";
+    Async.return ()
 
 end

src/import_page_handler.ml

@@ -1,4 +1,5 @@
 open Core
+open Async
 
 let t ~args = object (self)
   inherit Html_handler.t ~args
@@ -14,5 +15,6 @@ let t ~args = object (self)
     printf "<input type='submit' name='submit' value='Import now'/>\n";
     printf "</form>";
     printf "</div>";
+    return ()
 
 end

src/javascript_only_handler.ml

@@ -1,5 +1,5 @@
 let t ~args = object (self)
   inherit Html_handler.t ~args
 
-  method private write_body = Printf.printf "<script src='rage.js'></script>"
+  method private write_body = Printf.printf "<script src='rage.js'></script>"; Async.return ()
 end

src/postgresql_async.ml

@@ -12,11 +12,6 @@ let connect ~conninfo =
 let close c =
   in_thread ~name:"Postgresql close connection" (fun () -> c#finish)
 
-let exec_exn ~(conn : Postgresql.connection) ~query =
-  in_thread ~name:"Postgresql query" (fun () ->
-      (* previous invocation might've left the connection in a bad state *)
-      conn#try_reset ; Sql.exec_exn ~conn ~query)
-
 module Lazy_pooled_resource = struct
   type 'a t = 'a Or_error.t Lazy_deferred.t Throttle.t
 
@@ -60,9 +55,15 @@ let connect_pool ~conninfo =
 
 let destroy_pool = Lazy_pooled_resource.destroy
 
+let wrap_sql ~(conn:t) f =
+  Lazy_pooled_resource.with_ conn ~f:(fun conn ->
+    in_thread ~name:"Postgresql query" (fun () ->
+      (* previous invocation might've left the connection in a bad state *)
+      conn#try_reset ; f ~conn))
+        |> Deferred.Or_error.ok_exn
+
 let exec_exn ~conn ~query =
-  Lazy_pooled_resource.with_ conn ~f:(fun conn -> exec_exn ~conn ~query)
-  |> Deferred.Or_error.ok_exn
+  wrap_sql ~conn (Sql.exec_exn ~query)
 
 let exec_exn_get_all ~conn ~query =
   exec_exn ~conn ~query >>| fun r -> r#get_all

src/postgresql_async.mli

@@ -4,3 +4,4 @@ val connect_pool: conninfo:string -> t
 val destroy_pool: t -> unit Deferred.t
 val exec_exn: conn:t -> query:string -> Postgresql.result Deferred.t
 val exec_exn_get_all: conn:t -> query:string -> string array array Deferred.t
+val wrap_sql: conn:t -> (conn:Postgresql.connection -> 'a) -> 'a Deferred.t

src/redirect_tiny_url_handler.ml

@@ -1,12 +1,13 @@
 open Core
+open Async
 
 let t ~args = object (self)
   inherit Html_handler.t ~args
 
   method handle =
     let id = int_of_string (self#get_param_exn "t") in
     let query = sprintf "SELECT url FROM tiny_urls WHERE key=%d" id in
-    let result = Sql.exec_exn ~conn ~query in
+    let%map result = Postgresql_async.exec_exn ~conn ~query in
     match result#ntuples with
     | 1 -> self#javascript_redirect (Sql.get_first_entry_exn ~result)
     | _ -> self#write_404

src/som_data_handler.ml

@@ -1,4 +1,5 @@
 open Core
+open Async
 open Fn
 open Utils
 
@@ -81,14 +82,14 @@ let t ~args = object (self)
 
   method private write_body =
     let som_id = int_of_string (self#get_param_exn "id") in
-    let tc_fqn, tc_config_tbl = get_tc_config_tbl_name conn som_id in
-    let som_config_tbl, som_tbl_exists = som_config_tbl_exists ~conn som_id in
+    let%bind tc_fqn, tc_config_tbl = get_tc_config_tbl_name conn som_id in
+    let%bind som_config_tbl, som_tbl_exists = som_config_tbl_exists ~conn som_id in
     (* determine filter columns and their types *)
     let tbls = ["measurements_2"; "soms_jobs"; "jobs"; "builds"; "tc_config"; "machines";
       tc_config_tbl] @
       (if som_tbl_exists then [som_config_tbl] else []) in
-    let col_fqns = get_column_fqns_many conn tbls in
-    let col_types = get_column_types_many conn tbls in
+    let%bind col_fqns = get_column_fqns_many conn tbls in
+    let%bind col_types = get_column_types_many conn tbls in
     (* Get axes selections. xaxis may be multi-valued; yaxis is single value. *)
     let xaxis = self#values_for_key "xaxis" ~default:["branch"] in
     (* xaxis could be ["one"; "two"] or ["one%2Ctwo"] -- both are equivalent *)
@@ -114,7 +115,7 @@ let t ~args = object (self)
     let filter = extract_filter col_fqns col_types params values_prefix in
     (* obtain SOM meta-data *)
     let query = sprintf "SELECT positive FROM soms WHERE som_id=%d" som_id in
-    let metadata = Sql.exec_exn ~conn ~query in
+    let%bind metadata = Postgresql_async.exec_exn ~conn ~query in
     let positive = String.(Sql.get_first_entry_exn ~result:metadata = "t") in
     (* obtain data from database *)
     let query =
@@ -140,7 +141,7 @@ let t ~args = object (self)
       (if not (String.is_empty filter) then sprintf " AND %s" filter else "") ^
       (sprintf " LIMIT %d" limit_rows)
     in
-    let data = Sql.exec_exn ~conn ~query in
+    let%bind data = Postgresql_async.exec_exn ~conn ~query in
     let rows = data#get_all in
     debug (sprintf "The query returned %d rows" (Array.length rows));
     (if Array.length rows = limit_rows then debug (sprintf "WARNING: truncation of data -- we are only returning the first %d rows" limit_rows));
@@ -200,5 +201,6 @@ let t ~args = object (self)
       printf "]}"
     in
     List.iteri (Hashtbl.Poly.to_alist all_series) ~f:process_series;
-    printf "]}"
+    printf "]}";
+    return ()
 end

src/som_page_handler.ml

@@ -1,4 +1,5 @@
 open Core
+open Async
 open Utils
 
 let jira_hostname = "jira.uk.xensource.com"
@@ -42,9 +43,9 @@ let t ~args = object (self)
 
     let machine_options_lst = options_lst_of_dbresult machines in
 
-    let config_options_lst = List.map config_column_names ~f:(fun config_name ->
+    let%map config_options_lst = Deferred.List.map config_column_names ~f:(fun config_name ->
       let query = sprintf "SELECT DISTINCT %s FROM %s ORDER BY %s" config_name tc_config_tbl config_name in
-      let configs = Sql.exec_exn ~conn ~query in
+      let%map configs = Postgresql_async.exec_exn ~conn ~query in
       get_options_for_field_once configs 0
     ) in
 
@@ -75,39 +76,41 @@ let t ~args = object (self)
 
   method private write_body =
     let som_id = int_of_string (List.Assoc.find_exn ~equal:String.equal params "som") in
-    let _, tc_config_tbl = get_tc_config_tbl_name conn som_id in
+    let%bind _, tc_config_tbl = get_tc_config_tbl_name conn som_id in
     let query =
       sprintf "SELECT * FROM soms WHERE som_id=%d" som_id in
-    let som_info = Sql.exec_exn ~conn ~query in
+    let%bind som_info =
+      Postgresql_async.exec_exn ~conn ~query in
     let query = "SELECT * FROM " ^ tc_config_tbl ^ " LIMIT 0" in
-    let config_columns = Sql.exec_exn ~conn ~query in
+    let%bind config_columns = Postgresql_async.exec_exn ~conn ~query in
     let job_fields = String.concat ~sep:", " Utils.job_fields in
     let query = "SELECT DISTINCT " ^ job_fields ^ " FROM soms_jobs WHERE " ^
       (sprintf "som_id=%d" som_id) in
-    let job_ids = Sql.exec_exn ~conn ~query in
+    let%bind job_ids = Postgresql_async.exec_exn ~conn ~query in
     let build_fields = String.concat ~sep:", " Utils.build_fields in
     let query =
       "SELECT DISTINCT " ^ build_fields ^ " " ^
       (sprintf "FROM builds AS b, jobs AS j, (select distinct job_id from soms_jobs where som_id=%d) AS m " som_id) ^
       "WHERE m.job_id=j.job_id AND j.build_id=b.build_id "
     in
-    let builds = Sql.exec_exn ~conn ~query in
+    let%bind builds = Postgresql_async.exec_exn ~conn ~query in
     let query = "SELECT DISTINCT " ^ (String.concat ~sep:", " Utils.tc_config_fields) ^ " " ^
       (sprintf "FROM tc_config AS c, jobs AS j, (select distinct job_id from soms_jobs where som_id=%d) AS m " som_id) ^
       "WHERE m.job_id=j.job_id AND j.job_id=c.job_id "
     in
-    let job_attributes = Sql.exec_exn ~conn ~query in
-    let som_config_tbl, som_tbl_exists = som_config_tbl_exists ~conn som_id in
-    let som_configs_opt =
-      if not som_tbl_exists then None else
+    let%bind job_attributes = Postgresql_async.exec_exn ~conn ~query in
+    let%bind som_config_tbl, som_tbl_exists = som_config_tbl_exists ~conn som_id in
+    let%bind som_configs_opt =
+      if not som_tbl_exists then return None else
       let query = sprintf "SELECT * FROM %s" som_config_tbl in
-      Some (Sql.exec_exn ~conn ~query) in
+      let%map r = Postgresql_async.exec_exn ~conn ~query in Some r
+    in
     let query =
       "SELECT DISTINCT machine_name, machine_type, cpu_model, number_of_cpus " ^
       (sprintf "FROM machines AS mn, tc_config AS c, (select distinct job_id from soms_jobs where som_id=%d) AS mr " som_id) ^
       "WHERE mn.machine_id=c.machine_id AND c.job_id=mr.job_id "
     in
-    let machines = Sql.exec_exn ~conn ~query in
+    let%bind machines = Postgresql_async.exec_exn ~conn ~query in
     printf "<form name='optionsForm'>\n";
     printf "<table width=\"100%%\" border=\"0\">\n<tr><td>\n";
     self#write_som_info som_info;
@@ -118,8 +121,8 @@ let t ~args = object (self)
     printf "<input value='Brief report analysis' id='preset-brief' type='button'>";
     printf "<br>";
     printf "<div id='axes_selectors'>";
-    print_x_axis_choice ~conn config_columns som_configs_opt;
-    print_y_axis_choice ~conn config_columns som_configs_opt;
+    let%bind () = print_x_axis_choice ~conn config_columns som_configs_opt in
+    let%bind () = print_y_axis_choice ~conn config_columns som_configs_opt in
     printf "</div>\n";
     let checkbox name caption =
       printf "<div id='%s' style='display: inline'>\n" name;
@@ -147,8 +150,8 @@ let t ~args = object (self)
     printf "<br />\n";
     printf "</td></tr>\n</table>\n";
     printf "<div class='filter_table_container'>";
-    self#write_filter_table job_ids builds job_attributes config_columns tc_config_tbl
-      som_configs_opt machines;
+    let%bind() = self#write_filter_table job_ids builds job_attributes config_columns tc_config_tbl
+      som_configs_opt machines in
     printf "</div>";
     printf "</form>\n";
     printf "<div id='graph_title'></div>\n";
@@ -174,4 +177,5 @@ let t ~args = object (self)
     printf "<div id='graph1' class='chart'></div>";
     printf "<div id='graph2' class='chart'></div>";
     self#include_javascript;
+    return ()
 end

src/soms_handler.ml

@@ -1,19 +1,21 @@
 open Core
+open Async
 open Utils
 
 let t ~args = object (self)
   inherit Json_handler.t ~args
 
   method private write_body =
     let query = "SELECT tc_fqn,description FROM test_cases ORDER BY tc_fqn" in
-    let tcs = Sql.exec_exn ~conn ~query in
+    let%bind tcs = Postgresql_async.exec_exn ~conn ~query in
     let json_of_tc tc =
       sprintf "\"%s\":{\"desc\":\"%s\"}" tc.(0) tc.(1) in
     let tcs_json = concat_array (Array.map ~f:json_of_tc tcs#get_all) in
     let query = "SELECT som_id,som_name,tc_fqn FROM soms ORDER BY som_id" in
-    let soms = Sql.exec_exn ~conn ~query in
+    let%bind soms = Postgresql_async.exec_exn ~conn ~query in
     let json_of_som som =
       sprintf "\"%s\":{\"name\":\"%s\",\"tc\":\"%s\"}" som.(0) som.(1) som.(2) in
     let soms_json = concat_array (Array.map ~f:json_of_som soms#get_all) in
-    printf "{\"tcs\":{%s},\"soms\":{%s}}" tcs_json soms_json
+    printf "{\"tcs\":{%s},\"soms\":{%s}}" tcs_json soms_json;
+    return ()
 end

src/std_axes_handler.ml

@@ -1,29 +1,34 @@
 open Core
+open Async
 
 let t ~args = object (self)
   inherit Json_handler.t ~args
 
   method private get_std_xy_choices =
-    let machine_field_lst =
-      List.tl_exn (Sql.get_col_names ~conn ~tbl:"machines") in
+    let%map machine_field_lst =
+      let%map r = Postgresql_async.wrap_sql ~conn (Sql.get_col_names ~tbl:"machines") in
+      List.tl_exn r in
     "branch" :: "build_number" :: "build_tag" :: "patches_applied" :: "build_is_release" ::
     "dom0_memory_static_max" :: "dom0_memory_target" ::
     "cc_restrictions" :: "redo_log" ::
     machine_field_lst
 
   method private get_std_x_choices = self#get_std_xy_choices
 
-  method private get_std_y_choices = "result" :: self#get_std_xy_choices
+  method private get_std_y_choices =
+    let%map r = self#get_std_xy_choices in
+    "result" :: r
 
   method private write_body =
-    let std_x_axes = self#get_std_x_choices in
-    let std_y_axes = self#get_std_y_choices in
+    let%bind std_x_axes = self#get_std_x_choices in
+    let%bind std_y_axes = self#get_std_y_choices in
     let string_of_axes choices =
       let quoted = List.map ~f:(fun c -> "\"" ^ c ^ "\"") choices in
       sprintf "[%s]" (String.concat ~sep:"," quoted)
     in
     printf "{";
     printf "\"std_x_axes\": %s," (string_of_axes std_x_axes);
     printf "\"std_y_axes\": %s" (string_of_axes std_y_axes);
-    printf "}"
+    printf "}";
+    return ()
 end