HaLVM IOM/scheduler bug

Client.config

name     = "Client"
kernel   = "Client"
memory   = 16
seclabel ='system_u:system_r:domU_t'

Client.hs

import Hypervisor.Console
import Hypervisor.Debug
import XenStore
import Common

main = do
  writeDebugConsole "CLIENT: Initializing XenStore.\n"
  xs  <- initXenStore
  writeDebugConsole "CLIENT: Initialising console.\n"
  con <- initXenConsole
  writeDebugConsole "CLIENT: Starting rendezvous.\n"
  _ <- runClient xs
  writeDebugConsole "CLIENT: Completed rendezvous.\n"

Common.hs

module Common where

import IVC
import Rendezvous
import XenStore

runServer :: XenStore -> (InChannel Int -> IO ()) -> IO ()
runClient :: XenStore -> IO (OutChannel Int)
(runServer, runClient) = clientServerConnection "ClientServerTest" 2

IVC.hs

{-# LANGUAGE BangPatterns, MultiParamTypeClasses, FunctionalDependencies, FlexibleInstances #-}
-- Copyright 2006-2008, Galois, Inc.
-- This software is distributed under a standard, three-clause BSD license.
-- Please see the file LICENSE, distributed with this software, for specific
-- terms and conditions.
-- |Support for inter-domain communication through typed
-- communication channels.  The module provides types that represent
-- the input and ouput ends of open unidirectional communication
-- channels.  These channels are parameterized over the types of
-- messages that can be sent over them, so that domains can exchange
-- messages in a type-safe manner.
--
-- There are also bidirectional channels, parameterized over the types
-- of messages in each direction.
--
module IVC(
         InChannel, OutChannel, InOutChannel
       , ReadableChan, WriteableChan
       , makeNewInChannel, acceptNewInChannel
       , makeNewOutChannel, acceptNewOutChannel
       , makeNewInOutChannel, acceptNewInOutChannel
       , get, put, peer
       )
 where

import Rendezvous
import Control.Concurrent.MVar
import Control.Exception
import Control.Monad
import Data.Binary hiding (get,put)
import Data.Binary.Get(runGet, getWordhost)
import Data.Binary.Put(runPut, putWordhost, putLazyByteString)
import qualified Data.ByteString as BSS
import Data.ByteString.Lazy(ByteString)
import qualified Data.ByteString.Lazy as BS
import Foreign.Marshal.Alloc
import Foreign.Ptr
import Foreign.Storable
import Hypervisor.DomainInfo
import Hypervisor.ErrorCodes
import Hypervisor.Memory
import Hypervisor.Port

data InChannel a = InChannel {
    ichSetupData :: Maybe (DomId, [GrantRef], Port)
  , ichInChannel :: InChan
  , ichPeer      :: DomId
  }

-- |Make a new input channel, targetting the given domain. The second argument
-- is the number of pages to use for the channel. (Note: the actual size of the
-- transfer buffer in memory will be slightly smaller than n * pageSize, because
-- some bookkeeping space is required)
makeNewInChannel :: Binary a => DomId -> Word -> IO (InChannel a)
makeNewInChannel target npages = do
  (grefs, port, ichn) <- makeNewChan target npages buildRawInChan
  return (InChannel (Just (target, grefs, port)) ichn target)

-- |Accept a new input channel, given the input data.
acceptNewInChannel :: Binary a =>
                      DomId -> [GrantRef] -> Port ->
                      IO (InChannel a)
acceptNewInChannel target grants port = do
  ichn <- acceptNewChan target grants port buildRawInChan
  return (InChannel Nothing ichn target)

data OutChannel a = OutChannel {
    ochSetupData  :: Maybe (DomId, [GrantRef], Port)
  , ochOutChannel :: OutChan
  , ochPeer       :: DomId
  }

-- |Make a new output channel, targetting the given domain. The second argument
-- is the number of pages to use for the channel. (Note: the actual size of the
-- transfer buffer in memory will be slightly smaller than n * pageSize, because
-- some bookkeeping space is required)
makeNewOutChannel :: Binary a =>
                     DomId -> Word ->
                     IO (OutChannel a)
makeNewOutChannel target npages = do
  (grefs, port, ochn) <- makeNewChan target npages buildRawOutChan
  return (OutChannel (Just (target, grefs, port)) ochn target)

-- |Accept a new output channel, given the input data
acceptNewOutChannel :: Binary a =>
                       DomId -> [GrantRef] -> Port ->
                       IO (OutChannel a)
acceptNewOutChannel target grants port = do
  ochn <- acceptNewChan target grants port buildRawOutChan
  return (OutChannel Nothing ochn target)

data InOutChannel a b = InOutChannel {
    bchSetupData  :: Maybe (DomId, [GrantRef], Port, Float)
  , bchInChannel  :: InChan
  , bchOutChannel :: OutChan
  , bchPeer       :: DomId
  }

-- |Make a new input / output channel targetting the given domain. The second
-- argument is the number of pages to use, while the third argument tells the
-- system what percentage to use for the input channel. This third argument
-- must be between 0 and 1, inclusive.
makeNewInOutChannel :: (Binary a, Binary b) =>
                       DomId -> Word -> Float ->
                       IO (InOutChannel a b)
makeNewInOutChannel target npages perc
  | (perc < 0) || (perc > 1.0) = throwIO EINVAL
  | otherwise                  = do
     (grs, p, (ich,och)) <- makeNewChan target npages (buildIOChan perc npages)
     return (InOutChannel (Just (target, grs, p, perc)) ich och target)

-- |Accept a new input / out channel, given the input data
acceptNewInOutChannel :: (Binary a, Binary b) =>
                         DomId -> [GrantRef] -> Port -> Float ->
                         IO (InOutChannel a b)
acceptNewInOutChannel target grants port perc
  | (perc < 0) || (perc > 1.0) = throwIO EINVAL
  | otherwise                  = do
     let npages = fromIntegral (length grants)
     (ichn, ochn) <- acceptNewChan target grants port (buildIOChan perc npages)
     return (InOutChannel Nothing ichn ochn target)

buildIOChan :: Float -> Word ->
               Bool -> Ptr Word8 -> Word -> Port ->
               IO (InChan, OutChan)
buildIOChan perc npages doClear ptr _ port = do
  let p1Size   = floor ((fromIntegral (npages * 4096)) * perc)
      p2Size   = (npages * 4096) - p1Size
      b1Size   = p1Size - bookkeepingOverhead
      b2Size   = p2Size - bookkeepingOverhead
  let (inPtr, inSize, outPtr, outSize) =
        if doClear
          then (ptr, b1Size, ptr `plusPtrW` p1Size, b2Size)
          else (ptr `plusPtrW` p1Size, b2Size, ptr, b1Size)
  ichn <- buildRawInChan  doClear inPtr  inSize  port
  ochn <- buildRawOutChan doClear outPtr outSize port
  setPortHandler port $ tryWriteData ochn >> tryReadData  ichn
  return (ichn, ochn)

makeNewChan :: DomId -> Word ->
               (Bool -> Ptr Word8 -> Word -> Port -> IO a) ->
               IO ([GrantRef], Port, a)
makeNewChan target npages buildChan = do
  ptr  <- mallocBytes (fromIntegral npages * 4096)
  refs <- grantAccess target ptr (fromIntegral npages * 4096) True
  port <- allocPort target
  ichn <- buildChan True ptr ((npages * 4096) - bookkeepingOverhead) port
  return (refs, port, ichn)

acceptNewChan :: DomId -> [GrantRef] -> Port ->
                 (Bool -> Ptr Word8 -> Word -> Port -> IO a) ->
                 IO a
acceptNewChan target grefs port buildChan = do
  myport <- bindRemotePort target port
  (ptr, _) <- mapGrants target grefs True
  let size = (length grefs * 4096) - bookkeepingOverhead
  buildChan False ptr (fromIntegral size) myport

-- -----------------------------------------------------------------------------

instance Binary a => RendezvousCapable Word (InChannel a) (OutChannel a) where
  makeConnection other size = do
    res <- makeNewOutChannel other size
    let Just (_, grs, ps) = ochSetupData res
    return (grs, [ps], return res)
  acceptConnection other refs [port] _ = acceptNewInChannel other refs port
  acceptConnection _ _ _ _ = fail "Should only have received one port!"

instance Binary a => RendezvousCapable Word (OutChannel a) (InChannel a) where
  makeConnection other size = do
    res <- makeNewInChannel other size
    let Just (_, grs, ps) = ichSetupData res
    return (grs, [ps], return res)
  acceptConnection other refs [port] _ = acceptNewOutChannel other refs port
  acceptConnection _ _ _ _ = fail "Should only have received one port!"

instance (Binary a, Binary b) =>
           RendezvousCapable (Float, Word) (InOutChannel a b) (InOutChannel b a)
 where
  makeConnection other (perc, size) = do
    res <- makeNewInOutChannel other size perc
    let Just (_, grs, ps, _) = bchSetupData res
    return (grs, [ps], return res)
  acceptConnection other refs [port] (perc, _) =
    acceptNewInOutChannel other refs port perc
  acceptConnection _ _ _ _ =
    fail "Should only have received one port!"

-- -----------------------------------------------------------------------------

class WriteableChan c a | c -> a where
  put :: c -> a -> IO ()

instance Binary a => WriteableChan (OutChannel a) a where
  put c = putBinary (ochOutChannel c)

instance Binary b => WriteableChan (InOutChannel a b) b where
  put c = putBinary (bchOutChannel c)

class ReadableChan c a | c -> a where
  get :: c -> IO a

instance Binary a => ReadableChan (InChannel a) a where
  get c = getBinary (ichInChannel c)

instance Binary a => ReadableChan (InOutChannel a b) a where
  get c = getBinary (bchInChannel c)

putBinary :: Binary a => OutChan -> a -> IO ()
putBinary oc x = runWriteRequest oc (encode x)

getBinary :: Binary a => InChan -> IO a
getBinary ic = decode `fmap` runReadRequest ic

class CommChan c where
  peer :: c -> DomId

instance CommChan (InChannel a) where
  peer = ichPeer

instance CommChan (OutChannel a) where
  peer = ochPeer

instance CommChan (InOutChannel a b) where
  peer = bchPeer


-- -----------------------------------------------------------------------------

--
-- A communications channel is composed of something of a pair of a pointer
-- and a size, where:
--
--  +-----------------------+ ptr + 0
--  +         ...           |
--  +         ...           |
--  +     buffer space      |
--  +         ...           |
--  +         ...           |
--  +-----------------------+ ptr + size
--  +    bytes consumed     |
--  +-----------------------+ ptr + size + 4
--  +    bytes produced     |
--  +-----------------------+ ptr + size + 8
--

bytesConsumed :: Ptr Word8 -> Word -> IO Word32
bytesConsumed p s = peekByteOff (castPtr p) (fromIntegral s)

bytesProduced :: Ptr Word8 -> Word -> IO Word32
bytesProduced p s = peekByteOff (castPtr p) (fromIntegral s + 4)

setBytesConsumed :: Ptr Word8 -> Word -> Word32 -> IO ()
setBytesConsumed p s v = pokeByteOff (castPtr p) (fromIntegral s) v

setBytesProduced :: Ptr Word8 -> Word -> Word32 -> IO ()
setBytesProduced p s v = pokeByteOff (castPtr p) (fromIntegral s + 4) v

bookkeepingOverhead :: Integral a => a
bookkeepingOverhead = 8

-- Internal-only data structure
data OutChan = OutChan {
    ocBuffer  :: Ptr Word8
  , ocSize    :: Word
  , ocModulus :: Word32
  , ocPort    :: Port
  , ocWaiting :: MVar [(ByteString, MVar ())]
  }

buildRawOutChan :: Bool -> Ptr Word8 -> Word -> Port -> IO OutChan
buildRawOutChan doClear buf size port = do
  when doClear $ bzero buf size
  waiters <- newMVar []
  let res = OutChan buf size (computeModulus size) port waiters
  setPortHandler port $ tryWriteData res
  return res

runWriteRequest :: OutChan -> ByteString -> IO ()
runWriteRequest och !bs = do
  resMV <- newEmptyMVar
  waiters <- takeMVar (ocWaiting och)
  putMVar (ocWaiting och) $! (msg, resMV) : waiters
  tryWriteData och
  takeMVar resMV
 where
  !msg = runPut $ do
     putWordhost (fromIntegral (BS.length bs))
     putLazyByteString bs

tryWriteData :: OutChan -> IO ()
tryWriteData och = do
  waiters           <- takeMVar (ocWaiting och)
  cons              <- bytesConsumed (ocBuffer och) (ocSize och)
  prod              <- bytesProduced (ocBuffer och) (ocSize och)
  (waiters', prod') <- doPossibleWrites prod cons waiters
  setBytesProduced (ocBuffer och) (ocSize och) prod'
  when (prod /= prod') $ sendOnPort (ocPort och)
  putMVar (ocWaiting och) $! waiters'
 where
  bufferSize = fromIntegral (ocSize och)
  --
  doPossibleWrites :: Word32 -> Word32 ->
                      [(ByteString, MVar())] ->
                      IO ([(ByteString, MVar())], Word32)
  doPossibleWrites prod _ [] = return ([], prod)
  doPossibleWrites prod cons ls@((bstr, resMV):rest) = do
    -- this is an awkward way to deal with rollver, but it should work.
    let unread = if prod >= cons then prod - cons else overflow
        overflow = prod + (ocModulus och - cons)
        avail = bufferSize - unread
        bstrLn  = fromIntegral (BS.length bstr)
    case () of
      -- In this case, the buffer is full.
      () | avail == 0 ->
        return (ls, prod)
      -- In this case, we have enough space to write the full bytestring.
      () | avail > bstrLn -> do
        writeBS (ocBuffer och) (ocSize och) prod bstr
        putMVar resMV ()
        let prod' = (prod + fromIntegral bstrLn) `mod` ocModulus och
        doPossibleWrites prod' cons rest
      -- In this case, we have space to do a write, but not the whole
      -- bytestring
      () | otherwise -> do
        let (h,t)   = BS.splitAt (fromIntegral avail) bstr
        writeBS (ocBuffer och) (ocSize och) prod h
        let prod' = fromIntegral (prod + avail) `mod` ocModulus och
        return ((t, resMV) : rest, prod')

writeBS :: Ptr Word8 -> Word -> Word32 -> ByteString -> IO ()
writeBS buffer size logical_off lbstr =
  foldM_ doWrite logical_off (BS.toChunks lbstr)
 where
  doWrite :: Word32 -> BSS.ByteString -> IO Word32
  doWrite loff bstr = BSS.useAsCStringLen bstr $ \ (dptr, dlenI) -> do
    let real_off = fromIntegral (loff `mod` fromIntegral size)
        destPtr  = buffer `plusPtrW` real_off
        dlen     = fromIntegral dlenI
    if real_off + dlen > size
      then do let part1s = size - real_off
                  part2s = dlen - part1s
              memcpy destPtr dptr                     part1s
              memcpy buffer  (dptr `plusPtrW` part1s) part2s
      else    memcpy destPtr dptr dlen
    return (loff + fromIntegral dlen)

-- Internal-only data structure
data InChan = InChan {
    icBuffer    :: Ptr Word8
  , icSize      :: Word
  , icModulus   :: Word32
  , icPort      :: Port
  , icStateMV   :: MVar InChanState
  }

data InChanState = NeedSize [MVar ByteString]
                 | GotSize !Word32 ByteString [MVar ByteString]

buildRawInChan :: Bool -> Ptr Word8 -> Word -> Port -> IO InChan
buildRawInChan doClear buf size port = do
  when doClear $ bzero buf size
  stateMV <- newMVar (NeedSize [])
  let res = InChan buf size (computeModulus size) port stateMV
  setPortHandler port $ tryReadData res
  return res

runReadRequest :: InChan -> IO ByteString
runReadRequest ich = do
  resMV <- newEmptyMVar
  istate <- takeMVar (icStateMV ich)
  case istate of
    NeedSize waiters ->
      putMVar (icStateMV ich) $! NeedSize (waiters ++ [resMV])
    GotSize n acc waiters ->
      putMVar (icStateMV ich) $! GotSize n acc (waiters ++ [resMV])
  tryReadData ich
  takeMVar resMV

tryReadData :: InChan -> IO ()
tryReadData ich = modifyMVar_ (icStateMV ich) $ \ istate -> do
  prod             <- bytesProduced (icBuffer ich) (icSize ich)
  cons             <- bytesConsumed (icBuffer ich) (icSize ich)
  (istate', cons') <- doPossibleReads prod cons istate
  setBytesConsumed (icBuffer ich) (icSize ich) cons'
  when (cons /= cons') $ sendOnPort (icPort ich)
  return istate'
 where
  doPossibleReads :: Word32 -> Word32 -> InChanState -> IO (InChanState, Word32)
  doPossibleReads prod cons istate = do
    let avail = if prod >= cons then prod - cons else overflow
        overflow = prod + (icModulus ich - cons)
    case istate of
      -- If we need to get a size, we have waiters, and there's at least
      -- four bytes to read, then we should read off the size.
      NeedSize ws@(_:_) | avail >= sizeSize -> do
        sizeBS <- readBS (icBuffer ich) (icSize ich) cons sizeSize
        let size = runGet getWordhost sizeBS
        let istate' = GotSize (fromIntegral size) BS.empty ws
            cons'   = (cons + sizeSize) `mod` icModulus ich
        doPossibleReads prod cons' istate'
      -- If we have some data, but not enough, update ourselves with the
      -- new data and the lesser requirement.
      GotSize n acc ws | (avail > 0) && (n > avail) -> do
        part <- readBS (icBuffer ich) (icSize ich) cons avail
        let istate' = GotSize (n - avail) (acc `BS.append` part) ws
            cons'   = (cons + avail) `mod` icModulus ich
        doPossibleReads prod cons' istate'
      -- If we can read everything, do it!
      GotSize n acc (f:rest) | (avail > 0) && (n <= avail) -> do
        endp <- readBS (icBuffer ich) (icSize ich) cons n
        putMVar f (acc `BS.append` endp)
        let cons' = (cons + n) `mod` icModulus ich
        doPossibleReads prod cons' (NeedSize rest)
      -- Otherwise, we can't do anything
      _ ->
        return (istate, cons)

readBS :: Ptr Word8 -> Word -> Word32 -> Word32 -> IO ByteString
readBS !buffer !sizeW !logical_off !amt = do
  let real_off = logical_off `mod` size
      readPtr  = buffer `plusPtrW` real_off
      part1sz  = size - real_off
      part2sz  = amt - part1sz
  if real_off + amt > size
    then do part1 <- packCStringLen readPtr part1sz
            part2 <- packCStringLen buffer  part2sz
            return $! BS.fromStrict part1 `BS.append` BS.fromStrict part2
    else    BS.fromStrict `fmap` packCStringLen readPtr amt
 where
  size = fromIntegral sizeW
  packCStringLen p s = BSS.packCStringLen (castPtr p, fromIntegral s)

plusPtrW :: Integral b => Ptr a -> b -> Ptr a
plusPtrW p x = p `plusPtr` (fromIntegral x)

sizeSize :: Integral a => a
sizeSize = fromIntegral (BS.length (runPut (putWordhost 0)))

computeModulus :: Word -> Word32
computeModulus size
  | base == 0 = fromIntegral q * (fromIntegral size - 1)
  | otherwise = base
  where
   base  = fromIntegral q * fromIntegral size
   size' = fromIntegral size :: Word64
   q     = 0x100000000 `div` size'

foreign import ccall unsafe "strings.h bzero"
  bzero :: Ptr a -> Word -> IO ()

foreign import ccall unsafe "string.h memcpy"
  memcpy :: Ptr a -> Ptr b -> Word -> IO ()

Makefile

# BANNERSTART
# - Copyright 2006-2008, Galois, Inc.
# - This software is distributed under a standard, three-clause BSD license.
# - Please see the file LICENSE, distributed with this software, for specific
# - terms and conditions.
# Author: Adam Wick <[email protected]>
# BANNEREND
#

GHC=halvm-ghc --make

all: Server Client

XenStore.hs: XenStore.hsc
	hsc2hs XenStore.hsc

Server: Server.hs Rendezvous.hs IVC.hs Common.hs XenStore.hs
	$(GHC) Server.hs

Client: Client.hs Rendezvous.hs IVC.hs Common.hs XenStore.hs
	$(GHC) Client.hs

run: Server Client
	-sudo rm -f /var/log/xen/console/*.log
	-sudo xenstore-rm /rendezvous/ClientServerTest
	-sudo xl destroy Server
	-sudo xl destroy Client1
	-sudo xl destroy Client2
	-sudo xl destroy Client3
	-sudo xl destroy Client4
	-sudo xl destroy Client5
	-sudo xl destroy Client6
	-sudo xl destroy Client7
	-sudo xl destroy Client8
	-sudo xl destroy Client9
	-sudo xl destroy Client10
	-sudo xl destroy Client11
	-sudo xl destroy Client12
	-sudo xl destroy Client13
	-sudo xl destroy Client14
	-sudo xl destroy Client15
	sudo xl create Server.config
	sudo xl create Client.config "name='Client1'"
	sudo xl create Client.config "name='Client2'"
	sudo xl create Client.config "name='Client3'"
	sudo xl create Client.config "name='Client4'"
	sudo xl create Client.config "name='Client5'"
	sudo xl create Client.config "name='Client6'"
	sudo xl create Client.config "name='Client7'"
	sudo xl create Client.config "name='Client8'"
	sudo xl create Client.config "name='Client9'"
	sudo xl create Client.config "name='Client10'"
	sudo xl create Client.config "name='Client11'"
	sudo xl create Client.config "name='Client12'"
	sudo xl create Client.config "name='Client13'"
	sudo xl create Client.config "name='Client14'"
	sudo xl create Client.config "name='Client15'"
	sleep 8
	sudo xl dmesg -c
	sudo xenstore-ls
	sudo xl list

clean:
	-rm *.hi *.o Client Server XenStore.hs

cleanup:
	-sudo xl destroy Server
	-sudo xl destroy Client1
	-sudo xl destroy Client2
	-sudo xl destroy Client3
	-sudo xl destroy Client4
	-sudo xl destroy Client5
	-sudo xl destroy Client6
	-sudo xl destroy Client7
	-sudo xl destroy Client8
	-sudo xl destroy Client9
	-sudo xl destroy Client10
	-sudo xl destroy Client11
	-sudo xl destroy Client12
	-sudo xl destroy Client13
	-sudo xl destroy Client14
	-sudo xl destroy Client15

Rendezvous.hs

{-# LANGUAGE MultiParamTypeClasses, FunctionalDependencies, MultiWayIf #-}
-- Copyright 2013, Galois, Inc.
-- This software is distributed under a standard, three-clause BSD license.
-- Please see the file LICENSE, distributed with this software, for specific
-- terms and conditions.
-- |Routines for automatically performing rendezvous between two domains.
module Rendezvous(
         RendezvousCapable(..)
       , peerConnection
       , clientServerConnection
       )
 where

import Control.Applicative
import Control.Concurrent
import Control.Exception
import Data.Word
import Hypervisor.DomainInfo
import Hypervisor.ErrorCodes
import Hypervisor.Memory
import Hypervisor.Port
import XenStore

-- |The class of objects that are connectable in a peer-to-peer fashion.
-- If your underlying system (whatever it may be) uses an interface like this,
-- then this library can automatically set up connection rendezvous for you
-- through the XenStore.
--
-- The first type is an "extra" bit of information that is useful to the
-- system.
--
-- The second type is the thing the "accepting" side will receive, the third
-- is the type of thing the "offering" side will receive.
class (Show a, Read a) => RendezvousCapable a b c | b c -> a, b -> c, c -> b where
  -- |Create the basic connection structures for a connection between the
  -- current domain and the given one. The returned values should be the
  -- list of grant references to share, a list of ports to share, and a
  -- thunk to invoke when the connection is complete.
  makeConnection    :: DomId -> a -> IO ([GrantRef], [Port], IO c)
  -- |Accept a connection offered by the other side of the rendezvous.
  acceptConnection  :: DomId -> [GrantRef] -> [Port] -> a -> IO b

-- |Given a name for the connection (which should be unique on the host for
-- the duration of the rendezvous) and the special extra information used in
-- the item, create thunks that, when executed, will perform rendezvous
-- between domains.
--
-- Typically, this will be invoked from a shared module, and one domain will
-- use one result while the other will use the other result.
peerConnection :: RendezvousCapable a b c =>
                  String -> a ->
                  (XenStore -> IO b, XenStore -> IO c)
peerConnection name extra = (runLeftSide, runRightSide)
 where
  targetPath = "/rendezvous/" ++ name
  --
  runLeftSide xs = do
    me    <- xsGetDomId xs
    removePath       xs targetPath
    xsMakeDirectory  xs targetPath
    xsSetPermissions xs targetPath [ReadWritePerm me]
    xsWrite          xs (targetPath ++ "/LeftDomId") (show me)
    other  <- read <$> waitForKey xs (targetPath ++ "/RightDomId")
    grants <- read <$> waitForKey xs (targetPath ++ "/RightGrantRefs")
    ports  <- read <$> waitForKey xs (targetPath ++ "/RightPorts")
    res    <- acceptConnection other grants ports extra
    xsWrite xs (targetPath ++ "/LeftConnectionConfirmed") "True"
    return res
  runRightSide xs = do
    other <- read `fmap` waitForKey xs (targetPath ++ "/LeftDomId")
    me    <- xsGetDomId xs
    (gs, ps, confirm) <- makeConnection other extra
    xsWrite xs (targetPath ++ "/RightDomId") (show me)
    xsWrite xs (targetPath ++ "/RightGrantRefs") (show gs)
    xsWrite xs (targetPath ++ "/RightPorts") (show ps)
    _ <- waitForKey xs (targetPath ++ "/LeftConnectionConfirmed")
    removePath xs targetPath
    confirm

clientServerConnection :: RendezvousCapable a b c =>
                          String -> a ->
                          (XenStore -> (b -> IO ()) -> IO (), XenStore -> IO c)
clientServerConnection name extra = (runServer, runClient)
 where
  targetPath = "/rendezvous/" ++ name
  --
  runServer xs callback = do
    me    <- xsGetDomId xs
    removePath xs targetPath
    xsMakeDirectory xs targetPath
    xsWrite         xs (targetPath ++ "/ServerDomId") (show me)
    xsWatch xs targetPath "" $ \ key _ -> do
      -- putStrLn $ "XenStore watch fired for " ++ key
      case reads (reverse $ takeWhile (/= '/') $ reverse key) of
        [(domid, "")] -> do xsWrite xs (key ++ "/ServerAcknowledge") "True"
                            g <- read <$> waitForKey xs (key ++ "/ClientGrants")
                            p <- read <$> waitForKey xs (key ++ "/ClientPorts")
                            res <- acceptConnection domid g p extra
                            putStrLn $ "Writing confirmation for " ++ key
                            xsWrite xs (key ++ "/ServerConfirmed") "True"
                            putStrLn $ "Calling back " ++ key
                            callback res -- might as well reuse this thread
        _             -> return ()
  --
  runClient xs = do
    me    <- xsGetDomId xs
    other <- read `fmap` waitForKey xs (targetPath ++ "/ServerDomId")
    (gs, ps, confirm) <- makeConnection other extra
    let targetPath' = targetPath ++ "/" ++ show me
        syn = do -- Repeat the mkDir until the server acks
          _ <- try (xsRemove xs targetPath') :: IO (Either SomeException ())
          xsMakeDirectory xs targetPath'
          s <- waitForKey' xs (targetPath' ++ "/ServerAcknowledge") 50
          maybe syn (\_ -> return ()) s
    syn
    xsWrite xs (targetPath' ++ "/ClientGrants") (show gs)
    xsWrite xs (targetPath' ++ "/ClientPorts")  (show ps)
    _ <- waitForKey xs (targetPath' ++ "/ServerConfirmed")
    putStrLn "Connection established."
    confirm

-- wait for key a couple of times, then give up.
-- Int param: number of retries, 100ms wait between them.
waitForKey' :: XenStore -> String -> Word64 -> IO (Maybe String)
waitForKey' xs key retries = do
  putStrLn $ "Waiting for " ++ key
  eres <- try (xsRead xs key) :: IO (Either SomeException String)
  either next (\s -> great >> return (Just s)) eres
  where 
    wait = threadDelay 100000 >> sigh
    sigh = putStrLn $ "I guess I will wait for " ++ key ++ " later..."
    nope = putStrLn $ "Gave up waiting for " ++ key
    great = putStrLn $ "Got " ++ key
    next _ = if | retries == 0 -> nope >> return Nothing
                | retries <  0 -> wait >> waitForKey' xs key  retries
                | otherwise    -> wait >> waitForKey' xs key (retries-1)


waitForKey :: XenStore -> String -> IO String
waitForKey xs key = do
  putStrLn $ "Waiting for " ++ key
  eres <- catch (Right <$> xsRead xs key) leftError
  case eres of
    Left _    -> sigh >> threadDelay 100000 >> waitForKey xs key
    Right res -> great >> return res
 where
  leftError :: ErrorCode -> IO (Either ErrorCode String)
  leftError = return . Left
  sigh = putStrLn $ "I guess I will wait for " ++ key ++ " later..."
  great = putStrLn $ "Got " ++ key

removePath :: XenStore -> String -> IO ()
removePath xs str = do catch remSubItems onECContinue
                       catch remItem     onECContinue
 where
  remSubItems = mapM_ (removePath xs) =<< xsDirectory xs str
  remItem     = xsRemove xs str

onECContinue :: ErrorCode -> IO ()
onECContinue _ = return ()

Server.config

name     = "Server"
kernel   = "Server"
memory   = 32
seclabel ='system_u:system_r:domU_t'

Server.hs

import Control.Concurrent
import Hypervisor.Console
import Hypervisor.Debug
import XenStore
import Common

main = do
  writeDebugConsole "SERVER: Initializing XenStore.\n"
  xs  <- initXenStore
  writeDebugConsole "SERVER: Initializing console.\n"
  con <- initXenConsole
  writeDebugConsole "SERVER: Initializing MVar.\n"
  countMV <- newMVar 0
  writeDebugConsole "SERVER: Starting rendezvous.\n"
  runServer xs $ \ _ -> do
    writeDebugConsole "SERVER: Found a client!\n"
    cur <- takeMVar countMV
    putMVar countMV $! cur + 1
  waitFor countMV 15
  writeDebugConsole "SERVER: Got all my clients!\n"

waitFor :: MVar Int -> Int -> IO ()
waitFor mv goal = do
  cur <- takeMVar mv
  if cur == goal
    then return ()
    else do putMVar mv cur
            threadDelay 10000
            waitFor mv goal

XenStore.hsc

{-# LANGUAGE ForeignFunctionInterface, DeriveGeneric #-}
-- Copyright 2013 Galois, Inc.
-- This software is distributed under a standard, three-clause BSD license.
-- Please see the file LICENSE, distributed with this software, for specific
-- terms and conditions.
module XenStore(
         XenStore
       , TransId
       , XSPerm(..)
       , initXenStore
       , initCustomXenStore
       , emptyTransaction
       , xsGetDomId
       , xsDirectory, xstDirectory
       , xsRead, xstRead
       , xsGetPermissions, xstGetPermissions
       , xsWatch
       , xsUnwatch, xstUnwatch
       , xsStartTransaction, xstStartTransaction
       , xstAbort, xstCommit
       , xsIntroduce, xstIntroduce
       , xsRelease, xstRelease
       , xsGetDomainPath, xstGetDomainPath
       , xsWrite, xstWrite
       , xsMakeDirectory, xstMakeDirectory
       , xsRemove, xstRemove
       , xsSetPermissions, xstSetPermissions
       , xsIsDomainIntroduced, xstIsDomainIntroduced
       , xsResume, xstResume
       , xsSetTarget, xstSetTarget
       , xsRestrict, xstRestrict
#ifdef XS_RESET_WATCHES
       , xsResetWatches, xstResetWatches
#endif
       )
 where

import Control.Applicative
import Control.Concurrent
import Control.Exception
import Control.Monad
import Data.Binary.Get
import Data.Binary.Put
import Data.Bits
import Data.ByteString(packCStringLen,useAsCStringLen)
import Data.ByteString.Lazy(ByteString)
import qualified Data.ByteString.Lazy as BS
import Data.Char
import Data.List
import Data.Map.Strict(Map)
import qualified Data.Map.Strict as Map
import Data.Word
import Foreign.C.String
import Foreign.C.Types
import Foreign.Ptr
import Foreign.Storable
import GHC.Generics
import Hypervisor.Debug
import Hypervisor.DomainInfo
import Hypervisor.ErrorCodes
import Hypervisor.Memory
import Hypervisor.Port

newtype XenStore = XenStore (MVar XenbusState)

data XSPerm =
    WritePerm     DomId
  | ReadPerm      DomId
  | ReadWritePerm DomId
  | NonePerm      DomId
 deriving (Eq, Show, Generic)

newtype ReqId   = ReqId { unReqId :: Word32 }
 deriving (Eq, Ord, Show)

advanceReqId :: ReqId -> ReqId
advanceReqId (ReqId x) = ReqId (if x + 1 == 0 then 1 else x + 1)

newtype TransId = TransId { unTransId :: Word32 }
 deriving (Eq, Generic, Show)

emptyTransaction :: TransId
emptyTransaction  = TransId 0

initXenStore :: IO XenStore
initXenStore  = do
  xsMFN  <- (toMFN . fromIntegral) <$> get_xenstore_mfn
  xsPort <- toPort <$> get_xenstore_evtchn
  initCustomXenStore xsMFN xsPort

initCustomXenStore :: MFN -> Port -> IO XenStore
initCustomXenStore xsMFN xsPort = do
  xsPtr <- handle (mapMFN xsMFN) (mfnToVPtr xsMFN)
  let initState = state0 xsPort xsPtr
  stateMV <- newMVar initState
  setPortHandler xsPort (onXenbusEvent stateMV)
  return (XenStore stateMV)
 where
  mapMFN :: MFN -> ErrorCode -> IO (VPtr a)
  mapMFN mfn _ = mapFrames [mfn]

xsGetDomId :: XenStore -> IO DomId
xsGetDomId xs = do
  val <- xsRead xs "domid"
  case reads val :: [(Word16, String)] of
    [(domid, "")] -> return (toDomId domid)
    _             -> throwIO EPROTO

xsDirectory :: XenStore -> String -> IO [String]
xsDirectory xs = xstDirectory xs emptyTransaction

xstDirectory :: XenStore -> TransId -> String -> IO [String]
xstDirectory (XenStore xbs) tid str =
  standardRequest xbs tid (XSDir str) RTDir parseStrings

xsRead :: XenStore -> String -> IO String
xsRead xs = xstRead xs emptyTransaction

xstRead :: XenStore -> TransId -> String -> IO String
xstRead (XenStore xbs) tid str =
  standardRequest xbs tid (XSRead str) RTRead parseString

xsGetPermissions :: XenStore -> String -> IO [XSPerm]
xsGetPermissions xs = xstGetPermissions xs emptyTransaction

xstGetPermissions :: XenStore -> TransId -> String -> IO [XSPerm]
xstGetPermissions (XenStore xbs) tid str =
  standardRequest xbs tid (XSGetPerms str) RTGetPerms parsePerms

xsWatch :: XenStore -> String -> String -> (String -> String -> IO ()) -> IO ()
xsWatch (XenStore xbs) str token handler =
  do modifyMVar_ xbs $ \ state ->
       let oldwatches = waitingWatches state
       in return state{ waitingWatches = oldwatches ++ [(str, token, handler)] }
     standardRequest xbs emptyTransaction (XSWatch str token) RTWatch (const ())

xsUnwatch :: XenStore -> String -> String -> IO ()
xsUnwatch xs = xstUnwatch xs emptyTransaction

xstUnwatch :: XenStore -> TransId -> String -> String -> IO ()
xstUnwatch (XenStore xbs) tid wpth tok =
  do modifyMVar_ xbs $ \ state ->
       let watches = filter (\ (a,b,_) -> (a /= wpth) || (b /= tok))
                            (waitingWatches state)
       in return state{ waitingWatches = watches }
     standardRequest xbs tid (XSUnwatch wpth tok) RTUnwatch (const ())

xsStartTransaction :: XenStore -> IO TransId
xsStartTransaction xs = xstStartTransaction xs emptyTransaction

xstStartTransaction :: XenStore -> TransId -> IO TransId
xstStartTransaction (XenStore xbs) tid =
  standardRequest xbs tid XSTransSt RTTransSt parseTransId

xstAbort :: XenStore -> TransId -> IO ()
xstAbort (XenStore xbs) tid =
  standardRequest xbs tid (XSTransEnd False) RTTransEnd (const ())

xstCommit :: XenStore -> TransId -> IO ()
xstCommit (XenStore xbs) tid =
  standardRequest xbs tid (XSTransEnd True) RTTransEnd (const ())

xsIntroduce :: XenStore -> DomId -> MFN -> Port -> IO ()
xsIntroduce xs = xstIntroduce xs emptyTransaction

xstIntroduce :: XenStore -> TransId -> DomId -> MFN -> Port -> IO ()
xstIntroduce (XenStore xbs) tid d m p =
  standardRequest xbs tid (XSIntro d m p) RTIntro (const ())

xsRelease :: XenStore -> DomId -> IO ()
xsRelease xs = xstRelease xs emptyTransaction

xstRelease :: XenStore -> TransId -> DomId -> IO ()
xstRelease (XenStore xbs) tid d =
  standardRequest xbs tid (XSRelease d) RTRelease (const ())

xsGetDomainPath :: XenStore -> DomId -> IO String
xsGetDomainPath xs = xstGetDomainPath xs emptyTransaction

xstGetDomainPath :: XenStore -> TransId -> DomId -> IO String
xstGetDomainPath (XenStore xbs) tid d =
  standardRequest xbs tid (XSGetPath d) RTGetPath parseString

xsWrite :: XenStore -> String -> String -> IO ()
xsWrite xs = xstWrite xs emptyTransaction

xstWrite :: XenStore -> TransId -> String -> String -> IO ()
xstWrite (XenStore xbs) tid k v =
  standardRequest xbs tid (XSWrite k v) RTWrite (const ())

xsMakeDirectory :: XenStore -> String -> IO ()
xsMakeDirectory xs = xstMakeDirectory xs emptyTransaction

xstMakeDirectory :: XenStore -> TransId -> String -> IO ()
xstMakeDirectory (XenStore xbs) tid d =
  standardRequest xbs tid (XSMkDir d) RTMkDir (const ())

xsRemove :: XenStore -> String -> IO ()
xsRemove xs = xstRemove xs emptyTransaction

xstRemove :: XenStore -> TransId -> String -> IO ()
xstRemove (XenStore xbs) tid k =
  standardRequest xbs tid (XSRm k) RTRm (const ())

xsSetPermissions :: XenStore -> String -> [XSPerm] -> IO ()
xsSetPermissions xs = xstSetPermissions xs emptyTransaction

xstSetPermissions :: XenStore -> TransId -> String -> [XSPerm] -> IO ()
xstSetPermissions (XenStore xbs) tid k ps =
  standardRequest xbs tid (XSSetPerms k ps) RTSetPerms (const ())

xsIsDomainIntroduced :: XenStore -> DomId -> IO Bool
xsIsDomainIntroduced xs = xstIsDomainIntroduced xs emptyTransaction

xstIsDomainIntroduced :: XenStore -> TransId -> DomId -> IO Bool
xstIsDomainIntroduced (XenStore xbs) tid d =
  standardRequest xbs tid (XSIsDomInt d) RTIsDomInt parseBool

xsResume :: XenStore -> DomId -> IO ()
xsResume xs = xstResume xs emptyTransaction

xstResume :: XenStore -> TransId -> DomId -> IO ()
xstResume (XenStore xbs) tid d =
  standardRequest xbs tid (XSResume d) RTResume (const ())

xsSetTarget :: XenStore -> DomId -> DomId -> IO ()
xsSetTarget xs = xstSetTarget xs emptyTransaction

xstSetTarget :: XenStore -> TransId -> DomId -> DomId -> IO ()
xstSetTarget (XenStore xbs) tid d td =
  standardRequest xbs tid (XSSetTarg d td) RTSetTarg (const ())

xsRestrict :: XenStore -> DomId -> IO ()
xsRestrict xs = xstRestrict xs emptyTransaction

xstRestrict :: XenStore -> TransId -> DomId -> IO ()
xstRestrict (XenStore xbs) tid d =
  standardRequest xbs tid (XSRestrict d) RTRestrict (const ())

#ifdef XS_RESET_WATCHES
xsResetWatches :: XenStore -> IO ()
xsResetWatches xs = xstResetWatches xs emptyTransaction

xstResetWatches :: XenStore -> TransId -> IO ()
xstResetWatches (XenStore xbs) tid =
  standardRequest xbs tid XSReset RTReset (const ())
#endif

standardRequest :: MVar XenbusState ->
                   TransId -> XenbusRequest ->
                   ResponseType -> (ByteString -> a) ->
                   IO a
standardRequest stateMV tid req goodresp converter =
  writeRequest stateMV tid req $ \ body ->
    case body of
      RespBody _ RTError bstr ->
        case reads (parseString bstr) of
          ((x,_):_)  -> throwIO (x :: ErrorCode)
          _          -> throwIO EIO
      RespBody _ rtype bstr | rtype == goodresp ->
        return (converter bstr)
      RespBody _ rtype _ ->
        do writeDebugConsole ("ERROR: Xenbus: Expected " ++ show goodresp ++
                              " but got " ++ show rtype ++ "\n")
           throwIO EIO

-- ----------------------------------------------------------------------------

data XenbusState = XBS {
    xbPort          :: Port
  , xbRing          :: XSRing
  , nextRequestId   :: ReqId
  , decodeStream    :: ByteString
  , pendingWrites   :: ByteString
  , waitingRequests :: Map ReqId (MVar ResponseBody)
  , waitingWatches  :: [(String, String, String -> String -> IO ())]
  }

state0 :: Port -> XSRing -> XenbusState
state0 port ring = XBS {
    xbPort          = port
  , xbRing          = ring
  , nextRequestId   = ReqId 1000
  , decodeStream    = BS.empty
  , pendingWrites   = BS.empty
  , waitingRequests = Map.empty
  , waitingWatches  = []
  }

writeRequest :: MVar XenbusState ->
                TransId -> XenbusRequest ->
                (ResponseBody -> IO a) ->
                IO a
writeRequest stateMV tid req k =
  do respMV <- newEmptyMVar
     state <- takeMVar stateMV
     let rid     = nextRequestId state
         newbstr = buildRequest rid tid req
         pend    = pendingWrites state `BS.append` newbstr
         table'  = Map.insert rid respMV (waitingRequests state)
     putStrLn $ "Writing request data for #" ++ show rid ++ " " ++ show req 
     remain <- writeRequestData (xbPort state) (xbRing state) pend
     putMVar stateMV $! state{ nextRequestId   = advanceReqId rid
                             , pendingWrites   = remain
                             , waitingRequests = table' }
     putStrLn $ "Waiting for response for #" ++ show rid ++ " " ++ show req
     st <- takeMVar respMV
     putStrLn $ "Got response for #" ++ show rid ++ " " ++ show req
     k st

onXenbusEvent :: MVar XenbusState -> IO ()
onXenbusEvent stateMV =
  do state <- takeMVar stateMV
     let port = xbPort state
         ring = xbRing state

     -- The devil lies here
     inputbstr <- readNewResponseData port ring (decodeStream state)
     let (resps, inputbstr') = parseResponses inputbstr
     putStr $ "Got events : (" ++ show resps ++ ", " ++ show inputbstr' ++ ")\n" ++
              "    (from " ++ show inputbstr ++ ")\n"
     -- enddevil

     tab <- foldM' (waitingRequests state) resps $ \ table resp ->
       case resp of
         -- Fire any watches associated with this event
         RespBody _ RTEvent bstr ->
           do let bsparts = BS.split 0 bstr
                  parts   = map (map (chr . fromIntegral) . BS.unpack) bsparts
                  (key:token:_) = parts
              forM_ (waitingWatches state) $ \ (watchOn, _, action) ->
                when (watchOn `isPrefixOf` key) $
                  forkIO_ (action key token)
              return table
         -- Run any handlers associated with this id
         RespBody rid _ _ ->
           case Map.lookup rid table of
             Nothing ->
               do writeDebugConsole ("WARNING: Xenbus: Response with bad id: "
                                      ++ (show rid) ++ "\n")
                  return table
             Just mvar ->
               do putMVar mvar resp
                  return (Map.delete rid table)
     remn <- writeRequestData port ring (pendingWrites state)
     putMVar stateMV $! state{ decodeStream = inputbstr'
                             , pendingWrites = remn
                             , waitingRequests = tab }
 where
  readNewResponseData port ring acc =
    do newstuff <- readResponseData port ring
       if BS.null newstuff
         then return acc
         else readNewResponseData port ring (acc `BS.append` newstuff)

parseResponses :: ByteString -> ([ResponseBody], ByteString)
parseResponses bstr =
  case runGetOrFail parseResponse bstr of
    Left  (remain, _, _)   -> ([], bstr)
    Right (rest,   _, req) ->
      let (res, remain) = parseResponses rest
      in (req : res, remain)

buildRequest :: ReqId -> TransId -> XenbusRequest -> ByteString
buildRequest rid tid xbr = runPut $ do
  putWord32host (requestId xbr)                 -- uint32_t type
  putWord32host (unReqId rid)                   -- uint32_t req_id
  putWord32host (unTransId tid)                 -- uint32_t tx_id
  putWord32host (fromIntegral (BS.length body)) -- uint32_t len
  putLazyByteString body
 where body = runPut (renderBody xbr)


-- ----------------------------------------------------------------------------

#include <stdint.h>
#include <sys/types.h>
#include <xen/io/xs_wire.h>

#ifndef XENSTORE_RING_SIZE
#error "BAD BAD BAD"
#endif

data ResponseBody = RespBody ReqId ResponseType ByteString
  deriving (Show)

parseResponse :: Get ResponseBody
parseResponse =
  do rtype <- getResponseType <$> getWord32host
     rid   <- getWord32host
     _tid  <- getWord32host
     len   <- getWord32host
     body  <- getLazyByteString (fromIntegral len)
     return (RespBody (ReqId rid) rtype body)

data ResponseType = RTRead     | RTWrite  | RTMkDir   | RTRm       | RTDir
                  | RTSetPerms | RTWatch  | RTUnwatch | RTRestrict | RTTransSt
                  | RTTransEnd | RTIntro  | RTRelease | RTGetPath  | RTError
                  | RTIsDomInt | RTResume | RTSetTarg | RTGetPerms | RTEvent
#ifdef XS_RESET_WATCHES
                  | RTReset
#endif
                  | RTUnknown Word32
 deriving (Show, Eq)

getResponseType :: Word32 -> ResponseType
getResponseType (#const XS_DIRECTORY)            = RTDir
getResponseType (#const XS_READ)                 = RTRead
getResponseType (#const XS_GET_PERMS)            = RTGetPerms
getResponseType (#const XS_WATCH)                = RTWatch
getResponseType (#const XS_UNWATCH)              = RTUnwatch
getResponseType (#const XS_TRANSACTION_START)    = RTTransSt
getResponseType (#const XS_TRANSACTION_END)      = RTTransEnd
getResponseType (#const XS_INTRODUCE)            = RTIntro
getResponseType (#const XS_RELEASE)              = RTRelease
getResponseType (#const XS_GET_DOMAIN_PATH)      = RTGetPath
getResponseType (#const XS_WRITE)                = RTWrite
getResponseType (#const XS_MKDIR)                = RTMkDir
getResponseType (#const XS_RM)                   = RTRm
getResponseType (#const XS_SET_PERMS)            = RTSetPerms
getResponseType (#const XS_WATCH_EVENT)          = RTEvent
getResponseType (#const XS_ERROR)                = RTError
getResponseType (#const XS_IS_DOMAIN_INTRODUCED) = RTIsDomInt
getResponseType (#const XS_RESUME)               = RTResume
getResponseType (#const XS_SET_TARGET)           = RTSetTarg
getResponseType (#const XS_RESTRICT)             = RTRestrict
#ifdef XS_RESET_WATCHES
getResponseType (#const XS_RESET_WATCHES)        = RTReset
#endif
getResponseType x                                = RTUnknown x

parseStrings :: ByteString -> [String]
parseStrings bstr = map translateAscii (BS.split 0 bstr)

parseString :: ByteString -> String
parseString bstr = translateAscii (BS.takeWhile (/= 0) bstr)

parseBool :: ByteString -> Bool
parseBool bstr = parseString bstr == "T"

parseTransId :: ByteString -> TransId
parseTransId bstr = TransId (read (parseString bstr))

parsePerms :: ByteString -> [XSPerm]
parsePerms bstr = map translateString (parseStrings bstr)
 where
  translateString ('r':rest) = ReadPerm      (toDomId (read rest :: Word16))
  translateString ('w':rest) = WritePerm     (toDomId (read rest :: Word16))
  translateString ('b':rest) = ReadWritePerm (toDomId (read rest :: Word16))
  translateString ('n':rest) = NonePerm      (toDomId (read rest :: Word16))
  translateString _          = throw EIO

translateAscii :: ByteString -> String
translateAscii  = map castCUCharToChar . map CUChar . BS.unpack

-- ----------------------------------------------------------------------------

data XenbusRequest =
    XSRead     String
  | XSWrite    String String
  | XSMkDir    String
  | XSRm       String
  | XSDir      String
  | XSGetPerms String
  | XSSetPerms String [XSPerm]
  | XSWatch    String String
  | XSUnwatch  String String
#ifdef XS_RESET_WATCHES
  | XSReset
#endif
  | XSTransSt
  | XSTransEnd Bool
  | XSIntro    DomId MFN Port
  | XSRelease  DomId
  | XSGetPath  DomId
  | XSIsDomInt DomId
  | XSResume   DomId
  | XSSetTarg  DomId DomId
  | XSRestrict DomId
  deriving (Show)

requestId :: XenbusRequest -> Word32
requestId (XSRead     _)     = (#const XS_READ)
requestId (XSWrite    _ _)   = (#const XS_WRITE)
requestId (XSMkDir    _)     = (#const XS_MKDIR)
requestId (XSRm       _)     = (#const XS_RM)
requestId (XSDir      _)     = (#const XS_DIRECTORY)
requestId (XSGetPerms _)     = (#const XS_GET_PERMS)
requestId (XSSetPerms _ _)   = (#const XS_SET_PERMS)
requestId (XSWatch    _ _)   = (#const XS_WATCH)
requestId (XSUnwatch  _ _)   = (#const XS_UNWATCH)
#ifdef XS_RESET_WATCHES
requestId  XSReset           = (#const XS_RESET_WATCHES)
#endif
requestId  XSTransSt         = (#const XS_TRANSACTION_START)
requestId (XSTransEnd _)     = (#const XS_TRANSACTION_END)
requestId (XSIntro    _ _ _) = (#const XS_INTRODUCE)
requestId (XSRelease  _)     = (#const XS_RELEASE)
requestId (XSGetPath  _)     = (#const XS_GET_DOMAIN_PATH)
requestId (XSIsDomInt _)     = (#const XS_IS_DOMAIN_INTRODUCED)
requestId (XSResume   _)     = (#const XS_RESUME)
requestId (XSSetTarg  _ _)   = (#const XS_SET_TARGET)
requestId (XSRestrict _)     = (#const XS_RESTRICT)

renderBody :: XenbusRequest -> Put
renderBody (XSRead     str)          =
  renderStr str >> addNull
renderBody (XSWrite    key  val)     =
  renderStr key >> addNull >> renderStr val
renderBody (XSMkDir    str)          =
  renderStr str >> addNull
renderBody (XSRm       str)          =
  renderStr str >> addNull
renderBody (XSDir      str)          =
  renderStr str >> addNull
renderBody (XSGetPerms str)          =
  renderStr str >> addNull
renderBody (XSSetPerms str  perms)   =
  renderStr str >> addNull >> renderPerms perms
renderBody (XSWatch    str  token)   =
  renderStr str >> addNull >> renderStr token >> addNull
renderBody (XSUnwatch  str  token)   =
  renderStr str >> addNull >> renderStr token >> addNull
#ifdef XS_RESET_WATCHES
renderBody  XSReset                  =
  addNull
#endif
renderBody  XSTransSt                =
  addNull
renderBody (XSTransEnd good)         =
  renderStr (if good then "T" else "F") >> addNull
renderBody (XSIntro    dom  mfn prt) =
  renderDom dom  >> addNull >>
  renderMFN mfn  >> addNull >>
  renderPort prt >> addNull
renderBody (XSRelease  dom)          =
  renderDom dom >> addNull
renderBody (XSGetPath  dom)          =
  renderDom dom >> addNull
renderBody (XSIsDomInt dom)          =
  renderDom dom >> addNull
renderBody (XSResume   dom)          =
  renderDom dom >> addNull
renderBody (XSSetTarg  dom  tdom)    =
  renderDom dom >> addNull >> renderDom tdom >> addNull
renderBody (XSRestrict dom)          =
  renderDom dom >> addNull

renderStr :: String -> Put
renderStr str = mapM_ putWord8 (map (unCUChar . castCharToCUChar) str)
 where unCUChar (CUChar x) = x

renderPerms :: [XSPerm] -> Put
renderPerms  = mapM_ (\ p -> renderPerm p >> addNull)

renderPerm :: XSPerm -> Put
renderPerm  (WritePerm     d) = renderStr "w" >> renderDom d
renderPerm  (ReadPerm      d) = renderStr "r" >> renderDom d
renderPerm  (ReadWritePerm d) = renderStr "b" >> renderDom d
renderPerm  (NonePerm      d) = renderStr "n" >> renderDom d

renderDom :: DomId -> Put
renderDom d = renderStr (show (fromDomId d :: Word16))

renderMFN :: MFN -> Put
renderMFN f = renderStr (show (fromMFN f))

renderPort :: Port -> Put
renderPort p = renderStr (show (fromPort p :: Word32))

addNull :: Put
addNull = putWord8 0

-- ----------------------------------------------------------------------------

writeRequestData :: Port -> XSRing -> ByteString -> IO ByteString
writeRequestData port ring bstr =
  do cons <- reqConsumed ring
     prod <- reqProduced ring
     systemMB
     when ((prod - cons) > reqRingSize) $ fail "XenStore invariant broke."
     let (buf, len) = getOutputChunk cons prod
     let (writeBuf, leftBuf) = BS.splitAt (fromIntegral len) bstr
     let len' = min (fromIntegral len) (fromIntegral (BS.length writeBuf))
     unless (len == 0) $
       do writeBS writeBuf buf
          systemWMB
          setReqProduced ring (prod + len')
          sendOnPort port
     return leftBuf
 where
  getOutputChunk cons prod =
    let prodMask = prod .&. (reqRingSize - 1)
        maxLen   = reqRingSize - prodMask
        roomLeft = reqRingSize - (prod - cons)
        len      = if roomLeft < maxLen then roomLeft else maxLen
    in (requestRing ring `plusPtr` fromIntegral prodMask, len)

readResponseData :: Port -> XSRing -> IO ByteString
readResponseData port ring = handle printError $
  do cons <- respConsumed ring
     prod <- respProduced ring
     systemMB
     when ((prod - cons) > respRingSize) $ fail "XenStore invariant broke."
     let (buf, len) = getInputChunk cons prod
     if len == 0
        then return BS.empty
        else do systemRMB
                bstr <- packCStringLen (buf, fromIntegral len)
                systemMB
                setRespConsumed ring (cons + len)
                sendOnPort port
                return (BS.fromStrict bstr)
 where
  printError :: SomeException -> IO ByteString
  printError e =
   do writeDebugConsole ("Xenbus: Caught exception: " ++ show e ++ "\n") 
      return BS.empty
  getInputChunk cons prod =
    let consMask = cons .&. (respRingSize - 1)
        maxLen   = respRingSize - consMask
        len      = if (prod - cons) < maxLen then prod - cons else maxLen
    in (responseRing ring `plusPtr` fromIntegral consMask, len)

-- ----------------------------------------------------------------------------

type XSRing = Ptr Word8

reqRingSize :: Word32
reqRingSize  = 1024

respRingSize :: Word32
respRingSize  = 1024

requestRing :: XSRing -> Ptr Word8
requestRing ring = ring `plusPtr` (#offset struct xenstore_domain_interface,req)

responseRing :: XSRing -> Ptr Word8
responseRing r = r `plusPtr` (#offset struct xenstore_domain_interface,rsp)

reqConsumed :: XSRing -> IO Word32
reqConsumed ring = (#peek struct xenstore_domain_interface,req_cons) ring

reqProduced :: XSRing -> IO Word32
reqProduced ring = (#peek struct xenstore_domain_interface,req_prod) ring

setReqProduced :: XSRing -> Word32 -> IO ()
setReqProduced ring v = (#poke struct xenstore_domain_interface,req_prod) ring v

respConsumed :: XSRing -> IO Word32
respConsumed ring = (#peek struct xenstore_domain_interface,rsp_cons) ring

setRespConsumed :: XSRing -> Word32 -> IO ()
setRespConsumed r v = (#poke struct xenstore_domain_interface,rsp_cons) r v

respProduced :: XSRing -> IO Word32
respProduced r = (#peek struct xenstore_domain_interface,rsp_prod) r

-- ----------------------------------------------------------------------------

forkIO_ :: IO () -> IO ()
forkIO_ m = forkIO m >> return ()

writeBS :: ByteString -> Ptr a -> IO ()
writeBS bstr buf = go (BS.toChunks bstr) buf
 where
  go [] _ = return ()
  go (f:rest) p = useAsCStringLen f $
    \ (ptr, len) ->
      do memcpy p (castPtr ptr) len
         go rest (p `plusPtr` len)

foldM' :: a -> [b] -> (a -> b -> IO a) -> IO a
foldM' val0 ls f = foldM f val0 ls 

-- ----------------------------------------------------------------------------

foreign import ccall unsafe "domain_info.h get_xenstore_evtchn"
  get_xenstore_evtchn :: IO Word32

foreign import ccall unsafe "domain_info.h get_xenstore_mfn"
  get_xenstore_mfn :: IO Word

foreign import ccall unsafe "string.h memcpy"
  memcpy :: Ptr a -> Ptr a -> Int -> IO ()