Rewrite the live view system to be explicitly incremental by dealing in diffs.

Previously we achieved incremental updates using a sort of "rebuild the world
but memoize all the things" strategy. I've come around to the idea that it's
better to deal with updates explicitly.

The new pattern could be called "defunctionalized state". You maintain a state
but instead of being able to update it arbitrarily (e.g. using `put` in the
state monad) you can only update it in a certain number of predefined ways,
expressed as a "diff" data type. These diffs can be sent between servers,
serialized to JSON and sent to the browser and so on. And they're expected to be
monoidal so you can batch a bunch of diffs together for efficiency and even hope
for some cancellation, say of pushes and pops.

The other part of the pattern is to build "state servers" that serve the current
state using these diffs. When you subscribe to them they send you a current
state and an endless stream of updates. You can wire the servers together to
achieve an incrementally updated pipeline.

This worked out really well! I was even able to add a new feature: tracking the
current state of a cell (pending/running/complete) in the web view. And the
whole thing is much snappier because every piece of it is just working with
updates instead of full states.

Author: dougalm

dex.cabal

@@ -58,6 +58,7 @@ library
                      , Generalize
                      , Imp
                      , ImpToLLVM
+                     , IncState
                      , Inference
                      , Inline
                      , IRVariants
@@ -72,6 +73,7 @@ library
                      , Linearize
                      , Logging
                      , Lower
+                     , MonadUtil
                      , MTL1
                      , Name
                      , Occurrence
@@ -103,7 +105,6 @@ library
   if flag(live)
     exposed-modules:   Actor
                      , Live.Eval
-                     , Live.Terminal
                      , Live.Web
                      , RenderHtml
   other-modules:       Paths_dex
@@ -136,6 +137,7 @@ library
                        -- Serialization
                      , aeson
                      , store
+                     , time
                        -- Floating-point pedanticness (correcting for GHC < 9.2.2)
                      , floating-bits
   if flag(live)

src/dex.hs

@@ -22,15 +22,15 @@ import qualified Data.Text as T
 import qualified Data.Text.Encoding as T
 import qualified Data.Map.Strict as M
 
-import PPrint (toJSONStr, printResult)
+import PPrint (resultAsJSON, printResult)
 import TopLevel
 import Err
 import Name
 import AbstractSyntax (parseTopDeclRepl)
 import ConcreteSyntax (keyWordStrs, preludeImportBlock)
 #ifdef DEX_LIVE
 import RenderHtml
-import Live.Terminal (runTerminal)
+-- import Live.Terminal (runTerminal)
 import Live.Web (runWeb)
 #endif
 import Core
@@ -84,14 +84,10 @@ runMode evalMode opts = case evalMode of
           _ -> liftIO $ putStrLn $ pprint result
   ClearCache -> clearCache
 #ifdef DEX_LIVE
-  -- These are broken if the prelude produces any arrays because the blockId
-  -- counter restarts at zero. TODO: make prelude an implicit import block
   WebMode    fname -> do
     env <- loadCache
     runWeb fname opts env
-  WatchMode  fname -> do
-    env <- loadCache
-    runTerminal fname opts env
+  WatchMode  _ -> error "not implemented"
 #endif
 
 printIncrementalSource :: DocFmt -> SourceBlock -> IO ()
@@ -106,7 +102,7 @@ printIncrementalSource fmt sb = case fmt of
 printIncrementalResult :: DocFmt -> Result -> IO ()
 printIncrementalResult fmt result = case fmt of
   ResultOnly -> case pprint result of [] -> return (); msg -> putStrLn msg
-  JSONDoc    -> case toJSONStr result of "{}" -> return (); s -> putStrLn s
+  JSONDoc    -> case resultAsJSON result of "{}" -> return (); s -> putStrLn s
   TextDoc    -> do
     isatty <- queryTerminal stdOutput
     putStr $ printResult isatty result

src/lib/Actor.hs

