Source file xs_server.ml

(*
 * Copyright (C) Citrix Systems Inc.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; version 2.1 only. with the special
 * exception on linking described in file LICENSE.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *)

open Lwt
open Xs_protocol

let ( |> ) a b = b a
let ( ++ ) f g x = f (g x)
let debug fmt = Logging.debug "xs_server" fmt
let error fmt = Logging.error "xs_server" fmt

let store =
  let store = Store.create () in
  List.iter
    (fun path ->
      let p = Store.Path.create path (Store.Path.getdomainpath 0) in
      if not (Store.exists store p) then
        Store.mkdir store 0 (Perms.of_domain 0) p)
    [
      "/local"
    ; "/local/domain"
    ; "/tool"
    ; "/tool/xenstored"
    ; "/tool/xenstored/quota"
    ; "/tool/xenstored/connection"
    ; "/tool/xenstored/log"
    ; "/tool/xenstored/memory"
    ];
  store

module type TRANSPORT = sig
  type 'a t = 'a Lwt.t

  val return : 'a -> 'a Lwt.t
  val ( >>= ) : 'a t -> ('a -> 'b Lwt.t) -> 'b Lwt.t

  type server

  val listen : unit -> server Lwt.t

  type channel

  val read : channel -> bytes -> int -> int -> int Lwt.t
  val write : channel -> bytes -> int -> int -> unit Lwt.t
  val destroy : channel -> unit Lwt.t
  val address_of : channel -> Xs_protocol.address Lwt.t
  val namespace_of : channel -> (module Namespace.IO) option
  val accept_forever : server -> (channel -> unit Lwt.t) -> 'a Lwt.t
end

module Server =
functor
  (T : TRANSPORT)
  ->
  struct
    module PS = PacketStream (T)

    let handle_connection t =
      T.address_of t >>= fun address ->
      let interface = T.namespace_of t in
      let c = Connection.create address interface in
      let channel = PS.make t in
      let m = Lwt_mutex.create () in
      let take_watch_events () =
        let q =
          List.rev
            (Queue.fold (fun acc x -> x :: acc) [] c.Connection.watch_events)
        in
        Queue.clear c.Connection.watch_events;
        q
      in
      let flush_watch_events q =
        Lwt_list.iter_s
          (fun (path, token) ->
            PS.send channel
              Xs_protocol.(Response.(print (Watchevent (path, token)) 0l 0l)))
          q
      in
      let (background_watch_event_flusher : unit Lwt.t) =
        let rec forever () =
          Lwt_mutex.with_lock m (fun () ->
              let rec loop () =
                if Queue.length c.Connection.watch_events = 0 then
                  Lwt_condition.wait ~mutex:m c.Connection.cvar >>= fun () ->
                  loop ()
                else return ()
              in
              loop () >>= fun () -> flush_watch_events (take_watch_events ()))
          >>= fun () -> forever ()
        in
        forever ()
      in

      Lwt.catch
        (fun () ->
          let rec forever () =
            (PS.recv channel >>= function
             | Ok x -> return x
             | Exception e -> Lwt.fail e)
            >>= fun request ->
            let events = take_watch_events () in
            let reply = Call.reply store c request in
            Lwt_mutex.with_lock m (fun () ->
                flush_watch_events events >>= fun () -> PS.send channel reply)
            >>= fun () -> forever ()
          in
          forever () >>= fun () -> T.destroy t)
        (fun _ ->
          Lwt.cancel background_watch_event_flusher;
          Connection.destroy address;
          T.destroy t)

    let serve_forever () =
      Parser.allow_oversize_packets := false;
      T.listen () >>= fun server -> T.accept_forever server handle_connection
  end