Simulation of Operational Transformation with Cloud Haskell

cloud-haskell-operational-transformation.hs

{-
  This is a simple simulation of OT with Cloud in which all slaves generate
  and apply random operations. It should work in theory. In practice, however
  I wasn't apply to test it because my installation of distributed-process is
  apparently broken. Specifically, `spawn` doesn't seem to work (I tested it
  with some examples from the Well-Typed blog).

  This code depends on https://github.com/timjb/haskell-operational-transformation.
-}

{-# LANGUAGE TemplateHaskell, OverloadedStrings, DeriveDataTypeable, BangPatterns #-}
{-# OPTIONS_GHC -fno-warn-orphans #-}

import Control.Distributed.Process
import Control.Distributed.Process.Closure
import Control.Distributed.Process.Node (initRemoteTable)
import Control.Distributed.Process.Backend.SimpleLocalnet
import Control.Monad (forM, forM_, liftM, when)
import Data.Binary (Binary (..))
import Data.Text (Text, pack, unpack)
import qualified Data.Text as T
import System.Environment (getArgs, getProgName)
import System.Exit (exitWith, ExitCode (..))
import System.IO (stderr, hPutStrLn)
import Test.QuickCheck.Gen (Gen, choose, oneof, sample', listOf1)
import Test.QuickCheck.Arbitrary (Arbitrary (..))
import Control.Applicative ((<$>))

import Control.OperationalTransformation
import Control.OperationalTransformation.Text
import Control.OperationalTransformation.Server
import Control.OperationalTransformation.Client

instance Binary Text where
  put = put . unpack
  get = fmap pack get

genOperation :: T.Text -> Gen TextOperation
genOperation = liftM TextOperation . gen
  where
    gen "" = oneof [return [], liftM ((:[]) . Insert) (arbitraryText maxLength)]
    gen s = do
      len <- choose (1, min maxLength (T.length s))
      oneof [ liftM (Retain len :) $ gen (T.drop len s)
            , do s2 <- arbitraryText len
                 liftM (Insert s2 :) $ gen s
            , liftM (Delete len :) $ gen (T.drop len s)
            ]
    maxLength = 32
    arbitraryText n = liftM (pack . take n) $ listOf1 arbitrary

client :: (ProcessId, Text, Revision) -> Process ()
client (server, initialDoc, initialRevision) = go initialDoc initialRevision initialClientState
  where
    go :: Text -> Revision -> ClientState TextOperation -> Process ()
    go doc revision clientState = do
      liftIO $ putStrLn "Client started"
      res <- receiveTimeout 1
        [ match $ \() -> case serverAck clientState of
            Nothing -> do
              liftIO $ hPutStrLn stderr "unexpected acknowledgement"
              return (doc, clientState)
            Just (mop, clientState') -> do
              case mop of
                Nothing -> return ()
                Just op -> do
                  me <- getSelfPid
                  send server (revision, op, me)
              return (doc, clientState')
        , match $ \operation -> case applyServer clientState operation of
            Left err -> do
              liftIO $ hPutStrLn stderr err
              return (doc, clientState)
            Right (operation', clientState') -> case apply operation' doc of
              Left err -> do
                liftIO $ hPutStrLn stderr $ "could not apply operation: " ++ err
                return (doc, clientState)
              Right doc' -> return (doc', clientState')
        ]
      case res of
        Just (doc', clientState') -> go doc' (revision + 1) clientState'
        Nothing -> do
          operation <- liftIO $ head <$> sample' (genOperation doc)
          liftIO $ putStrLn $ "generated operation: " ++ show operation
          case applyClient clientState operation of
            Left err -> error $ "this should not happen: " ++ err
            Right (shouldSend, clientState') -> do
              when shouldSend $ do
                me <- getSelfPid
                send server (revision, operation, me)
              case apply operation doc of
                Left err -> error $ "this should not happen: " ++ err
                Right doc' -> go doc' revision clientState'

remotable ['client]

server :: [NodeId] -> Process ()
server slaves = do
  liftIO $ putStrLn "Starting server ..."
  pid <- getSelfPid
  let doc = ""
  clientPids <- forM slaves $ \slave -> do
    liftIO $ putStrLn "Starting client process on slave"
    spawn slave $ $(mkClosure 'client) (pid, doc, 0 :: Integer)
  go clientPids $ initialServerState doc
  where
    go :: [ProcessId] -> ServerState Text TextOperation -> Process ()
    go clientPids serverState = do
      (revision, operation, author) <- expect
      liftIO $ putStrLn "got operation"
      case applyOperation serverState revision operation of
        Left err -> do
          liftIO $ hPutStrLn stderr err
          go clientPids serverState
        Right (operation', serverState') -> do
          let ServerState _ doc _ = serverState'
          liftIO $ print doc
          forM_ clientPids $ \clientPid -> if clientPid == author
            then send clientPid ()
            else send clientPid operation'
          go clientPids serverState'


main :: IO ()
main = do
  args <- getArgs
  case args of
    ["server", host, port] -> do
      backend <- initializeBackend host port initRemoteTable
      startMaster backend server
    ["client", host, port] -> do
      backend <- initializeBackend host port initRemoteTable
      startSlave backend
    _ -> do
      prog <- getProgName
      hPutStrLn stderr $ prog ++ " client|server host port"
      exitWith $ ExitFailure 1