Merge pull request #1 from edchapman88/policyserver

Policyserver implementation for RLPolicyType

Author: eeelllccc

bin/main.ml

@@ -8,8 +8,18 @@
 *)
 
 open Blue
-module Agent = Markov.Agent.Make (MarkovCompressor) (Reward) (CountBasedPolicy)
+
+module type P =
+  Markov.Agent.RLPolicyType
+    with type state = MarkovCompressor.state
+    with type reward = Reward.t
+
+let policy_module is_server =
+  if is_server then (module ServerPolicy : P) else (module CountBasedPolicy)
 
 let () =
   Cli.arg_parse ();
+  let is_server = Cli.server_policy () |> Option.is_some in
+  let module Policy = (val policy_module is_server) in
+  let module Agent = Markov.Agent.Make (MarkovCompressor) (Reward) (Policy) in
   Agent.act (Agent.init_policy ())

blue/cli.ml

@@ -9,23 +9,26 @@ let usage_msg =
    /dev/ttyUSB0 -s 3.0 \n\n\
   \ Options:"
 
+let parse_ip raw_addr =
+  try
+    match String.split_on_char ':' raw_addr with
+    | [ ip; port ] ->
+        Unix.ADDR_INET (Unix.inet_addr_of_string ip, int_of_string port)
+    | _ -> raise @@ Failure "Network addresses must have exactly one ':'."
+  with Failure msg ->
+    failwith
+    @@ Printf.sprintf
+         "Failed parsing address: '%s'.\n\
+         \ Error: %s\n\
+         \ Accepted format is e.g. '172.0.1.1:8081'" raw_addr msg
+
 let parse_addr raw_addr =
   if String.contains raw_addr ':' then
-    try
-      match String.split_on_char ':' raw_addr with
-      | [ ip; port ] ->
-          Unix.ADDR_INET (Unix.inet_addr_of_string ip, int_of_string port)
-      | _ -> raise @@ Failure "Network addresses must have exactly one ':'."
-    with Failure msg ->
-      failwith
-      @@ Printf.sprintf
-           "Failed parsing response signal address: '%s'. Accepted formats are \
-            e.g. '/dev/ttyUSB0' or '172.0.1.1:8081'. %s"
-           raw_addr msg
+    try parse_ip raw_addr
+    with Failure msg -> failwith @@ msg ^ " or '/dev/USB0'"
   else Unix.ADDR_UNIX raw_addr
 
 let _log_path = ref ""
-let _n_exploration_steps = ref 300
 let _obs_time_delay = ref 5.0
 let _acceptable_fraction = ref 0.8
 let _request_interval = ref 1.
@@ -34,6 +37,9 @@ let _red_ip = ref "172.0.0.3"
 let _response_signal_addr = ref "/dev/ttyUSB0"
 let _parsed_response_signal_addr = ref (parse_addr !_response_signal_addr)
 let _rolling_window_secs = ref 3.0
+let _n_exploration_steps = ref 300
+let _server_policy_addr = ref ""
+let _parsed_server_policy_addr = ref None
 
 let speclist =
   [
@@ -42,10 +48,6 @@ let speclist =
       ": Optionally write a log file in the specified directory with \
        information about the sequence of states observed and actions taken. If \
        no log file is specified, the information is written to stdout.\n" );
-    ( "-e",
-      Arg.Set_int _n_exploration_steps,
-      ": Set the number of observations used by the policy for exploration, \
-       after which actions are selected for exploitation. Defaults to 300.\n" );
     ( "-t",
       Arg.Set_float _obs_time_delay,
       ": Set the constant time delay between observations in seconds. Defaults \
@@ -82,22 +84,36 @@ let speclist =
       ": Set the length of the rolling window used to evaluate the average OK \
        response rate indicated by the data received over the out-of-band \
        channel with the client. Defaults to 3.0.\n" );
+    ( "-e",
+      Arg.Set_int _n_exploration_steps,
+      ": Set the number of observations used by the policy for exploration, \
+       after which actions are selected for exploitation. Defaults to 300.\n" );
+    ( "--server-policy",
+      Arg.Set_string _server_policy_addr,
+      ": Set the IP address of a policy server, overriding the \
+       CountBasedPolicy. Serialised observations and rewards are sent as POST \
+       requests to the configured address and actions are parsed from the \
+       responses. The '-e' flag, used to configure the CountBasedPolicy, is \
+       ignored when '--server-policy' is used.\n" );
   ]
 
 let log_path () =
   if String.length !_log_path == 0 then None else Some !_log_path
 
