git.iko.soy/project-m36
on ca59541 help sign in
project-m36/src/lib/ProjectM36/Client.hs
684 lines 36 kB history raw
1{-# LANGUAGE DeriveAnyClass, DeriveGeneric, ScopedTypeVariables #-}
2module ProjectM36.Client
3 (ConnectionInfo(..),
4 Connection(..),
5 Port,
6 Hostname,
7 DatabaseName,
8 ConnectionError,
9 connectProjectM36,
10 close,
11 executeRelationalExpr,
12 executeDatabaseContextExpr,
13 executeDatabaseContextIOExpr,
14 executeGraphExpr,
15 executeSchemaExpr,
16 executeTransGraphRelationalExpr,
17 commit,
18 rollback,
19 typeForRelationalExpr,
20 inclusionDependencies,
21 planForDatabaseContextExpr,
22 processPersistence,
23 currentSchemaName,
24 setCurrentSchemaName,
25 transactionGraphAsRelation,
26 relationVariablesAsRelation,
27 headName,
28 remoteDBLookupName,
29 defaultServerPort,
30 headTransactionId,
31 defaultDatabaseName,
32 defaultRemoteConnectionInfo,
33 defaultHeadName,
34 PersistenceStrategy(..),
35 RelationalExpr,
36 RelationalExprBase(..),
37 DatabaseContextExpr(..),
38 Attribute(..),
39 attributesFromList,
40 createNodeId,
41 createSessionAtCommit,
42 createSessionAtHead,
43 closeSession,
44 addClientNode,
45 RelationCardinality(..),
46 TransactionGraphOperator(..),
47 TransGraphRelationalExpr,
48 TransactionIdLookup(..),
49 TransactionIdHeadBacktrack(..),
50 NodeId(..),
51 Atom(..),
52 Session,
53 SessionId,
54 NotificationCallback,
55 emptyNotificationCallback,
56 EvaluatedNotification(..),
57 atomTypesAsRelation,
58 AttributeExpr,
59 AttributeExprBase(..),
60 TypeConstructor(..),
61 AtomType(..)) where
62import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
63import qualified ProjectM36.Base as B
64import ProjectM36.Error
65import ProjectM36.StaticOptimizer
66import qualified ProjectM36.IsomorphicSchema as Schema
67import Control.Monad.State
68import qualified ProjectM36.RelationalExpression as RE
69import ProjectM36.DatabaseContext (basicDatabaseContext)
70import ProjectM36.TransactionGraph
71import qualified ProjectM36.Transaction as Trans
72import ProjectM36.TransactionGraph.Persist
73import ProjectM36.Attribute hiding (atomTypes)
74import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
75import ProjectM36.Persist (DiskSync(..))
76import ProjectM36.Notifications
77import ProjectM36.Server.RemoteCallTypes
78import qualified ProjectM36.DisconnectedTransaction as Discon
79import ProjectM36.Relation (typesAsRelation)
80import ProjectM36.AtomFunctionBody (initScriptSession, ScriptSession)
81import qualified ProjectM36.Relation as R
82import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
83import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)
84import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
85import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
86import Control.Distributed.Process (NodeId(..))
88import Data.UUID.V4 (nextRandom)
89import Control.Concurrent.STM
90import Data.Word
91import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)
92import Control.Exception (IOException, handle, AsyncException, throwIO, fromException)
93import Control.Concurrent.MVar
94import qualified Data.Map as M
95import Control.Distributed.Process.Serializable (Serializable)
96--import Control.Distributed.Process.Debug
97import qualified STMContainers.Map as STMMap
98import qualified STMContainers.Set as STMSet
99import qualified ProjectM36.Session as Sess
100import ProjectM36.Session
101import ProjectM36.Sessions
102import ListT
103import Data.Binary (Binary)
104import GHC.Generics (Generic)
105import Control.DeepSeq (force)
106import System.IO
107--import Debug.Trace
109type Hostname = String
111type Port = Word16
113type DatabaseName = String
115-- | The type for notifications callbacks in the client. When a registered notification fires due to a changed relational expression evaluation, the server propagates the notifications to the clients in the form of the callback.
116type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()
118-- | The empty notification callback ignores all callbacks.
119emptyNotificationCallback :: NotificationCallback
120emptyNotificationCallback _ _ = pure ()
122type GhcPkgPath = String
124-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'.
125data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
126 RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
128type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
130-- | Used for callbacks from the server when monitored changes have been made.
131data NotificationMessage = NotificationMessage EvaluatedNotifications
132 deriving (Binary, Eq, Show, Generic)
134-- | When a notification is fired, the 'reportExpr' is evaluated in the commit's context, so that is returned along with the original notification.
135data EvaluatedNotification = EvaluatedNotification {
136 notification :: Notification,
137 reportRelation :: Either RelationalError Relation
138 }
139 deriving(Binary, Eq, Show, Generic)
141createNodeId :: Hostname -> Port -> NodeId
142createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0
144defaultServerPort :: Port
145defaultServerPort = 6543
147defaultDatabaseName :: DatabaseName
148defaultDatabaseName = "base"
150defaultHeadName :: HeadName
151defaultHeadName = "master"
153defaultRemoteConnectionInfo :: ConnectionInfo
154defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
156-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
157type ClientNodes = STMSet.Set ProcessId
159data Connection = InProcessConnection PersistenceStrategy ClientNodes Sessions (TVar TransactionGraph) (Maybe ScriptSession) |
160 RemoteProcessConnection LocalNode ProcessId
162-- | There are several reasons why a connection can fail.
163data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
164 IOExceptionError IOException |
165 NoSuchDatabaseByNameError DatabaseName |
166 LoginError
167 deriving (Show, Eq, Generic)
169remoteDBLookupName :: DatabaseName -> String
170remoteDBLookupName = (++) "db-"
172commonLocalNode :: IO (Either ConnectionError LocalNode)
173commonLocalNode = do
174 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
175 case eLocalTransport of
176 Left err -> pure (Left $ IOExceptionError err)
177 Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right
179notificationListener :: NotificationCallback -> Process ()
180notificationListener callback = do
181 --pid <- getSelfPid
182 --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
183 _ <- forever $ do
184 receiveWait [
185 match (\(NotificationMessage eNots) -> do
186 --say $ "NOTIFICATION: " ++ show eNots
187 liftIO $ mapM_ (uncurry callback) (M.toList eNots)
188 )
189 ]
190 --say "LISTENER THREAD EXIT"
191 pure ()
193startNotificationListener :: NotificationCallback -> IO (ProcessId)
194startNotificationListener callback = do
195 eLocalNode <- commonLocalNode
196 case eLocalNode of
197 Left err -> error ("Failed to start local notification listener: " ++ show err)
198 Right localNode -> forkProcess localNode (notificationListener callback)
200createScriptSession :: [String] -> IO (Maybe ScriptSession)
201createScriptSession ghcPkgPaths = do
202 eScriptSession <- initScriptSession ghcPkgPaths
203 case eScriptSession of
204 Left err -> hPutStrLn stderr ("Failed to load scripting engine- scripting disabled: " ++ (show err)) >> pure Nothing --not a fatal error, but the scripting feature must be disabled
205 Right s -> pure (Just s)
207-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.
208connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
209--create a new in-memory database/transaction graph
210connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPaths) = do
211 freshId <- nextRandom
212 let bootstrapContext = basicDatabaseContext
213 freshGraph = bootstrapTransactionGraph freshId bootstrapContext
214 case strat of
215 --create date examples graph for now- probably should be empty context in the future
216 NoPersistence -> do
217 graphTvar <- newTVarIO freshGraph
218 clientNodes <- STMSet.newIO
219 sessions <- STMMap.newIO
220 notificationPid <- startNotificationListener notificationCallback
221 mScriptSession <- createScriptSession ghcPkgPaths
222 let conn = InProcessConnection strat clientNodes sessions graphTvar mScriptSession
223 addClientNode conn notificationPid
224 pure (Right conn)
225 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
226 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
228connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
229 connStatus <- newEmptyMVar
230 eLocalNode <- commonLocalNode
231 notificationListenerPid <- startNotificationListener notificationCallback
232 let dbName = remoteDBLookupName databaseName
233 --putStrLn $ show serverNodeId ++ " " ++ dbName
234 case eLocalNode of
235 Left err -> pure (Left err)
236 Right localNode -> do
237 runProcess localNode $ do
238 mServerProcessId <- whereisRemote serverNodeId dbName
239 case mServerProcessId of
240 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
241 Just serverProcessId -> do
242 loginConfirmation <- call serverProcessId (Login notificationListenerPid)
243 if not loginConfirmation then
244 liftIO $ putMVar connStatus (Left LoginError)
245 else do
246 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection localNode serverProcessId)
247 status <- takeMVar connStatus
248 pure status
250connectPersistentProjectM36 :: PersistenceStrategy ->
251 DiskSync ->
252 FilePath ->
253 TransactionGraph ->
254 NotificationCallback ->
255 [GhcPkgPath] ->
256 IO (Either ConnectionError Connection)
257connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
258 err <- setupDatabaseDir sync dbdir freshGraph
259 case err of
260 Just err' -> return $ Left (SetupDatabaseDirectoryError err')
261 Nothing -> do
262 graph <- transactionGraphLoad dbdir emptyTransactionGraph
263 case graph of
264 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
265 Right graph' -> do
266 tvarGraph <- newTVarIO graph'
267 sessions <- STMMap.newIO
268 clientNodes <- STMSet.newIO
269 mScriptSession <- createScriptSession ghcPkgPaths
270 let conn = InProcessConnection strat clientNodes sessions tvarGraph mScriptSession
271 notificationPid <- startNotificationListener notificationCallback
272 addClientNode conn notificationPid
273 pure (Right conn)
275-- | Create a new session at the transaction id and return the session's Id.
276createSessionAtCommit :: TransactionId -> Connection -> IO (Either RelationalError SessionId)
277createSessionAtCommit commitId conn@(InProcessConnection _ _ _ _ _) = do
278 newSessionId <- nextRandom
279 atomically $ do
280 createSessionAtCommit_ commitId newSessionId conn
281createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)
283createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
284createSessionAtCommit_ commitId newSessionId (InProcessConnection _ _ sessions graphTvar _) = do
285 graph <- readTVar graphTvar
286 case transactionForId commitId graph of
287 Left err -> pure (Left err)
288 Right transaction -> do
289 let freshDiscon = DisconnectedTransaction commitId (Trans.schemas transaction)
290 keyDuplication <- STMMap.lookup newSessionId sessions
291 case keyDuplication of
292 Just _ -> pure $ Left (SessionIdInUseError newSessionId)
293 Nothing -> do
294 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
295 pure $ Right newSessionId
296createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionAtCommit_ called on remote connection"
298-- | Call 'createSessionAtHead' with a transaction graph's head's name to create a new session pinned to that head. This function returns a 'SessionId' which can be used in other function calls to reference the point in the transaction graph.
299createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
300createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar _) = do
301 newSessionId <- nextRandom
302 atomically $ do
303 graph <- readTVar graphTvar
304 case transactionForHead headn graph of
305 Nothing -> pure $ Left (NoSuchHeadNameError headn)
306 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
307createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)
309-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
310addClientNode :: Connection -> ProcessId -> IO ()
311addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"
312addClientNode (InProcessConnection _ clientNodes _ _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)
314-- | Discards a session, eliminating any uncommitted changes present in the session.
315closeSession :: SessionId -> Connection -> IO ()
316closeSession sessionId (InProcessConnection _ _ sessions _ _) = do
317 atomically $ STMMap.delete sessionId sessions
318closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)
319-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.
320close :: Connection -> IO ()
321close (InProcessConnection _ _ sessions _ _) = atomically $ do
322 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
323 pure ()
325close (RemoteProcessConnection localNode serverProcessId) = do
326 runProcessResult localNode $ do
327 call serverProcessId Logout
329--within the database server, we must catch and handle all exception lest they take down the database process- this handling might be different for other use-cases
330--exceptions should generally *NOT* be thrown from any Project:M36 code paths, but third-party code such as AtomFunction scripts could conceivably throw undefined, etc.
332excMaybe :: IO (Maybe RelationalError) -> IO (Maybe RelationalError)
333excMaybe m = handle handler m
334 where
335 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
336 | otherwise = pure (Just (UnhandledExceptionError (show exc)))
338excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
339excEither m = handle handler m
340 where
341 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
342 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
344runProcessResult :: LocalNode -> Process a -> IO a
345runProcessResult localNode proc = do
346 ret <- newEmptyMVar
347 runProcess localNode $ do
348 val <- proc
349 liftIO $ putMVar ret val
350 takeMVar ret
352remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
353remoteCall (InProcessConnection _ _ _ _ _) _ = error "remoteCall called on local connection"
354remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
355 ret <- safeCall serverProcessId arg
356 case ret of
357 Left err -> error (show err)
358 Right ret' -> pure ret'
360sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
361sessionForSessionId sessionId sessions = do
362 maybeSession <- STMMap.lookup sessionId sessions
363 pure $ maybe (Left $ NoSuchSessionError sessionId) Right maybeSession
365schemaForSessionId :: Session -> STM (Either RelationalError Schema)
366schemaForSessionId session = do
367 let sname = schemaName session
368 if sname == defaultSchemaName then
369 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
370 else
371 case M.lookup sname (subschemas session) of
372 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
373 Just schema -> pure (Right schema)
375sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
376sessionAndSchema sessionId sessions = do
377 eSession <- sessionForSessionId sessionId sessions
378 case eSession of
379 Left err -> pure (Left err)
380 Right session -> do
381 eSchema <- schemaForSessionId session
382 case eSchema of
383 Left err -> pure (Left err)
384 Right schema -> pure (Right (session, schema))
386currentSchemaName :: SessionId -> Connection -> IO (Maybe SchemaName)
387currentSchemaName sessionId (InProcessConnection _ _ sessions _ _) = atomically $ do
388 eSession <- sessionForSessionId sessionId sessions
389 case eSession of
390 Left _ -> pure Nothing
391 Right session -> pure (Just (Sess.schemaName session))
392currentSchemaName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
394setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Maybe RelationalError)
395setCurrentSchemaName sessionId (InProcessConnection _ _ sessions _ _) sname = atomically $ do
396 eSession <- sessionForSessionId sessionId sessions
397 case eSession of
398 Left _ -> pure Nothing
399 Right session -> case Sess.setSchemaName sname session of
400 Left err -> pure (Just err)
401 Right newSession -> STMMap.insert newSession sessionId sessions >> pure Nothing
402setCurrentSchemaName sessionId conn@(RemoteProcessConnection _ _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
404-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
405executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
406executeRelationalExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excEither $ atomically $ do
407 eSession <- sessionAndSchema sessionId sessions
408 case eSession of
409 Left err -> pure $ Left err
410 Right (session, schema) -> case Schema.processRelationalExprInSchema schema expr of
411 Left err -> pure (Left err)
412 Right expr' -> case evalState (RE.evalRelationalExpr expr') (Sess.concreteDatabaseContext session) of
413 Left err -> pure (Left err)
414 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
415executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
417-- | Execute a database context expression in the context of the session and connection. Database expressions modify the current session's disconnected transaction but cannot modify the transaction graph.
418executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Maybe RelationalError)
419executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excMaybe $ atomically $ do
420 eSession <- sessionAndSchema sessionId sessions
421 case eSession of
422 Left err -> pure $ Just err
423 Right (session, schema) -> do
424 case Schema.processDatabaseContextExprInSchema schema expr of
425 Left err -> pure (Just err)
426 Right expr' -> case runState (RE.evalContextExpr expr') (Sess.concreteDatabaseContext session) of
427 (Just err,_) -> return $ Just err
428 (Nothing, context') -> do
429 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas
430 newSchemas = Schemas context' (Sess.subschemas session)
431 newSession = Session newDiscon (Sess.schemaName session)
432 STMMap.insert newSession sessionId sessions
433 pure Nothing
435executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
437-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr` modify the DatabaseContext but cannot be purely implemented.
438--this is almost completely identical to executeDatabaseContextExpr above
439executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Maybe RelationalError)
440executeDatabaseContextIOExpr sessionId (InProcessConnection _ _ sessions _ scriptSession) expr = excMaybe $ do
441 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
442 case eSession of
443 Left err -> pure $ Just err
444 Right session -> do
445 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
446 case res of
447 Left err -> pure (Just err)
448 Right context' -> do
449 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas
450 newSchemas = Schemas context' (Sess.subschemas session)
451 newSession = Session newDiscon (Sess.schemaName session)
452 atomically $ STMMap.insert newSession sessionId sessions
453 pure Nothing
454executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
456executeGraphExprSTM_ :: TransactionId -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
457executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do
458 graph <- readTVar graphTvar
459 eSession <- sessionForSessionId sessionId sessions
460 case eSession of
461 Left err -> pure $ Left err
462 Right session -> case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
463 Left err -> pure $ Left err
464 Right (discon', graph') -> do
465 writeTVar graphTvar graph'
466 let newSession = Session discon' (Sess.schemaName session)
467 STMMap.insert newSession sessionId sessions
468 pure $ Right graph'
470-- process notifications for commits
471executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
472executeCommitExprSTM_ oldContext newContext nodes = do
473 let nots = notifications oldContext
474 fireNots = notificationChanges nots oldContext newContext
475 evaldNots = M.map mkEvaldNot fireNots
476 mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) oldContext }
477 pure (evaldNots, nodes)
480-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
481executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
482executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar _) graphExpr = excMaybe $ do
483 freshId <- nextRandom
484 manip <- atomically $ do
485 eSession <- sessionForSessionId sessionId sessions
486 oldGraph <- readTVar graphTvar
487 case eSession of
488 Left err -> pure (Left err)
489 Right session -> do
490 eGraph <- executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar
491 case eGraph of
492 Left err -> pure (Left err)
493 Right newGraph -> do
494 if graphExpr == Commit then
495 case transactionForId (Sess.parentId session) oldGraph of
496 Left err -> pure $ Left err
497 Right previousTrans -> do
498 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
499 nodesToNotify <- toList (STMSet.stream nodes)
500 pure $ Right (evaldNots, nodesToNotify, newGraph)
501 else
502 pure $ Right (M.empty, [], newGraph)
503 case manip of
504 Left err -> return $ Just err
505 Right (notsToFire, nodesToNotify, newGraph) -> do
506 --update filesystem database, if necessary
507 --this should really grab a lock at the beginning of the method to be threadsafe
508 processPersistence strat newGraph
509 sendNotifications nodesToNotify notsToFire --TODO: schema mod
510 return Nothing
511executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
513{-
514commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
515commitSTM_ freshUUID sessionId sessions graph = do
516 mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph
517 pure $ case mErr of
518 Left err -> Just err
519 Right _ -> Nothing
520-}
522-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
523executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
524executeTransGraphRelationalExpr _ (InProcessConnection _ _ _ graphTvar _) tgraphExpr = excEither . atomically $ do
525 graph <- readTVar graphTvar
526 case evalTransGraphRelationalExpr tgraphExpr graph of
527 Left err -> pure (Left err)
528 Right relExpr -> case evalState (RE.evalRelationalExpr relExpr) RE.emptyDatabaseContext of
529 Left err -> pure (Left err)
530 Right rel -> pure (force (Right rel))
531executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _ _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
533executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Maybe RelationalError)
534executeSchemaExpr sessionId (InProcessConnection _ _ sessions _ _) schemaExpr = atomically $ do
535 eSession <- sessionAndSchema sessionId sessions
536 case eSession of
537 Left err -> pure (Just err)
538 Right (session, schema) -> do
539 let subschemas' = subschemas session
540 underlyingRelVarNames = if schemaName session == defaultSchemaName then
541 M.keysSet (relationVariables (Sess.concreteDatabaseContext session))
542 else
543 Schema.isomorphsInRelVarNames (Schema.isomorphs schema)
544 case Schema.evalSchemaExpr schemaExpr underlyingRelVarNames subschemas' of
545 Left err -> pure (Just err)
546 Right newSubschemas -> do
547 --hm- maybe we should start using lenses
548 let discon = Sess.disconnectedTransaction session
549 newSchemas = Schemas (Sess.concreteDatabaseContext session) newSubschemas
550 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas) (Sess.schemaName session)
551 STMMap.insert newSession sessionId sessions
552 pure Nothing
553executeSchemaExpr sessionId conn@(RemoteProcessConnection _ _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
555-- | After modifying a session, 'commit' the transaction to the transaction graph at the head which the session is referencing. This will also trigger checks for any notifications which need to be propagated.
556commit :: SessionId -> Connection -> IO (Maybe RelationalError)
557commit sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Commit
558commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
560sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()
561sendNotifications pids nots = mapM_ sendNots pids
562 where
563 sendNots remoteClientPid = do
564 eLocalNode <- commonLocalNode
565 case eLocalNode of
566 Left err -> error ("Failed to get local node: " ++ show err)
567 Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
569-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.
570rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
571rollback sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Rollback
572rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
574processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
575processPersistence NoPersistence _ = return ()
576processPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph
577processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph
579-- | Return a relation whose type would match that of the relational expression if it were executed. This is useful for checking types and validating a relational expression's types.
580typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
581typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
582typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
584typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
585typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _ _) relExpr = do
586 eSession <- sessionAndSchema sessionId sessions
587 case eSession of
588 Left err -> pure $ Left err
589 Right (session, schema) -> do
590 case Schema.processRelationalExprInSchema schema relExpr of
591 Left err -> pure (Left err)
592 Right relExpr' -> pure $ evalState (RE.typeForRelationalExpr relExpr') (Sess.concreteDatabaseContext session)
594typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
596-- | Return a 'Map' of the database's constraints at the context of the session and connection.
597inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))
598inclusionDependencies sessionId (InProcessConnection _ _ sessions _ _) = do
599 atomically $ do
600 eSession <- sessionAndSchema sessionId sessions
601 case eSession of
602 Left err -> pure $ Left err --TODO: schema application
603 Right (session, schema) -> pure $ Right (B.inclusionDependencies (Sess.concreteDatabaseContext session))
605inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
607-- | Return an optimized database expression which is logically equivalent to the input database expression. This function can be used to determine which expression will actually be evaluated.
608planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
609planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) dbExpr = do
610 atomically $ do
611 eSession <- sessionAndSchema sessionId sessions
612 case eSession of
613 Left err -> pure $ Left err --TODO: schema application
614 Right (session, schema) -> pure $ evalState (applyStaticDatabaseOptimization dbExpr) (Sess.concreteDatabaseContext session)
615planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
617-- | Return a relation which represents the current state of the global transaction graph. The attributes are
618-- * current- boolean attribute representing whether or not the current session references this transaction
619-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
620-- * id- id attribute of the transaction
621-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
622transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
623transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar _) = do
624 atomically $ do
625 eSession <- sessionForSessionId sessionId sessions
626 case eSession of
627 Left err -> pure $ Left err
628 Right session -> do
629 graph <- readTVar tvar
630 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
632transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveTransactionGraph sessionId)
634relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
635relationVariablesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
636 atomically $ do
637 eSession <- sessionAndSchema sessionId sessions
638 case eSession of
639 Left err -> pure (Left err)
640 Right (session, schema) -> do
641 case Schema.relationVariablesInSchema schema (Sess.concreteDatabaseContext session) of
642 Left err -> pure (Left err)
643 Right relvars -> pure $ R.relationVariablesAsRelation relvars
645relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
647-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
648headTransactionId :: SessionId -> Connection -> IO (Maybe TransactionId)
649headTransactionId sessionId (InProcessConnection _ _ sessions _ _) = do
650 atomically $ do
651 eSession <- sessionForSessionId sessionId sessions
652 case eSession of
653 Left _ -> pure Nothing
654 Right session -> pure $ Just (Sess.parentId session)
655headTransactionId sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
657headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)
658headNameSTM_ sessionId sessions graphTvar = do
659 graph <- readTVar graphTvar
660 eSession <- sessionForSessionId sessionId sessions
661 case eSession of
662 Left _ -> pure $ Nothing
663 Right session -> pure $ case transactionForId (Sess.parentId session) graph of
664 Left _ -> Nothing
665 Right parentTrans -> headNameForTransaction parentTrans graph
667-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
668headName :: SessionId -> Connection -> IO (Maybe HeadName)
669headName sessionId (InProcessConnection _ _ sessions graphTvar _) = do
670 atomically (headNameSTM_ sessionId sessions graphTvar)
671headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)
673atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
674atomTypesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
675 atomically $ do
676 eSession <- sessionForSessionId sessionId sessions
677 case eSession of
678 Left err -> pure (Left err)
679 Right session -> do
680 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
681 Left err -> pure (Left err)
682 Right rel -> pure (Right rel)
683atomTypesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)