tricu

Reactor.hs (14387B)

---

 {-# LANGUAGE ForeignFunctionInterface #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 module Reactor where
 
 import qualified CompilerProbe as C
 import qualified BrowserRuntime as R
 import qualified BrowserModules as M
 import Data.Array (Array, (!))
 import Text.Read (readMaybe)
 import Runtime.Tree
 import Runtime.Value
 import Control.Exception (SomeException, SomeAsyncException, catch, evaluate, fromException, displayException)
 import Control.Monad (unless)
 import Data.IORef
 import Data.List (nub)
 import qualified Data.ByteString as BS
 import qualified Data.Text as Text
 import qualified Data.Text.Encoding as Text
 import Foreign.Ptr (Ptr, castPtr)
 import Foreign.C.Types (CInt(..))
 import Foreign.Marshal.Utils (copyBytes)
 import System.IO.Unsafe (unsafePerformIO)
 import System.Mem (performGC)
 import System.IO.Error (isUserError, ioeGetErrorString)
 
 {-# NOINLINE context #-}
 context :: IORef (Maybe (T,T))
 context = unsafePerformIO (newIORef Nothing)
 {-# NOINLINE current #-}
 current :: IORef [(String,T)]
 current = unsafePerformIO (newIORef [])
 {-# NOINLINE compilerLibrary #-}
 compilerLibrary :: IORef M.Bindings
 compilerLibrary = unsafePerformIO (newIORef [])
 {-# NOINLINE expressionBindings #-}
 expressionBindings :: IORef M.Bindings
 expressionBindings = unsafePerformIO (newIORef [])
 {-# NOINLINE pendingModule #-}
 pendingModule :: IORef (Maybe (M.Stage,[(String,M.Bindings)]))
 pendingModule = unsafePerformIO (newIORef Nothing)
 {-# NOINLINE response #-}
 response :: IORef BS.ByteString
 response = unsafePerformIO (newIORef BS.empty)
 
 -- One owned response at a time; no tree handles or Haskell heap addresses cross
 -- the boundary. JS copies it out before another request. Inputs are copied in.
 reply :: String -> IO ()
 reply s = do
   let bounded=take (16777216+1) s
   unless (length bounded<=16777216) (fail "response limit exceeded")
   let bytes=Text.encodeUtf8 (Text.pack bounded)
   unless (BS.length bytes<=16777216) (fail "response byte limit exceeded")
   _<-evaluate (BS.length bytes)
   writeIORef response bytes
 
 guardRequest :: IO () -> IO CInt
 guardRequest action = (action >> pure 0) `catch` \(e::SomeException) -> do
   let message=case fromException e of
         Just ioError | isUserError ioError -> ioeGetErrorString ioError
         _ -> displayException e
   reply ("error\n"++take 2048 message)
   pure (case (fromException e :: Maybe SomeAsyncException) of Just _->2;Nothing->1)
 
 input :: Ptr () -> CInt -> IO String
 input ptr len = do
   unless (len>=0 && len<=16777216) (fail "input byte limit exceeded")
   bytes<-BS.packCStringLen (castPtr ptr,fromIntegral len)
   either (fail . show) (pure . Text.unpack) (Text.decodeUtf8' bytes)
 
 initialize :: IO CInt
 initialize = guardRequest $ do
   existing<-readIORef context
   case existing of Just _->fail "already initialized"; Nothing->pure ()
   (compiler,env,library)<-C.loadLibrary "/data"
   writeIORef context (Just (compiler,env))
   writeIORef compilerLibrary library
   reply "ok\nready"
 
 compileSource :: Ptr () -> CInt -> IO CInt
 compileSource ptr len = guardRequest $ do
   source<-input ptr len
   ctx<-readIORef context
   (compiler,env)<-maybe (fail "not initialized") pure ctx
   compileWith compiler env source
 
 compileWith :: T -> T -> String -> IO ()
 compileWith compiler env source = do
   requireIdle
   unless(not(null source)&&length source<=16384)(fail "source scalar limit exceeded")
   let argument=ofString source
   _<-evaluate(C.force argument)
   defs<-case C.definitions(C.compile compiler env argument) of
     Left message->fail message
     Right values->pure values
   let names=map fst(take 17 defs)
   unless(not(null names)&&length names<=16&&all (\n->not(null n)&&length n<=128) names&&length(nub names)==length names)
     (fail "invalid compiler export names")
   reply("ok\n"++unlines names)
   writeIORef current defs
   writeIORef expressionBindings defs
   writeIORef pendingModule Nothing
   writeIORef ioResult Nothing
 
 observe :: Ptr () -> CInt -> IO CInt
 observe ptr len = guardRequest $ do
   request<-input ptr len
   (name,args)<-case lines request of []->fail "missing export"; n:as->pure(n,as)
   unless (length args<=16 && all ((<=50000).length) args) (fail "argument limit exceeded")
   defs<-readIORef current
   term<-maybe (fail "missing export") pure (lookup name defs)
   let result=foldl apply term (map C.fromTernary args)
   reply ("ok\n"++C.ternary result)
 
 dump :: IO CInt
 dump = guardRequest $ do
   defs<-readIORef current
   reply ("ok\n"++unlines [name++"\t"++C.ternary term | (name,term)<-defs])
 
 reset :: IO CInt
 reset = guardRequest $ do
   writeIORef current []
   writeIORef expressionBindings []
   writeIORef pendingModule Nothing
   writeIORef interaction Nothing
   writeIORef ioResult Nothing
   writeIORef response BS.empty
   performGC
   reply "ok\nreset"
 
 collect :: IO CInt
 collect = guardRequest (performGC >> reply "ok\ncollected")
 
 responseSize :: IO CInt
 responseSize = fromIntegral . BS.length <$> readIORef response
 responseCopy :: Ptr () -> CInt -> IO CInt
 responseCopy ptr capacity = do
   bytes<-readIORef response
   if fromIntegral capacity<BS.length bytes then pure (-1) else do
     BS.useAsCStringLen bytes $ \(src,len)->copyBytes ptr (castPtr src) len
     writeIORef response BS.empty
     pure (fromIntegral (BS.length bytes))
 
 
 {-# NOINLINE interaction #-}
 interaction :: IORef (Maybe (Integer,R.Pending))
 interaction = unsafePerformIO (newIORef Nothing)
 {-# NOINLINE nextToken #-}
 nextToken :: IORef Integer
 nextToken = unsafePerformIO (newIORef 0)
 {-# NOINLINE ioResult #-}
 ioResult :: IORef (Maybe T)
 ioResult = unsafePerformIO (newIORef Nothing)
 
 requireIdle :: IO ()
 requireIdle = do
   active<-readIORef interaction
   case active of Nothing->pure ();Just _->fail "IO interaction is pending"
 
 emitEvent :: R.Pending -> IO ()
 emitEvent pending=do
   token<-(+1) <$> readIORef nextToken
   value<-R.event token pending
   reply ("ok\n"++R.json value)
   writeIORef nextToken token
   case pending of
     R.Pending (R.Done value)->do
       writeIORef ioResult (Just value)
       writeIORef interaction Nothing
     _->writeIORef interaction (Just(token,pending))
 
 select :: BS.ByteString -> BS.ByteString -> [BS.ByteString] -> IO T
 select nameBytes dag roots=do
   name<-R.utf8 nameBytes
   defs<-readIORef current
   worker<-maybe (fail "missing export") pure (lookup name defs)
   unless(length roots<=16)(fail "argument limit exceeded")
   (trees,n)<-R.loadDAG dag
   indices<-mapM (R.index n) roots
   pure(foldl apply worker [trees!i | i<-indices])
 
 bindings :: Array Int T -> Int -> [BS.ByteString] -> IO [(String,T)]
 bindings trees n entries=do
   unless(length entries<=8192&&even(length entries))(fail "invalid binding table")
   let pairs []=pure []
       pairs (nameBytes:rootBytes:rest)=do
         name<-R.utf8 nameBytes
         unless(not(null name)&&length name<=4096)(fail "invalid binding name")
         root<-R.index n rootBytes
         ((name,trees!root):) <$> pairs rest
       pairs _=fail "invalid binding table"
   defs<-pairs entries
   unless(length(nub(map fst defs))==length defs)(fail "duplicate binding name")
   pure defs
 
 request :: Ptr () -> CInt -> IO CInt
 request ptr len = guardRequest $ do
   unless(len>=0 && len<=16777216)(fail "input byte limit exceeded")
   bytes<-BS.packCStringLen(castPtr ptr,fromIntegral len)
   fs<-R.fields bytes
   (op,arguments)<-case fs of
     tag:rest->do name<-R.utf8 tag;pure(name,rest)
     _->fail "missing operation"
   case (op,arguments) of
     ("module-header",[sourceBytes])->do
       requireIdle
       initialized<-readIORef context
       case initialized of Nothing->fail "compiler not initialized"; Just _->pure ()
       library<-readIORef compilerLibrary
       source<-R.utf8 sourceBytes
       staged<-M.stage library source
       reply("ok\n"++R.json(M.stageJSON staged))
     ("module-stage",[sourceBytes])->do
       requireIdle
       writeIORef current []
       writeIORef expressionBindings []
       writeIORef pendingModule Nothing
       writeIORef ioResult Nothing
       initialized<-readIORef context
       case initialized of Nothing->fail "compiler not initialized"; Just _->pure ()
       library<-readIORef compilerLibrary
       source<-R.utf8 sourceBytes
       staged<-M.stage library source
       reply("ok\n"++R.json(M.stageJSON staged))
       writeIORef pendingModule (Just (staged,[]))
     ("module-dependency",refBytes:dag:entries)->do
       requireIdle
       saved<-readIORef pendingModule
       (staged,modules)<-maybe (fail "No pending module") pure saved
       ref<-R.utf8 refBytes
       unless(length ref==64 && all (`elem` ("0123456789abcdef"::String)) ref && not(ref `elem` map fst modules))
         (fail "Invalid or duplicate module reference")
       (trees,n)<-R.loadDAG dag
       values<-bindings trees n entries
       writeIORef pendingModule (Just(staged,modules++[(ref,values)]))
       reply "ok\nloaded"
     ("module-compile",[])->do
       requireIdle
       saved<-readIORef pendingModule
       (staged,modules)<-maybe (fail "No pending module") pure saved
       ctx<-readIORef context
       (compiler,_)<-maybe (fail "compiler not initialized") pure ctx
       library<-readIORef compilerLibrary
       (public,scope)<-M.compile library compiler staged modules
       reply("ok\n"++unlines(map fst public))
       writeIORef current public
       writeIORef expressionBindings scope
       writeIORef pendingModule Nothing
     ("module-scope",[])->do
       scope<-readIORef expressionBindings
       reply("ok\n"++R.json(R.A(map (R.S . fst) scope)))
     ("module-abort",[])->do
       requireIdle
       writeIORef current []
       writeIORef expressionBindings []
       writeIORef pendingModule Nothing
       writeIORef ioResult Nothing
       reply "ok\naborted"
     ("compile",sourceBytes:dag:entries)->do
       requireIdle
       ctx<-readIORef context
       (compiler,_)<-maybe (fail "compiler not initialized") pure ctx
       source<-R.utf8 sourceBytes
       (trees,n)<-R.loadDAG dag
       defs<-bindings trees n entries
       compileWith compiler (ofList [Fork (ofString name) tree | (name,tree)<-defs]) source
     -- Compile against existing named trees without serializing, evaluating or
     -- replacing them. The temporary result lives only in this call/IO session.
     ("expression",[nameBytes,sourceBytes,modeBytes,codec])->do
       requireIdle
       ctx<-readIORef context
       (compiler,_)<-maybe (fail "compiler not initialized") pure ctx
       name<-R.utf8 nameBytes
       source<-R.utf8 sourceBytes
       mode<-R.utf8 modeBytes
       kind<-R.utf8 codec
       unless(not(null source)&&length source<=16384)(fail "source scalar limit exceeded")
       defs<-readIORef expressionBindings
       let env=ofList [Fork (ofString key) tree | (key,tree)<-defs]
       values<-either fail pure (C.definitions(C.compile compiler env (ofString source)))
       term<-case values of
         [(key,value)] | key==name->pure value
         _->fail "Enter one expression, not a source unit"
       case mode of
         "pure"->do
           value<-R.view kind term
           reply("ok\n"++R.json value)
         "io"->do
           phase<-R.beginIO term
           emitEvent(R.Pending phase)
           writeIORef ioResult Nothing
         _->fail "invalid expression mode"
     ("load",dag:entries)->do
       requireIdle
       (trees,n)<-R.loadDAG dag
       unless(not(null entries)&&length entries<=8192&&even(length entries))(fail "invalid export table")
       defs<-bindings trees n entries
       reply("ok\n"++R.json(R.A(map (R.S . fst) defs)))
       writeIORef current defs
       writeIORef expressionBindings defs
       writeIORef pendingModule Nothing
       writeIORef ioResult Nothing
     ("evaluate",name:codec:dag:roots)->do
       kind<-R.utf8 codec
       term<-select name dag roots
       value<-R.view kind term
       reply("ok\n"++R.json value)
     ("preview",name:limit:dag:roots)->do
       size<-R.index 257 limit
       term<-select name dag roots
       value<-R.preview size term
       reply("ok\n"++R.json value)
     ("module-snapshot",[])->do
       requireIdle
       defs<-readIORef current
       unless(not(null defs) && length defs<=4096)(fail "No compiled module")
       value<-R.preview 100000 (foldr (Fork . snd) Leaf defs)
       reply("ok\n"++R.json(R.O [("names",R.A(map (R.S . fst) defs)),("snapshot",value)]))
     ("snapshot",name:limit:dag:roots)->do
       size<-R.index 100001 limit
       term<-select name dag roots
       value<-R.preview size term
       reply("ok\n"++R.json value)
     ("io-start",name:dag:roots)->do
       requireIdle
       term<-select name dag roots
       phase<-R.beginIO term
       emitEvent(R.Pending phase)
       writeIORef ioResult Nothing
     ("io-resume",tokenBytes:valueBytes:[])->do
       tokenText<-R.utf8 tokenBytes
       supplied<-maybe (fail "invalid resume token") pure (readMaybe tokenText)
       saved<-readIORef interaction
       (token,pending)<-maybe (fail "no IO interaction") pure saved
       unless(supplied==token)(fail "stale IO resume token")
       value<-R.utf8 valueBytes
       next<-R.resumeIO pending value
       emitEvent next
     ("io-cancel",[])->do
       writeIORef interaction Nothing
       writeIORef ioResult Nothing
       reply("ok\n"++R.json(R.O[("kind",R.S "cancelled")]))
     ("io-snapshot",[limit])->do
       saved<-readIORef ioResult
       result<-maybe (fail "no completed IO result") pure saved
       size<-R.index 100001 limit
       value<-R.preview size result
       reply("ok\n"++R.json value)
     ("io-view",[codec])->do
       saved<-readIORef ioResult
       result<-maybe (fail "no completed IO result") pure saved
       kind<-R.utf8 codec
       value<-R.view kind result
       reply("ok\n"++R.json value)
     _->fail "invalid operation fields"
 
 initializeEmpty :: IO CInt
 initializeEmpty = guardRequest (reply "ok\nready")
 
 foreign export ccall "tricu_request" request :: Ptr () -> CInt -> IO CInt
 foreign export ccall "tricu_initialize_empty" initializeEmpty :: IO CInt
 
 foreign export ccall "tricu_initialize" initialize :: IO CInt
 foreign export ccall "tricu_compile" compileSource :: Ptr () -> CInt -> IO CInt
 foreign export ccall "tricu_observe" observe :: Ptr () -> CInt -> IO CInt
 foreign export ccall "tricu_dump" dump :: IO CInt
 foreign export ccall "tricu_reset" reset :: IO CInt
 foreign export ccall "tricu_collect" collect :: IO CInt
 foreign export ccall "tricu_response_size" responseSize :: IO CInt
 foreign export ccall "tricu_response_copy" responseCopy :: Ptr () -> CInt -> IO CInt