git.iko.soy/project-m36
on 0ff751e help sign in
project-m36/src/lib/ProjectM36/Client.hs
1179 lines 62 kB history raw
1{-# LANGUAGE DeriveAnyClass, DeriveGeneric, ScopedTypeVariables, BangPatterns, MonoLocalBinds #-}
2{-|
3Module: ProjectM36.Client
5Client interface to local and remote Project:M36 databases. To get started, connect with 'connectProjectM36', then run some database changes with 'executeDatabaseContextExpr', and issue queries using 'executeRelationalExpr'.
6-}
7module ProjectM36.Client
8 (ConnectionInfo(..),
9 Connection(..),
10 Port,
11 Hostname,
12 DatabaseName,
13 ConnectionError(..),
14 connectProjectM36,
15 close,
16 closeRemote_,
17 executeRelationalExpr,
18 executeDatabaseContextExpr,
19 executeDatabaseContextIOExpr,
20 executeDataFrameExpr,
21 executeGraphExpr,
22 executeSchemaExpr,
23 executeTransGraphRelationalExpr,
24 commit,
25 rollback,
26 typeForRelationalExpr,
27 inclusionDependencies,
28 ProjectM36.Client.typeConstructorMapping,
29 ProjectM36.Client.databaseContextFunctionsAsRelation,
30 planForDatabaseContextExpr,
31 currentSchemaName,
32 SchemaName,
33 HeadName,
34 setCurrentSchemaName,
35 transactionGraphAsRelation,
36 relationVariablesAsRelation,
37 ProjectM36.Client.atomFunctionsAsRelation,
38 disconnectedTransactionIsDirty,
39 headName,
40 remoteDBLookupName,
41 defaultServerPort,
42 headTransactionId,
43 defaultDatabaseName,
44 defaultRemoteConnectionInfo,
45 defaultHeadName,
46 PersistenceStrategy(..),
47 RelationalExpr,
48 RelationalExprBase(..),
49 DatabaseContextExpr(..),
50 DatabaseContextIOExpr(..),
51 Attribute(..),
52 MergeStrategy(..),
53 attributesFromList,
54 createNodeId,
55 createSessionAtCommit,
56 createSessionAtHead,
57 closeSession,
58 addClientNode,
59 callTestTimeout_,
60 RelationCardinality(..),
61 TransactionGraphOperator(..),
62 ProjectM36.Client.autoMergeToHead,
63 transactionGraph_,
64 disconnectedTransaction_,
65 TransGraphRelationalExpr,
66 TransactionIdLookup(..),
67 TransactionIdHeadBacktrack(..),
68 NodeId(..),
69 Atom(..),
70 Session,
71 SessionId,
72 NotificationCallback,
73 emptyNotificationCallback,
74 EvaluatedNotification(..),
75 atomTypesAsRelation,
76 AttributeExpr,
77 inclusionDependencyForKey,
78 databaseContextExprForUniqueKey,
79 databaseContextExprForForeignKey,
80 createScriptedAtomFunction,
81 AttributeExprBase(..),
82 TypeConstructorBase(..),
83 TypeConstructorDef(..),
84 DataConstructorDef(..),
85 AttributeNamesBase(..),
86 RelVarName,
87 IncDepName,
88 InclusionDependency(..),
89 AttributeName,
90 DF.DataFrame,
91 DF.DataFrameExpr,
92 DF.AttributeOrderExpr,
93 DF.Order(..),
94 RelationalError(..),
95 RequestTimeoutException(..),
96 RemoteProcessDiedException(..),
97 AtomType(..),
98 Atomable(..),
99 TupleExprBase(..),
100 AtomExprBase(..),
101 RestrictionPredicateExprBase(..),
102 withTransaction
103 ) where
104import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
105import qualified ProjectM36.Base as B
106import ProjectM36.Error
107import ProjectM36.Atomable
108import ProjectM36.AtomFunction as AF
109import ProjectM36.StaticOptimizer
110import ProjectM36.Key
111import qualified ProjectM36.DataFrame as DF
112import ProjectM36.DatabaseContextFunction as DCF
113import qualified ProjectM36.IsomorphicSchema as Schema
114import Control.Monad.State
115import Control.Monad.Trans.Reader
116import qualified ProjectM36.RelationalExpression as RE
117import ProjectM36.DatabaseContext (basicDatabaseContext)
118import qualified ProjectM36.TransactionGraph as Graph
119import ProjectM36.TransactionGraph
120import qualified ProjectM36.Transaction as Trans
121import ProjectM36.TransactionGraph.Persist
122import ProjectM36.Attribute
123import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
124import ProjectM36.Persist (DiskSync(..))
125import ProjectM36.FileLock
126import ProjectM36.Notifications
127import ProjectM36.Server.RemoteCallTypes
128import qualified ProjectM36.DisconnectedTransaction as Discon
129import ProjectM36.Relation (typesAsRelation)
130import ProjectM36.ScriptSession (initScriptSession, ScriptSession)
131import qualified ProjectM36.Relation as R
132import qualified ProjectM36.DatabaseContext as DBC
133import Control.Exception.Base
134import GHC.Conc.Sync
136import Network.Transport (Transport(closeTransport))
138#if MIN_VERSION_network_transport_tcp(0,7,0)
139import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr)
140import Network.Transport.TCP.Internal (encodeEndPointAddress)
141#elif MIN_VERSION_network_transport_tcp(0,6,0)
142import Network.Transport.TCP (createTransport, defaultTCPParameters)
143import Network.Transport.TCP.Internal (encodeEndPointAddress)
144#else
145import Network.Transport.TCP (encodeEndPointAddress, createTransport, defaultTCPParameters)
146#endif
148import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
149import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
150import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
151import Data.Either (isRight)
152import Data.UUID.V4 (nextRandom)
153import Data.Word
154import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, NodeId(..), reconnect)
155import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
156import Control.Concurrent.MVar
157import qualified Data.Map as M
158import Control.Distributed.Process.Serializable (Serializable)
159#if MIN_VERSION_stm_containers(1,0,0)
160import qualified StmContainers.Map as StmMap
161import qualified StmContainers.Set as StmSet
162#else
163import qualified STMContainers.Map as StmMap
164import qualified STMContainers.Set as StmSet
165#endif
166import qualified ProjectM36.Session as Sess
167import ProjectM36.Session
168import ProjectM36.Sessions
169import Data.Binary (Binary)
170import GHC.Generics (Generic)
171import Control.DeepSeq (force)
172import System.IO
173import Data.Time.Clock
175type Hostname = String
177type Port = Word16
179type DatabaseName = String
181-- | 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.
182type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()
184-- | The empty notification callback ignores all callbacks.
185emptyNotificationCallback :: NotificationCallback
186emptyNotificationCallback _ _ = pure ()
188type GhcPkgPath = String
190data RemoteProcessDiedException = RemoteProcessDiedException
191 deriving (Show, Eq)
193instance Exception RemoteProcessDiedException
195data RequestTimeoutException = RequestTimeoutException
196 deriving (Show, Eq)
198instance Exception RequestTimeoutException
200-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'. The database can be run within the current process or running remotely via distributed-process.
201data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
202 RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
204type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
206-- | Used for callbacks from the server when monitored changes have been made.
207newtype NotificationMessage = NotificationMessage EvaluatedNotifications
208 deriving (Binary, Eq, Show, Generic)
210-- | When a notification is fired, the 'reportOldExpr' is evaluated in the commit's pre-change context while the 'reportNewExpr' is evaluated in the post-change context and they are returned along with the original notification.
211data EvaluatedNotification = EvaluatedNotification {
212 notification :: Notification,
213 reportOldRelation :: Either RelationalError Relation,
214 reportNewRelation :: Either RelationalError Relation
215 }
216 deriving(Binary, Eq, Show, Generic)
219-- | Create a 'NodeId' for use in connecting to a remote server using distributed-process.
220createNodeId :: Hostname -> Port -> NodeId
221createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 1
223-- | Use this for connecting to remote servers on the default port.
224defaultServerPort :: Port
225defaultServerPort = 6543
227-- | Use this for connecting to remote servers with the default database name.
228defaultDatabaseName :: DatabaseName
229defaultDatabaseName = "base"
231-- | Use this for connecting to remote servers with the default head name.
232defaultHeadName :: HeadName
233defaultHeadName = "master"
235-- | Create a connection configuration which connects to the localhost on the default server port and default server database name. The configured notification callback is set to ignore all events.
236defaultRemoteConnectionInfo :: ConnectionInfo
237defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
239-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
240type ClientNodes = StmSet.Set ProcessId
242-- internal structure specific to in-process connections
243data InProcessConnectionConf = InProcessConnectionConf {
244 ipPersistenceStrategy :: PersistenceStrategy,
245 ipClientNodes :: ClientNodes,
246 ipSessions :: Sessions,
247 ipTransactionGraph :: TVar TransactionGraph,
248 ipScriptSession :: Maybe ScriptSession,
249 ipLocalNode :: LocalNode,
250 ipTransport :: Transport, -- we hold onto this so that we can close it gracefully
251 ipLocks :: Maybe (LockFile, MVar LockFileHash) -- nothing when NoPersistence
252 }
254data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
255 rLocalNode :: LocalNode,
256 rProcessId :: ProcessId, --remote processId
257 rTransport :: Transport --the TCP socket transport
258 }
260data Connection = InProcessConnection InProcessConnectionConf |
261 RemoteProcessConnection RemoteProcessConnectionConf
263-- | There are several reasons why a connection can fail.
264data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
265 IOExceptionError IOException |
266 NoSuchDatabaseByNameError DatabaseName |
267 LoginError
268 deriving (Show, Eq, Generic)
270remoteDBLookupName :: DatabaseName -> String
271remoteDBLookupName = (++) "db-"
273createLocalNode :: IO (LocalNode, Transport)
274createLocalNode = do
275#if MIN_VERSION_network_transport_tcp(0,7,0)
276 eLocalTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters
277#elif MIN_VERSION_network_transport_tcp(0,6,0)
278 eLocalTransport <- createTransport "127.0.0.1" "0" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters
279#else
280 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
281#endif
282 case eLocalTransport of
283 Left err -> error ("failed to create transport: " ++ show err)
284 Right localTransport -> do
285 localNode <- newLocalNode localTransport initRemoteTable
286 pure (localNode, localTransport)
288notificationListener :: NotificationCallback -> Process ()
289notificationListener callback = do
290 --pid <- getSelfPid
291 --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
292 _ <- forever $
293 receiveWait [
294 match (\(NotificationMessage eNots) ->
295 --say $ "NOTIFICATION: " ++ show eNots
296 -- when notifications are thrown, they are not adjusted for the current schema which could be problematic, but we don't have access to the current session here
297 liftIO $ mapM_ (uncurry callback) (M.toList eNots)
298 )
299 ]
300 --say "LISTENER THREAD EXIT"
301 pure ()
303startNotificationListener :: LocalNode -> NotificationCallback -> IO ProcessId
304startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
306createScriptSession :: [String] -> IO (Maybe ScriptSession)
307createScriptSession ghcPkgPaths = do
308 eScriptSession <- initScriptSession ghcPkgPaths
309 case eScriptSession of
310 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
311 Right s -> pure (Just s)
313-- | To create a 'Connection' to a remote or local database, create a 'ConnectionInfo' and call 'connectProjectM36'.
314connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
315--create a new in-memory database/transaction graph
316connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPaths) = do
317 freshId <- nextRandom
318 stamp <- getCurrentTime
319 let bootstrapContext = basicDatabaseContext
320 freshGraph = bootstrapTransactionGraph stamp freshId bootstrapContext
321 case strat of
322 --create date examples graph for now- probably should be empty context in the future
323 NoPersistence -> do
324 graphTvar <- newTVarIO freshGraph
325 clientNodes <- StmSet.newIO
326 sessions <- StmMap.newIO
327 (localNode, transport) <- createLocalNode
328 notificationPid <- startNotificationListener localNode notificationCallback
329 mScriptSession <- createScriptSession ghcPkgPaths
331 let conn = InProcessConnection InProcessConnectionConf {
332 ipPersistenceStrategy = strat,
333 ipClientNodes = clientNodes,
334 ipSessions = sessions,
335 ipTransactionGraph = graphTvar,
336 ipScriptSession = mScriptSession,
337 ipLocalNode = localNode,
338 ipTransport = transport,
339 ipLocks = Nothing}
340 addClientNode conn notificationPid
341 pure (Right conn)
342 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
343 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
345connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
346 connStatus <- newEmptyMVar
347 (localNode, transport) <- createLocalNode
348 let dbName = remoteDBLookupName databaseName
349 --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
350 notificationListenerPid <- startNotificationListener localNode notificationCallback
351 runProcess localNode $ do
352 --even a failed connection retains an open socket!
353 mServerProcessId <- whereisRemote serverNodeId dbName
354 case mServerProcessId of
355 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
356 Just serverProcessId -> do
357 loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
358 if not loginConfirmation then
359 liftIO $ putMVar connStatus (Left LoginError)
360 else
361 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport})
362 takeMVar connStatus
364connectPersistentProjectM36 :: PersistenceStrategy ->
365 DiskSync ->
366 FilePath ->
367 TransactionGraph ->
368 NotificationCallback ->
369 [GhcPkgPath] ->
370 IO (Either ConnectionError Connection)
371connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
372 err <- setupDatabaseDir sync dbdir freshGraph
373 case err of
374 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
375 Right (lockFileH, digest) -> do
376 mScriptSession <- createScriptSession ghcPkgPaths
377 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession
378 case graph of
379 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
380 Right graph' -> do
381 tvarGraph <- newTVarIO graph'
382 sessions <- StmMap.newIO
383 clientNodes <- StmSet.newIO
384 (localNode, transport) <- createLocalNode
385 lockMVar <- newMVar digest
386 let conn = InProcessConnection InProcessConnectionConf {
387 ipPersistenceStrategy = strat,
388 ipClientNodes = clientNodes,
389 ipSessions = sessions,
390 ipTransactionGraph = tvarGraph,
391 ipScriptSession = mScriptSession,
392 ipLocalNode = localNode,
393 ipTransport = transport,
394 ipLocks = Just (lockFileH, lockMVar)
395 }
397 notificationPid <- startNotificationListener localNode notificationCallback
398 addClientNode conn notificationPid
399 pure (Right conn)
401-- | Create a new session at the transaction id and return the session's Id.
402createSessionAtCommit :: Connection -> TransactionId -> IO (Either RelationalError SessionId)
403createSessionAtCommit conn@(InProcessConnection _) commitId = do
404 newSessionId <- nextRandom
405 atomically $ createSessionAtCommit_ commitId newSessionId conn
406createSessionAtCommit conn@(RemoteProcessConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
408createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
409createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
410 let sessions = ipSessions conf
411 graphTvar = ipTransactionGraph conf
412 graph <- readTVar graphTvar
413 case transactionForId commitId graph of
414 Left err -> pure (Left err)
415 Right transaction -> do
416 let freshDiscon = DisconnectedTransaction commitId (Trans.schemas transaction) False
417 keyDuplication <- StmMap.lookup newSessionId sessions
418 case keyDuplication of
419 Just _ -> pure $ Left (SessionIdInUseError newSessionId)
420 Nothing -> do
421 StmMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
422 pure $ Right newSessionId
423createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
425-- | 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.
426createSessionAtHead :: Connection -> HeadName -> IO (Either RelationalError SessionId)
427createSessionAtHead conn@(InProcessConnection conf) headn = do
428 let graphTvar = ipTransactionGraph conf
429 newSessionId <- nextRandom
430 atomically $ do
431 graph <- readTVar graphTvar
432 case transactionForHead headn graph of
433 Nothing -> pure $ Left (NoSuchHeadNameError headn)
434 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
435createSessionAtHead conn@(RemoteProcessConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
437-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
438addClientNode :: Connection -> ProcessId -> IO ()
439addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
440addClientNode (InProcessConnection conf) newProcessId = atomically (StmSet.insert newProcessId (ipClientNodes conf))
442-- | Discards a session, eliminating any uncommitted changes present in the session.
443closeSession :: SessionId -> Connection -> IO ()
444closeSession sessionId (InProcessConnection conf) =
445 atomically $ StmMap.delete sessionId (ipSessions conf)
446closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
448-- | 'close' cleans up the database access connection and closes any relevant sockets.
449close :: Connection -> IO ()
450close (InProcessConnection conf) = do
451 atomically $ do
452 let sessions = ipSessions conf
453#if MIN_VERSION_stm_containers(1,0,0)
454 StmMap.reset sessions
455#else
456 StmMap.deleteAll sessions
457#endif
458 pure ()
459 closeLocalNode (ipLocalNode conf)
460 closeTransport (ipTransport conf)
461 let mLocks = ipLocks conf
462 case mLocks of
463 Nothing -> pure ()
464 Just (lockFileH, _) -> closeLockFile lockFileH
466close conn@(RemoteProcessConnection conf) = do
467 _ <- remoteCall conn Logout :: IO Bool
468 closeLocalNode (rLocalNode conf)
469 closeTransport (rTransport conf)
471--used only by the server EntryPoints
472closeRemote_ :: Connection -> IO ()
473closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
474closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
476 --we need to actually close the localNode's connection to the remote
477--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
478--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.
480excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
481excEither = handle handler
482 where
483 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
484 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
486runProcessResult :: LocalNode -> Process a -> IO a
487runProcessResult localNode proc = do
488 ret <- newEmptyMVar
489 runProcess localNode $ do
490 val <- proc
491 liftIO $ putMVar ret val
492 takeMVar ret
494safeLogin :: Login -> ProcessId -> Process Bool
495safeLogin login procId = do
496 ret <- call procId login
497 case ret of
498 Left (_ :: ServerError) -> pure False
499 Right val -> pure val
501remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
502remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
503remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
504 ret <- safeCall serverProcessId arg
505 case ret of
506 Left err -> error ("server died: " ++ show err)
507 Right ret' -> case ret' of
508 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
509 Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
510 Right val -> pure val
511 where
512 localNode = rLocalNode conf
513 serverProcessId = rProcessId conf
515sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
516sessionForSessionId sessionId sessions =
517 maybe (Left $ NoSuchSessionError sessionId) Right <$> StmMap.lookup sessionId sessions
519schemaForSessionId :: Session -> STM (Either RelationalError Schema)
520schemaForSessionId session = do
521 let sname = schemaName session
522 if sname == defaultSchemaName then
523 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
524 else
525 case M.lookup sname (subschemas session) of
526 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
527 Just schema -> pure (Right schema)
529sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
530sessionAndSchema sessionId sessions = do
531 eSession <- sessionForSessionId sessionId sessions
532 case eSession of
533 Left err -> pure (Left err)
534 Right session -> do
535 eSchema <- schemaForSessionId session
536 case eSchema of
537 Left err -> pure (Left err)
538 Right schema -> pure (Right (session, schema))
540-- | Returns the name of the currently selected isomorphic schema.
541currentSchemaName :: SessionId -> Connection -> IO (Either RelationalError SchemaName)
542currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
543 let sessions = ipSessions conf
544 eSession <- sessionForSessionId sessionId sessions
545 case eSession of
546 Left err -> pure (Left err)
547 Right session -> pure (Right (Sess.schemaName session))
548currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
550-- | Switch to the named isomorphic schema.
551setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
552setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
553 let sessions = ipSessions conf
554 eSession <- sessionForSessionId sessionId sessions
555 case eSession of
556 Left err -> pure (Left err)
557 Right session -> case Sess.setSchemaName sname session of
558 Left err -> pure (Left err)
559 Right newSession -> StmMap.insert newSession sessionId sessions >> pure (Right ())
560setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
562-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
563executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
564executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
565 let sessions = ipSessions conf
566 eSession <- sessionAndSchema sessionId sessions
567 case eSession of
568 Left err -> pure $ Left err
569 Right (session, schema) -> do
570 let expr' = if schemaName session /= defaultSchemaName then
571 Schema.processRelationalExprInSchema schema expr
572 else
573 Right expr
574 case expr' of
575 Left err -> pure (Left err)
576 Right expr'' -> case optimizeRelationalExpr (Sess.concreteDatabaseContext session) expr'' of
577 Left err -> pure (Left err)
578 Right optExpr -> do
579 let evald = runReader (RE.evalRelationalExpr optExpr) (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
580 case evald of
581 Left err -> pure (Left err)
582 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
583executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
585-- | 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.
586executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
587executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
588 let sessions = ipSessions conf
589 eSession <- sessionAndSchema sessionId sessions
590 case eSession of
591 Left err -> pure (Left err)
592 Right (session, schema) -> do
593 let expr' = if schemaName session == defaultSchemaName then
594 Right expr
595 else
596 Schema.processDatabaseContextExprInSchema schema expr
597 case expr' of
598 Left err -> pure (Left err)
599 Right expr'' -> case runState (do
600 eOptExpr <- applyStaticDatabaseOptimization expr''
601 case eOptExpr of
602 Left err -> pure (Left err)
603 Right optExpr ->
604 RE.evalDatabaseContextExpr optExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session)) of
605 (Left err,_) -> pure (Left err)
606 (Right (), (_,_,False)) -> pure (Right ()) --optimization- if nothing was dirtied, nothing to do
607 (Right (), (!context',_,True)) -> do
608 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
609 newSubschemas = Schema.processDatabaseContextExprSchemasUpdate (Sess.subschemas session) expr
610 newSchemas = Schemas context' newSubschemas
611 newSession = Session newDiscon (Sess.schemaName session)
612 StmMap.insert newSession sessionId sessions
613 pure (Right ())
614executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
616-- | Similar to a git rebase, 'autoMergeToHead' atomically creates a temporary branch and merges it to the latest commit of the branch referred to by the 'HeadName' and commits the merge. This is useful to reduce incidents of 'TransactionIsNotAHeadError's but at the risk of merge errors (thus making it similar to rebasing). Alternatively, as an optimization, if a simple commit is possible (meaning that the head has not changed), then a fast-forward commit takes place instead.
617autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
618autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
619 let sessions = ipSessions conf
620 id1 <- nextRandom
621 id2 <- nextRandom
622 id3 <- nextRandom
623 stamp <- getCurrentTime
624 commitLock_ sessionId conf $ \graph -> do
625 eSession <- sessionForSessionId sessionId sessions
626 case eSession of
627 Left err -> pure (Left err)
628 Right session ->
629 case Graph.transactionForHead headName' graph of
630 Nothing -> pure (Left (NoSuchHeadNameError headName'))
631 Just headTrans -> do
632 --attempt fast-forward commit, if possible
633 let graphInfo = if Sess.parentId session == transactionId headTrans then do
634 ret <- Graph.evalGraphOp stamp id1 (Sess.disconnectedTransaction session) graph Commit
635 pure (ret, [id1])
636 else do
637 ret <- Graph.autoMergeToHead stamp (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph
638 pure (ret, [id1,id2,id3])
639 case graphInfo of
640 Left err -> pure (Left err)
641 Right ((discon', graph'), transactionIdsAdded) ->
642 pure (Right (discon', graph', transactionIdsAdded))
643autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
645-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
646--this is almost completely identical to executeDatabaseContextExpr above
647executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Either RelationalError ())
648executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEither $ do
649 let sessions = ipSessions conf
650 scriptSession = ipScriptSession conf
651 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
652 case eSession of
653 Left err -> pure (Left err)
654 Right session -> do
655 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
656 case res of
657 Left err -> pure (Left err)
658 Right context' -> do
659 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
660 newSchemas = Schemas context' (Sess.subschemas session)
661 newSession = Session newDiscon (Sess.schemaName session)
662 atomically $ StmMap.insert newSession sessionId sessions
663 pure (Right ())
664executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
666{-
667executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
668executeGraphExprSTM_ freshId sessionId session sessions graphExpr graph graphTVar= do
669 case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
670 Left err -> do
671 when updateGraphOnError (writeTVar graphTVar graph)
672 pure $ Left err
673 Right (discon', graph') -> do
674 writeTVar graphTVar graph'
675 let newSession = Session discon' (Sess.schemaName session)
676 STMMap.insert newSession sessionId sessions
677 pure $ Right graph'
678-}
680-- process notifications for commits
681executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
682executeCommitExprSTM_ oldContext newContext nodes = do
683 let nots = notifications oldContext
684 fireNots = notificationChanges nots oldContext newContext
685 evaldNots = M.map mkEvaldNot fireNots
686 evalInContext expr ctx = runReader (RE.evalRelationalExpr expr) (RE.mkRelationalExprState ctx)
687 mkEvaldNot notif = EvaluatedNotification { notification = notif,
688 reportOldRelation = evalInContext (reportOldExpr notif) oldContext,
689 reportNewRelation = evalInContext (reportNewExpr notif) newContext}
690 pure (evaldNots, nodes)
692-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
694-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
695-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
696-- perhaps keep hash of data file instead of checking if our head was updated on every write
697executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Either RelationalError ())
698executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
699 let sessions = ipSessions conf
700 freshId <- nextRandom
701 stamp <- getCurrentTime
702 commitLock_ sessionId conf $ \updatedGraph -> do
703 eSession <- sessionForSessionId sessionId sessions
704 case eSession of
705 Left err -> pure (Left err)
706 Right session -> do
707 let discon = Sess.disconnectedTransaction session
708 case evalGraphOp stamp freshId discon updatedGraph graphExpr of
709 Left err -> pure (Left err)
710 Right (discon', graph') -> do
711 --if freshId appears in the graph, then we need to pass it on
712 let transIds = [freshId | isRight (transactionForId freshId graph')]
713 pure (Right (discon', graph', transIds))
714{-
715executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
716 let strat = ipPersistenceStrategy conf
717 clientNodes = ipClientNodes conf
718 sessions = ipSessions conf
719 graphTvar = ipTransactionGraph conf
720 mLockFileH = ipLocks conf
721 lockHandler body = case graphExpr of
722 Commit -> case mLockFileH of
723 Nothing -> body False
724 Just (lockFileH, lockMVar) -> let acquireLocks = do
725 lastWrittenDigest <- takeMVar lockMVar
726 lockFile lockFileH WriteLock
727 latestDigest <- readGraphTransactionIdDigest strat
728 pure (latestDigest /= lastWrittenDigest)
730 releaseLocks _ = do
731 --still holding the lock- get the latest digest
732 gDigest <- readGraphTransactionIdDigest strat
733 unlockFile lockFileH
734 putMVar lockMVar gDigest
735 in bracket acquireLocks releaseLocks body
736 _ -> body False
737 freshId <- nextRandom
738 lockHandler $ \dbWrittenByOtherProcess -> do
739 --if the database file has been updated since we wrote it last, load it before trying to sync our version done- this can result in TransactionNotAHeadErrors
740 --read transaction data and compare to existing graph
741 --in the future, we can detect if updated transaction graph can be safely merged (such as with a transaction on a separate head) (rebase-able commits should force the user to rebase from the client to confirm that the action makes sense)
742 manip <- atomically $ do
743 eSession <- sessionForSessionId sessionId sessions
744 --handle graph update by other process
745 oldGraph <- readTVar graphTvar
746 case eSession of
747 Left err -> pure (Left err)
748 Right session -> do
749 let mScriptSession = ipScriptSession conf
750 dbdir = case strat of
751 MinimalPersistence x -> x
752 CrashSafePersistence x -> x
753 _ -> error "accessing dbdir on non-persisted connection"
754 eRefreshedGraph <- if dbWrittenByOtherProcess then
755 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
756 else
757 pure (Right oldGraph)
758 case eRefreshedGraph of
759 Left err -> pure (Left (DatabaseLoadError err))
760 Right refreshedGraph -> do
761 --snip it
762 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
763 --snip it
764 case eGraph of
765 Left err -> pure (Left err)
766 Right newGraph -> do
767 --handle commit
768 if isCommit graphExpr then do
769 case transactionForId (Sess.parentId session) oldGraph of
770 Left err -> pure $ Left err
771 Right previousTrans -> do
772 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
773 nodesToNotify <- toList (StmSet.stream nodes)
774 pure $ Right (evaldNots, nodesToNotify, newGraph)
775 else
776 pure $ Right (M.empty, [], newGraph)
777 case manip of
778 Left err -> return $ Just err
779 Right (notsToFire, nodesToNotify, newGraph) -> do
780 --update filesystem database, if necessary
781 processTransactionGraphPersistence strat newGraph
782 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
783 pure Nothing
784-}
785executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
787-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
788executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
789executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
790 let graphTvar = ipTransactionGraph conf
791 graph <- readTVar graphTvar
792 case evalTransGraphRelationalExpr tgraphExpr graph of
793 Left err -> pure (Left err)
794 Right relExpr -> case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
795 Left err -> pure (Left err)
796 Right rel -> pure (force (Right rel))
797executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
799-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
800executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
801executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
802 let sessions = ipSessions conf
803 eSession <- sessionAndSchema sessionId sessions
804 case eSession of
805 Left err -> pure (Left err)
806 Right (session, _) -> do
807 let subschemas' = subschemas session
808 case Schema.evalSchemaExpr schemaExpr (Sess.concreteDatabaseContext session) subschemas' of
809 Left err -> pure (Left err)
810 Right (newSubschemas, newContext) -> do
811 --hm- maybe we should start using lenses
812 let discon = Sess.disconnectedTransaction session
813 newSchemas = Schemas newContext newSubschemas
814 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas True) (Sess.schemaName session)
815 StmMap.insert newSession sessionId sessions
816 pure (Right ())
817executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
819-- | After modifying a 'DatabaseContext', '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.
820commit :: SessionId -> Connection -> IO (Either RelationalError ())
821commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
822commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
824sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
825sendNotifications pids localNode nots = mapM_ sendNots pids
826 where
827 sendNots remoteClientPid =
828 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
830-- | Discard any changes made in the current 'Session' and 'DatabaseContext'. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.
831rollback :: SessionId -> Connection -> IO (Either RelationalError ())
832rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
833rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
835-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
836processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
837processTransactionGraphPersistence NoPersistence _ _ = pure ()
838processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()
839processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()
841readGraphTransactionIdDigest :: PersistenceStrategy -> IO LockFileHash
842readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
843readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
844readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
846-- | 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.
847typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
848typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
849typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
851typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
852typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
853 let sessions = ipSessions conf
854 eSession <- sessionAndSchema sessionId sessions
855 case eSession of
856 Left err -> pure $ Left err
857 Right (session, schema) -> do
858 let processed = if schemaName session == defaultSchemaName then
859 Right relExpr
860 else
861 Schema.processRelationalExprInSchema schema relExpr
862 case processed of
863 Left err -> pure (Left err)
864 Right relExpr' -> pure $ runReader (RE.typeForRelationalExpr relExpr') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
866typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
868-- | Return a 'Map' of the database's constraints at the context of the session and connection.
869inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
870inclusionDependencies sessionId (InProcessConnection conf) = do
871 let sessions = ipSessions conf
872 atomically $ do
873 eSession <- sessionAndSchema sessionId sessions
874 case eSession of
875 Left err -> pure $ Left err
876 Right (session, schema) -> do
877 let context = Sess.concreteDatabaseContext session
878 if schemaName session == defaultSchemaName then
879 pure $ Right (B.inclusionDependencies context)
880 else
881 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
883inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
885typeConstructorMapping :: SessionId -> Connection -> IO (Either RelationalError TypeConstructorMapping)
886typeConstructorMapping sessionId (InProcessConnection conf) = do
887 let sessions = ipSessions conf
888 atomically $ do
889 eSession <- sessionAndSchema sessionId sessions
890 case eSession of
891 Left err -> pure $ Left err
892 Right (session, _) -> --warning, no schema support for typeconstructors
893 pure (Right (B.typeConstructorMapping (Sess.concreteDatabaseContext session)))
894typeConstructorMapping sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTypeConstructorMapping sessionId)
896-- | 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.
897planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
898planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
899 let sessions = ipSessions conf
900 atomically $ do
901 eSession <- sessionAndSchema sessionId sessions
902 case eSession of
903 Left err -> pure $ Left err
904 Right (session, _) -> if schemaName session == defaultSchemaName then
905 pure $ evalState (applyStaticDatabaseOptimization dbExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session))
906 else -- don't show any optimization because the current optimization infrastructure relies on access to the base context- this probably underscores the need for each schema to have its own DatabaseContext, even if it is generated on-the-fly
907 pure (Right dbExpr)
909planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
911-- | Return a relation which represents the current state of the global transaction graph. The attributes are
912-- * current- boolean attribute representing whether or not the current session references this transaction
913-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
914-- * id- id attribute of the transaction
915-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
916transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
917transactionGraphAsRelation sessionId (InProcessConnection conf) = do
918 let sessions = ipSessions conf
919 tvar = ipTransactionGraph conf
920 atomically $ do
921 eSession <- sessionForSessionId sessionId sessions
922 case eSession of
923 Left err -> pure $ Left err
924 Right session ->
925 graphAsRelation (Sess.disconnectedTransaction session) <$> readTVar tvar
927transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
929-- | Returns the names and types of the relation variables in the current 'Session'.
930relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
931relationVariablesAsRelation sessionId (InProcessConnection conf) = do
932 let sessions = ipSessions conf
933 atomically $ do
934 eSession <- sessionAndSchema sessionId sessions
935 case eSession of
936 Left err -> pure (Left err)
937 Right (session, schema) -> do
938 let context = Sess.concreteDatabaseContext session
939 if Sess.schemaName session == defaultSchemaName then
940 pure $ R.relationVariablesAsRelation (relationVariables context)
941 else
942 case Schema.relationVariablesInSchema schema context of
943 Left err -> pure (Left err)
944 Right relvars -> pure $ R.relationVariablesAsRelation relvars
946relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
948-- | Returns the names and types of the atom functions in the current 'Session'.
949atomFunctionsAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
950atomFunctionsAsRelation sessionId (InProcessConnection conf) = do
951 let sessions = ipSessions conf
952 atomically $ do
953 eSession <- sessionAndSchema sessionId sessions
954 case eSession of
955 Left err -> pure (Left err)
956 Right (session, _) ->
957 pure (AF.atomFunctionsAsRelation (atomFunctions (concreteDatabaseContext session)))
959atomFunctionsAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomFunctionSummary sessionId)
961databaseContextFunctionsAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
962databaseContextFunctionsAsRelation sessionId (InProcessConnection conf) = do
963 let sessions = ipSessions conf
964 atomically $ do
965 eSession <- sessionAndSchema sessionId sessions
966 case eSession of
967 Left err -> pure (Left err)
968 Right (session, _) ->
969 pure (DCF.databaseContextFunctionsAsRelation (dbcFunctions (concreteDatabaseContext session)))
971databaseContextFunctionsAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveDatabaseContextFunctionSummary sessionId)
973-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
974headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
975headTransactionId sessionId (InProcessConnection conf) = do
976 let sessions = ipSessions conf
977 atomically $ do
978 eSession <- sessionForSessionId sessionId sessions
979 case eSession of
980 Left err -> pure (Left err)
981 Right session -> pure $ Right (Sess.parentId session)
982headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
984headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
985headNameSTM_ sessionId sessions graphTvar = do
986 graph <- readTVar graphTvar
987 eSession <- sessionForSessionId sessionId sessions
988 case eSession of
989 Left err -> pure (Left err)
990 Right session -> case transactionForId (Sess.parentId session) graph of
991 Left err -> pure (Left err)
992 Right parentTrans -> case headNameForTransaction parentTrans graph of
993 Nothing -> pure (Left UnknownHeadError)
994 Just headName' -> pure (Right headName')
996-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
997headName :: SessionId -> Connection -> IO (Either RelationalError HeadName)
998headName sessionId (InProcessConnection conf) = do
999 let sessions = ipSessions conf
1000 graphTvar = ipTransactionGraph conf
1001 atomically (headNameSTM_ sessionId sessions graphTvar)
1002headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
1004-- | Returns a listing of all available atom types.
1005atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
1006atomTypesAsRelation sessionId (InProcessConnection conf) = do
1007 let sessions = ipSessions conf
1008 atomically $ do
1009 eSession <- sessionForSessionId sessionId sessions
1010 case eSession of
1011 Left err -> pure (Left err)
1012 Right session ->
1013 case typesAsRelation (B.typeConstructorMapping (Sess.concreteDatabaseContext session)) of
1014 Left err -> pure (Left err)
1015 Right rel -> pure (Right rel)
1016atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
1018disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
1019disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
1020 let sessions = ipSessions conf
1021 atomically $ do
1022 eSession <- sessionForSessionId sessionId sessions
1023 case eSession of
1024 Left err -> pure (Left err)
1025 Right session ->
1026 pure (Right (isDirty session))
1027disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
1029--used only for testing- we expect this to throw an exception
1030callTestTimeout_ :: SessionId -> Connection -> IO Bool
1031callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
1032callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
1034--used in tests only
1035transactionGraph_ :: Connection -> IO TransactionGraph
1036transactionGraph_ (InProcessConnection conf) = readTVarIO (ipTransactionGraph conf)
1037transactionGraph_ _ = error "remote connection used"
1039--used in tests only
1040disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
1041disconnectedTransaction_ sessionId (InProcessConnection conf) = do
1042 let sessions = ipSessions conf
1043 mSession <- atomically $ StmMap.lookup sessionId sessions
1044 case mSession of
1045 Nothing -> error "No such session"
1046 Just (Sess.Session discon _) -> pure discon
1047disconnectedTransaction_ _ _= error "remote connection used"
1049-- wrap a graph evaluation in file locking
1050commitLock_ :: SessionId ->
1051 InProcessConnectionConf ->
1052 (TransactionGraph ->
1053 STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->
1054 IO (Either RelationalError ())
1055commitLock_ sessionId conf stmBlock = do
1056 let sessions = ipSessions conf
1057 strat = ipPersistenceStrategy conf
1058 mScriptSession = ipScriptSession conf
1059 graphTvar = ipTransactionGraph conf
1060 clientNodes = ipClientNodes conf
1061 mLockFileH = ipLocks conf
1062 lockHandler body = case mLockFileH of
1063 Nothing -> body False
1064 Just (lockFileH, lockMVar) ->
1065 let acquireLocks = do
1066 lastWrittenDigest <- takeMVar lockMVar
1067 lockFile lockFileH WriteLock
1068 latestDigest <- readGraphTransactionIdDigest strat
1069 pure (latestDigest /= lastWrittenDigest)
1070 releaseLocks _ = do
1071 --still holding the lock- get the latest digest
1072 gDigest <- readGraphTransactionIdDigest strat
1073 unlockFile lockFileH
1074 putMVar lockMVar gDigest
1075 in bracket acquireLocks releaseLocks body
1076 manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
1077 eSession <- sessionForSessionId sessionId sessions
1078 --handle graph update by other process
1079 oldGraph <- readTVar graphTvar
1080 case eSession of
1081 Left err -> pure (Left err)
1082 Right session -> do
1083 let dbdir = case strat of
1084 MinimalPersistence x -> x
1085 CrashSafePersistence x -> x
1086 _ -> error "accessing dbdir on non-persisted connection"
1087 --this should also happen for non-commit expressions
1088 eRefreshedGraph <- if dbWrittenByOtherProcess then
1089 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
1090 else
1091 pure (Right oldGraph)
1092 case eRefreshedGraph of
1093 Left err -> pure (Left (DatabaseLoadError err))
1094 Right refreshedGraph -> do
1095 eGraph <- stmBlock refreshedGraph
1096 case eGraph of
1097 Left err -> pure (Left err)
1098 Right (discon', graph', transactionIdsToPersist) -> do
1099 writeTVar graphTvar graph'
1100 let newSession = Session discon' (Sess.schemaName session)
1101 StmMap.insert newSession sessionId sessions
1102 case transactionForId (Sess.parentId session) oldGraph of
1103 Left err -> pure $ Left err
1104 Right previousTrans ->
1105 if not (Prelude.null transactionIdsToPersist) then do
1106 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
1107 nodesToNotify <- stmSetToList nodes
1108 pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)
1109 else pure (Right (M.empty, [], graph', []))
1111 --handle notification firing
1112 case manip of
1113 Left err -> pure (Left err)
1114 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
1115 --update filesystem database, if necessary
1116 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1117 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1118 pure (Right ())
1120writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
1121writeDisconAndGraph_ graphTvar sessionId session sessions discon graph = do
1122 writeTVar graphTvar graph
1123 let newSession = Session discon (Sess.schemaName session)
1124 StmMap.insert newSession sessionId sessions
1127-- | Runs an IO monad, commits the result when the monad returns no errors, otherwise, rolls back the changes and the error.
1128withTransaction :: SessionId -> Connection -> IO (Either RelationalError a) -> IO (Either RelationalError ()) -> IO (Either RelationalError a)
1129withTransaction sessionId conn io successFunc = bracketOnError (pure ()) (const do_rollback) block
1130 where
1131 do_rollback = rollback sessionId conn
1132 block _ = do
1133 eErr <- io
1134 case eErr of
1135 Left err -> do
1136 _ <- do_rollback
1137 pure (Left err)
1138 Right val -> do
1139 eIsDirty <- disconnectedTransactionIsDirty sessionId conn
1140 case eIsDirty of
1141 Left err -> pure (Left err)
1142 Right dirty ->
1143 if dirty then do
1144 res <- successFunc
1145 case res of
1146 Left err -> pure (Left err)
1147 Right _ -> pure (Right val)
1148 else -- no updates executed, so don't create a commit
1149 pure (Right val)
1151executeDataFrameExpr :: SessionId -> Connection -> DF.DataFrameExpr -> IO (Either RelationalError DF.DataFrame)
1152executeDataFrameExpr sessionId conn@(InProcessConnection _) dfExpr = do
1153 eRel <- executeRelationalExpr sessionId conn (DF.convertExpr dfExpr)
1154 case eRel of
1155 Left err -> pure (Left err)
1156 Right rel -> do
1157 let relAttrs = R.attributes rel
1158 attrName (DF.AttributeOrderExpr name _) = name
1159 order (DF.AttributeOrderExpr _ ord) = ord
1160 orders = map order (DF.orderExprs dfExpr)
1161 attributeForName' = flip attributeForName relAttrs
1162 attrNames = map attrName (DF.orderExprs dfExpr)
1163 verified = forM attrNames attributeForName'
1164 case verified of
1165 Left err -> pure (Left err)
1166 Right attrs -> do
1167 let attrOrders = zipWith
1168 (\attr' order'
1169 -> DF.AttributeOrder (attributeName attr') order')
1170 attrs
1171 orders
1172 case DF.sortDataFrameBy attrOrders . DF.toDataFrame $ rel of
1173 Left err -> pure (Left err)
1174 Right dFrame -> do
1175 let dFrame' = maybe dFrame (`DF.drop'` dFrame) (DF.offset dfExpr)
1176 dFrame'' = maybe dFrame' (`DF.take'` dFrame') (DF.limit dfExpr)
1177 pure (Right dFrame'')
1178executeDataFrameExpr sessionId conn@(RemoteProcessConnection _) dfExpr = remoteCall conn (ExecuteDataFrameExpr sessionId dfExpr)