-let n_exploration_steps () = !_n_exploration_steps
 let obs_time_delay () = !_obs_time_delay
 let acceptable_fraction () = !_acceptable_fraction
 let request_interval () = !_request_interval
 let green_ip () = !_green_ip
 let red_ip () = !_red_ip
 let response_signal_addr () = !_parsed_response_signal_addr
 let rolling_window_secs () = !_rolling_window_secs
+let n_exploration_steps () = !_n_exploration_steps
+let server_policy () = !_parsed_server_policy_addr
 
 let arg_parse () =
   let parse_positional_args _ = () in
   Arg.parse speclist parse_positional_args usage_msg;
   let raw_addr = !_response_signal_addr in
-  _parsed_response_signal_addr := parse_addr raw_addr
+  _parsed_response_signal_addr := parse_addr raw_addr;
+  if String.length !_server_policy_addr > 0 then
+    _parsed_server_policy_addr := Some (parse_ip !_server_policy_addr)

blue/serverPolicy.ml

@@ -0,0 +1,84 @@
+type reward = Reward.t
+type state = MarkovCompressor.state
+type observer = unit -> state
+type action = unit -> unit
+
+type t = int
+(** Number of policy steps. *)
+
+type inference = {
+  action : action;
+  observer : observer;
+  policy : t;
+}
+
+let init () = 0
+let init_observer = MarkovCompressor.observe
+let time_before_retry = 3
+
+let request (state, reward) =
+  let addr = Cli.server_policy () |> Option.get in
+  let sock = Unix.(socket PF_INET SOCK_STREAM 0) in
+  let _ = Unix.connect sock addr in
+
+  let in_ch = Unix.in_channel_of_descr sock in
+  let out_ch = Unix.out_channel_of_descr sock in
+  let body = (state, reward) |> Reward.string_of_state_reward in
+  output_string out_ch
+  @@ Printf.sprintf
+       "POST / HTTP/1.1\r\n\
+        Accept: application/json\r\n\
+        Content-Type: application/json\r\n\
+        Content-Length: %d\r\n\
+        \r\n\
+        %s"
+       (String.length body) body;
+  flush out_ch;
+
+  let rec readall ic acc =
+    try readall ic (acc ^ input_line ic)
+    with End_of_file ->
+      Unix.close sock;
+      acc
+  in
+
+  let res = readall in_ch "" in
+  let start_bod = String.rindex_from res (String.length res - 1) '\r' + 1 in
+  let len_bod = String.length res - start_bod in
+  String.sub res start_bod len_bod
+
+let post (state, reward) = request (state, reward) |> System.eff_of_string
+
+let infer policy (state, reward) =
+  (* Log state and reward. *)
+  Log.write_msg @@ Reward.string_of_state_reward (state, reward);
+
+  (* Send [state] and [reward] to policy server. *)
+  let chosen_eff =
+    match post (state, reward) with
+    | Error msg -> (
+        (* Write policy server failure to log and do one retry. *)
+        Log.write_msg msg;
+        Unix.sleep time_before_retry;
+        (* 'Fail-fast' if the policy server fails after one retry. *)
+        match post (state, reward) with
+        | Error msg ->
+            let msg = "On first (and only) retry: " ^ msg in
+            Log.write_msg msg;
+            failwith msg
+        | Ok eff -> eff)
+    | Ok eff -> eff
+  in
+
+  (* Log chosen effect. *)
+  Log.write_msg @@ System.string_of_eff chosen_eff;
+
+  let action () = System.exec_eff chosen_eff in
+  {
+    action;
+    observer =
+      (fun () ->
+        Unix.sleepf @@ Cli.obs_time_delay ();
+        MarkovCompressor.observe ());
+    policy = policy + 1;
+  }

blue/system.ml

@@ -38,12 +38,17 @@ type eff =
   | ToggleGreen
   | ToggleRed
 
-let string_of_eff e =
-  match e with
+let string_of_eff = function
   | Wait -> "Wait"
   | ToggleGreen -> "ToggleGreen"
   | ToggleRed -> "ToggleRed"
 
+let eff_of_string = function
+  | "Wait" -> Ok Wait
+  | "ToggleGreen" -> Ok ToggleGreen
+  | "ToggleRed" -> Ok ToggleRed
+  | s -> Error (Printf.sprintf "Unrecognised effect: '%s'" s)
+
 let random_eff () =
   match Random.int 3 with
   | 0 -> Wait