@@ -1,70 +1,229 @@
--- Copyright 2022 Google LLC
+-- Copyright 2023 Google LLC
 --
 -- Use of this source code is governed by a BSD-style
 -- license that can be found in the LICENSE file or at
 -- https://developers.google.com/open-source/licenses/bsd
 
-module Actor (PChan, sendPChan, sendOnly, subChan,
-              Actor, runActor, spawn,
-              LogServerMsg (..), logServer) where
+{-# LANGUAGE UndecidableInstances #-}
 
-import Control.Concurrent (Chan, forkIO, newChan, readChan, ThreadId, writeChan)
-import Control.Monad.State.Strict
+module Actor (
+  ActorM, Actor (..), launchActor, send, selfMailbox, messageLoop,
+  sliceMailbox, SubscribeMsg (..), IncServer, IncServerT, FileWatcher,
+  StateServer, flushDiffs, handleSubscribeMsg, subscribe, subscribeIO, sendSync,
+  runIncServerT, launchFileWatcher
+  ) where
 
-import Util (onFst, onSnd)
+import Control.Concurrent
+import Control.Monad
+import Control.Monad.State.Strict hiding (get)
+import Control.Monad.Reader
+import qualified Data.ByteString as BS
+import Data.IORef
+import Data.Text.Encoding qualified as T
+import Data.Text (Text)
+import System.Directory (getModificationTime)
+import GHC.Generics
 
--- Micro-actors.
+import IncState
+import MonadUtil
 
--- In this model, an "actor" is just an IO computation (presumably
--- running on its own Haskell thread) that receives messages on a
--- Control.Concurrent.Chan channel.  The idea is that the actor thread
--- only receives information (or synchronization) from other threads
--- through messages sent on that one channel, and no other thread
--- reads messages from that channel.
+-- === Actor implementation ===
 
--- We start the actor with a two-way view of its input channel so it
--- can subscribe itself to message streams by passing (a send-only
--- view of) it to another actor.
-type Actor a = Chan a -> IO ()
+newtype ActorM msg a = ActorM { runActorM ::  ReaderT (Chan msg) IO a }
+  deriving (Functor, Applicative, Monad, MonadIO)
 
--- We also define a send-only channel type, to help ourselves not
--- accidentally read from channels we aren't supposed to.
-newtype PChan a = PChan { sendPChan :: a -> IO () }
+newtype Mailbox a = Mailbox { sendToMailbox :: a -> IO () }
 
-sendOnly :: Chan a -> PChan a
-sendOnly chan = PChan $ \ !x -> writeChan chan x
+class (Show msg, MonadIO m) => Actor msg m | m -> msg where
+  selfChan :: m (Chan msg)
 
-subChan :: (a -> b) -> PChan b -> PChan a
-subChan f chan = PChan (sendPChan chan . f)
+instance Show msg => Actor msg (ActorM msg) where
+  selfChan = ActorM ask
 
--- Synchronously execute an actor.
-runActor :: Actor a -> IO ()
-runActor actor = newChan >>= actor
+instance Actor msg m => Actor msg (ReaderT r m) where selfChan = lift $ selfChan
+instance Actor msg m => Actor msg (StateT  s m) where selfChan = lift $ selfChan
 
--- Asynchronously launch an actor.  Immediately returns permission to
--- kill that actor and to send it messages.
-spawn :: Actor a -> IO (ThreadId, PChan a)
-spawn actor = do
+send :: MonadIO m => Mailbox msg -> msg -> m ()
+send chan msg = liftIO $ sendToMailbox chan msg
+
+selfMailbox :: Actor msg m => (a -> msg) -> m (Mailbox a)
+selfMailbox asSelfMessage = do
+  chan <- selfChan
+  return $ Mailbox \msg -> writeChan chan (asSelfMessage msg)
+
+launchActor :: MonadIO m => ActorM msg () -> m (Mailbox msg)
+launchActor m = liftIO do
   chan <- newChan
-  tid <- forkIO $ actor chan
-  return (tid, sendOnly chan)
-
--- A log server.  Combines inputs monoidally and pushes incremental
--- updates to subscribers.
-
-data LogServerMsg a = Subscribe (PChan a)
-                    | Publish a
-
-logServer :: Monoid a => Actor (LogServerMsg a)
-logServer self = flip evalStateT (mempty, []) $ forever $ do
-  msg <- liftIO $ readChan self
-  case msg of
-    Subscribe chan -> do
-      curVal <- gets fst
-      liftIO $ chan `sendPChan` curVal
-      modify $ onSnd (chan:)
-    Publish x -> do
-      modify $ onFst (<> x)
-      subscribers <- gets snd
-      mapM_ (liftIO . (`sendPChan` x)) subscribers
+  void $ forkIO $ runReaderT (runActorM m) chan
+  return $ Mailbox \msg -> writeChan chan msg
+
+messageLoop :: Actor msg m => (msg -> m ()) -> m ()
+messageLoop handleMessage = do
+  forever do
+    msg <- liftIO . readChan =<< selfChan
+    handleMessage msg
+
+sliceMailbox :: (b -> a) -> Mailbox a -> Mailbox b
+sliceMailbox f (Mailbox sendMsg) = Mailbox $ sendMsg . f
+
+-- === Promises ===
+
+newtype Promise a = Promise (MVar a)
+newtype PromiseSetter a = PromiseSetter (MVar a)
+
+newPromise :: MonadIO m => m (Promise a, PromiseSetter a)
+newPromise = do
+  v <- liftIO $ newEmptyMVar
+  return (Promise v, PromiseSetter v)
+
+waitForPromise :: MonadIO m => Promise a -> m a
+waitForPromise (Promise v) = liftIO $ readMVar v
+
+setPromise :: MonadIO m => PromiseSetter a -> a -> m ()
+setPromise (PromiseSetter v) x = liftIO $ putMVar v x
+
+-- Message that expects a synchronous reponse
+data SyncMsg msg response = SyncMsg msg (PromiseSetter response)
+
+sendSync :: MonadIO m => Mailbox (SyncMsg msg response) -> msg -> m response
+sendSync mailbox msg = do
+  (result, resultSetter) <- newPromise
+  send mailbox (SyncMsg msg resultSetter)
+  waitForPromise result
+
+
+-- === Diff server ===
+
+data IncServerState s d = IncServerState
+  { subscribers     :: [Mailbox d]
+  , bufferedUpdates :: d
+  , curIncState     :: s }
+  deriving (Show, Generic)
+
+class (Monoid d, MonadIO m) => IncServer s d m | m -> s, m -> d where
+  getIncServerStateRef :: m (IORef (IncServerState s d))
+
+data SubscribeMsg s d = Subscribe (SyncMsg (Mailbox d) s)  deriving (Show)
+
+getIncServerState :: IncServer s d m => m (IncServerState s d)
+getIncServerState = readRef =<< getIncServerStateRef
+
+updateIncServerState :: IncServer s d m => (IncServerState s d -> IncServerState s d) -> m ()
+updateIncServerState f = do
+  ref <- getIncServerStateRef
+  prev <- readRef ref
+  writeRef ref $ f prev
+
+handleSubscribeMsg :: IncServer s d m => SubscribeMsg s d -> m ()
+handleSubscribeMsg (Subscribe (SyncMsg newSub response)) = do
+  flushDiffs
+  updateIncServerState \s -> s { subscribers = newSub : subscribers s }
+  curState <- curIncState <$> getIncServerState
+  setPromise response curState
+
+flushDiffs :: IncServer s d m => m ()
+flushDiffs = do
+  d <- bufferedUpdates <$> getIncServerState
+  updateIncServerState \s -> s { bufferedUpdates = mempty }
+  subs <- subscribers <$> getIncServerState
+  -- TODO: consider testing for emptiness here
+  forM_ subs \sub -> send sub d
+
+type StateServer s d = Mailbox (SubscribeMsg s d)
+
+subscribe :: Actor msg m => (d -> msg) -> StateServer s d -> m s
+subscribe inject server = do
+  updateChannel <- selfMailbox inject
+  sendSync (sliceMailbox Subscribe server) updateChannel
+
+subscribeIO :: StateServer s d -> IO (s, Chan d)
+subscribeIO server = do
+  chan <- newChan
+  let mailbox = Mailbox (writeChan chan)
+  s <- sendSync (sliceMailbox Subscribe server) mailbox
+  return (s, chan)
+
+newtype IncServerT s d m a = IncServerT { runIncServerT' :: ReaderT (Ref (IncServerState s d)) m a }
+  deriving (Functor, Applicative, Monad, MonadIO, Actor msg, FreshNames name, MonadTrans)
+
+instance (MonadIO m, IncState s d) => IncServer s d (IncServerT s d m) where
+  getIncServerStateRef = IncServerT ask
+
+instance (MonadIO m, IncState s d) => DefuncState d (IncServerT s d m) where
+  update d = updateIncServerState \s -> s
+    { bufferedUpdates = bufferedUpdates s <> d
+    , curIncState     = curIncState     s `applyDiff` d}
+
+instance (MonadIO m, IncState s d) => LabelReader (SingletonLabel s) (IncServerT s d m) where
+  getl It = curIncState <$> getIncServerState
+
+runIncServerT :: (MonadIO m, IncState s d) => s -> IncServerT s d m a -> m a
+runIncServerT s cont = do
+  ref <- newRef $ IncServerState [] mempty s
+  runReaderT (runIncServerT' cont) ref
+
+-- === Refs ===
+-- Just a wrapper around IORef lifted to `MonadIO`
+
+type Ref = IORef
+
+newRef :: MonadIO m => a -> m (Ref a)
+newRef = liftIO . newIORef
+
+readRef :: MonadIO m => Ref a -> m a
+readRef = liftIO . readIORef
+
+writeRef :: MonadIO m => Ref a -> a -> m ()
+writeRef ref val = liftIO $ writeIORef ref val
+
+-- === Clock ===
+
+-- Provides a periodic clock signal. The time interval is in microseconds.
+launchClock :: MonadIO m => Int -> Mailbox () -> m ()
+launchClock intervalMicroseconds mailbox =
+  liftIO $ void $ forkIO $ forever do
+    threadDelay intervalMicroseconds
+    send mailbox ()
+
+-- === File watcher ===
+
+type SourceFileContents = Text
+type FileWatcher = StateServer SourceFileContents (Overwrite SourceFileContents)
+
+readFileContents :: MonadIO m => FilePath -> m Text
+readFileContents path = liftIO $ T.decodeUtf8 <$> BS.readFile path
+
+data FileWatcherMsg =
+   ClockSignal_FW ()
+ | Subscribe_FW (SubscribeMsg Text (Overwrite Text))
+   deriving (Show)
+
+launchFileWatcher :: MonadIO m => FilePath -> m FileWatcher
+launchFileWatcher path = sliceMailbox Subscribe_FW <$> launchActor (fileWatcherImpl path)
+
+fileWatcherImpl :: FilePath -> ActorM FileWatcherMsg ()
+fileWatcherImpl path = do
+  initContents <- readFileContents path
+  t0 <- liftIO $ getModificationTime path
+  launchClock 100000 =<< selfMailbox ClockSignal_FW
+  modTimeRef <- newRef t0
+  runIncServerT initContents $ messageLoop \case
+    Subscribe_FW msg -> handleSubscribeMsg msg
+    ClockSignal_FW () -> do
+      tOld <- readRef modTimeRef
+      tNew <- liftIO $ getModificationTime path
+      when (tNew /= tOld) do
+        newContents <- readFileContents path
+        update $ OverwriteWith newContents
+        flushDiffs
+        writeRef modTimeRef tNew
+
+-- === instances ===
+
+instance Show msg => Show (SyncMsg msg response) where
+  show (SyncMsg msg _) = show msg
+
+instance Show (Mailbox a) where
+  show _ = "mailbox"
 
+deriving instance Actor msg m => Actor msg (FreshNameT m)