f1e4777 · project-m36 · git.iko.soy

git.iko.soy/project-m36

Merge branch 'notifications'

commitf1e47770f37a78fe6a9329ee8af671f2c0043469 authorAgentM <agentm@themactionfaction.com> dateMon 1 Feb 2016, 16:18 (11 years ago) parentaf513d3 c08df81

21 files changed+455−107

compare with master

unifiedsplit

ProjectM36/Base.hs+17−1

@@ -246,11 +246,22 @@ data RelationalExpr where

246246

247247instance Binary RelationalExpr

248248

249+type NotificationName = StringType

250+type Notifications = M.Map NotificationName Notification

251+

252+-- | When the changeExpr returns a different result in the database context, then the reportExpr is triggered and sent asynchronously to all clients.

253+data Notification = Notification {

254+ changeExpr :: RelationalExpr,

255+ reportExpr :: RelationalExpr

256+ }

257+ deriving (Show, Eq, Binary, Generic)

258+

249259-- | The DatabaseContext is a snapshot of a database's evolving state and contains everything a database client can change over time.

250260data DatabaseContext = DatabaseContext {

251261 inclusionDependencies :: M.Map IncDepName InclusionDependency,

252262 relationVariables :: M.Map RelVarName Relation,

253− atomFunctions :: AtomFunctions

263+ atomFunctions :: AtomFunctions,

264+ notifications :: Notifications

254265 } deriving (Show, Generic)

255266

256267type IncDepName = StringType

@@ -268,8 +279,13 @@ data DatabaseExpr where

268279 Insert :: RelVarName -> RelationalExpr -> DatabaseExpr

269280 Delete :: RelVarName -> RestrictionPredicateExpr -> DatabaseExpr

270281 Update :: RelVarName -> M.Map AttributeName Atom -> RestrictionPredicateExpr -> DatabaseExpr -- needs restriction support

282+

271283 AddInclusionDependency :: IncDepName -> InclusionDependency -> DatabaseExpr

272284 RemoveInclusionDependency :: IncDepName -> DatabaseExpr

285+

286+ AddNotification :: NotificationName -> RelationalExpr -> RelationalExpr -> DatabaseExpr

287+ RemoveNotification :: NotificationName -> DatabaseExpr

288+

273289 MultipleExpr :: [DatabaseExpr] -> DatabaseExpr

274290 deriving (Show, Eq, Binary, Generic)

275291

ProjectM36/Client.hs+173−52

@@ -1,4 +1,4 @@

1−{-# LANGUAGE OverloadedStrings #-}

1+{-# LANGUAGE OverloadedStrings, DeriveAnyClass, DeriveGeneric #-}

22module ProjectM36.Client

33 (ConnectionInfo(..),

44 Connection(..),

@@ -34,12 +34,16 @@ module ProjectM36.Client

3434 createSessionAtCommit,

3535 createSessionAtHead,

3636 closeSession,

37+ addClientNode,

3738 TransactionGraphOperator(..),

3839 Atomable,

3940 NodeId(..),

4041 Atom(..),

4142 Session,

4243 SessionId,

44+ NotificationCallback,

45+ emptyNotificationCallback,

46+ EvaluatedNotification(..),

4347 AtomType(..)) where

4448import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well

4549import qualified ProjectM36.Base as B

@@ -51,9 +55,10 @@ import ProjectM36.TransactionGraph

5155import ProjectM36.TransactionGraph.Persist

5256import ProjectM36.Attribute

5357import ProjectM36.Persist (DiskSync(..))

58+import ProjectM36.Notifications

5459import ProjectM36.Server.RemoteCallTypes

5560import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)

56−import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)

61+import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)

5762import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

5863import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

5964import Control.Distributed.Process (NodeId(..))

@@ -62,16 +67,20 @@ import Data.UUID (UUID)

6267import Data.UUID.V4 (nextRandom)

6368import Control.Concurrent.STM

6469import Data.Word

65−import Control.Distributed.Process (ProcessId, Process)

70+import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)

6671import Control.Exception (IOException)

6772import Control.Concurrent.MVar

6873import qualified Data.Map as M

6974import Control.Distributed.Process.Serializable (Serializable)

7075--import Control.Distributed.Process.Debug

7176import qualified STMContainers.Map as STMMap

77+import qualified STMContainers.Set as STMSet

7278import ProjectM36.Session

7379import ProjectM36.Sessions

7480import ListT

81+import Data.Binary (Binary)

82+import GHC.Generics (Generic)

83+--import Debug.Trace

7584

7685type Hostname = String

7786

@@ -79,8 +88,29 @@ type Port = Word16

7988

8089type DatabaseName = String

8190

82−data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy |

83− RemoteProcessConnectionInfo DatabaseName NodeId

91+-- | 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.

92+type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()

93+

94+-- | The empty notification callback ignores all callbacks.

95+emptyNotificationCallback :: NotificationCallback

96+emptyNotificationCallback _ _ = pure ()

97+

98+-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'.

99+data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback |

100+ RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback

101+

102+type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification

103+

104+-- | Used for callbacks from the server when monitored changes have been made.

105+data NotificationMessage = NotificationMessage EvaluatedNotifications

106+ deriving (Binary, Eq, Show, Generic)

107+

108+-- | When a notification is fired, the 'reportExpr' is evaluated in the commit's context, so that is returned along with the original notification.

109+data EvaluatedNotification = EvaluatedNotification {

110+ notification :: Notification,

111+ reportRelation :: Either RelationalError Relation

112+ }

113+ deriving(Binary, Eq, Show, Generic)

84114

85115createNodeId :: Hostname -> Port -> NodeId

86116createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0

@@ -95,10 +125,12 @@ defaultHeadName :: HeadName

95125defaultHeadName = "master"

96126

97127defaultRemoteConnectionInfo :: ConnectionInfo

98−defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort)

128+defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback

99129

100130-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.

101−data Connection = InProcessConnection PersistenceStrategy Sessions (TVar TransactionGraph) |

131+type ClientNodes = STMSet.Set ProcessId

132+

133+data Connection = InProcessConnection PersistenceStrategy ClientNodes Sessions (TVar TransactionGraph) |

102134 RemoteProcessConnection LocalNode ProcessId

103135

104136-- | There are several reasons why a connection can fail.

@@ -118,11 +150,32 @@ commonLocalNode = do

118150 case eLocalTransport of

119151 Left err -> pure (Left $ IOExceptionError err)

120152 Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right

153+

154+notificationListener :: NotificationCallback -> Process ()

155+notificationListener callback = do

156+ --pid <- getSelfPid

157+ --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid

158+ _ <- forever $ do

159+ receiveWait [

160+ match (\(NotificationMessage eNots) -> do

161+ --say $ "NOTIFICATION: " ++ show eNots

162+ liftIO $ mapM_ (uncurry callback) (M.toList eNots)

163+ )

164+ ]

165+ --say "LISTENER THREAD EXIT"

166+ pure ()

167+

168+startNotificationListener :: NotificationCallback -> IO (ProcessId)

169+startNotificationListener callback = do

170+ eLocalNode <- commonLocalNode

171+ case eLocalNode of

172+ Left err -> error ("Failed to start local notification listener: " ++ show err)

173+ Right localNode -> forkProcess localNode (notificationListener callback)

121174

122175-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.

123176connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)

124177--create a new in-memory database/transaction graph

125−connectProjectM36 (InProcessConnectionInfo strat) = do

178+connectProjectM36 (InProcessConnectionInfo strat notificationCallback) = do

126179 freshUUID <- nextRandom

127180 let bootstrapContext = RE.basicDatabaseContext

128181 freshGraph = bootstrapTransactionGraph freshUUID bootstrapContext

@@ -130,14 +183,19 @@ connectProjectM36 (InProcessConnectionInfo strat) = do

130183 --create date examples graph for now- probably should be empty context in the future

131184 NoPersistence -> do

132185 graphTvar <- newTVarIO freshGraph

186+ clientNodes <- STMSet.newIO

133187 sessions <- STMMap.newIO

134− return $ Right $ InProcessConnection strat sessions graphTvar

135− MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

136− CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

188+ notificationPid <- startNotificationListener notificationCallback

189+ let conn = InProcessConnection strat clientNodes sessions graphTvar

190+ addClientNode conn notificationPid

191+ pure (Right conn)

192+ MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback

193+ CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback

137194

138−connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

195+connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do

139196 connStatus <- newEmptyMVar

140197 eLocalNode <- commonLocalNode

198+ notificationListenerPid <- startNotificationListener notificationCallback

141199 let dbName = remoteDBLookupName databaseName

142200 putStrLn $ show serverNodeId ++ " " ++ dbName

143201 case eLocalNode of

@@ -148,7 +206,7 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

148206 case mServerProcessId of

149207 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)

150208 Just serverProcessId -> do

151− loginConfirmation <- call serverProcessId Login

209+ loginConfirmation <- call serverProcessId (Login notificationListenerPid)

152210 if not loginConfirmation then

153211 liftIO $ putMVar connStatus (Left LoginError)

154212 else do

@@ -159,9 +217,10 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

159217connectPersistentProjectM36 :: PersistenceStrategy ->

160218 DiskSync ->

161219 FilePath ->

162− TransactionGraph ->

220+ TransactionGraph ->

221+ NotificationCallback ->

163222 IO (Either ConnectionError Connection)

164−connectPersistentProjectM36 strat sync dbdir freshGraph = do

223+connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback = do

165224 err <- setupDatabaseDir sync dbdir freshGraph

166225 case err of

167226 Just err' -> return $ Left (SetupDatabaseDirectoryError err')

@@ -172,18 +231,22 @@ connectPersistentProjectM36 strat sync dbdir freshGraph = do

172231 Right graph' -> do

173232 tvarGraph <- newTVarIO graph'

174233 sessions <- STMMap.newIO

175− return $ Right $ InProcessConnection strat sessions tvarGraph

176−

234+ clientNodes <- STMSet.newIO

235+ let conn = InProcessConnection strat clientNodes sessions tvarGraph

236+ notificationPid <- startNotificationListener notificationCallback

237+ addClientNode conn notificationPid

238+ pure (Right conn)

239+

177240-- | Create a new session at the transaction UUID and return the session's Id.

178241createSessionAtCommit :: UUID -> Connection -> IO (Either RelationalError SessionId)

179−createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _) = do

242+createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _ _) = do

180243 newSessionId <- nextRandom

181244 atomically $ do

182245 createSessionAtCommit_ commitUUID newSessionId conn

183246createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)

184247

185248createSessionAtCommit_ :: UUID -> SessionId -> Connection -> STM (Either RelationalError SessionId)

186−createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ sessions graphTvar) = do

249+createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ _ sessions graphTvar) = do

187250 graph <- readTVar graphTvar

188251 case transactionForUUID commitUUID graph of

189252 Left err -> pure (Left err)

@@ -199,26 +262,31 @@ createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionA

199262

200263-- | 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.

201264createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)

202−createSessionAtHead headn conn@(InProcessConnection _ _ tvar) = do

265+createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar) = do

203266 newSessionId <- nextRandom

204267 atomically $ do

205− graph <- readTVar tvar

268+ graph <- readTVar graphTvar

206269 case transactionForHead headn graph of

207270 Nothing -> pure $ Left (NoSuchHeadNameError headn)

208271 Just trans -> createSessionAtCommit_ (transactionUUID trans) newSessionId conn

209−createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)

272+createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)

273+

274+-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.

275+addClientNode :: Connection -> ProcessId -> IO ()

276+addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"

277+addClientNode (InProcessConnection _ clientNodes _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)

210278

211279-- | Discards a session, eliminating any uncommitted changes present in the session.

212280closeSession :: SessionId -> Connection -> IO ()

213−closeSession sessionId (InProcessConnection _ sessions _) = do

214− atomically $ do

215− STMMap.delete sessionId sessions

281+closeSession sessionId (InProcessConnection _ _ sessions _) = do

282+ atomically $ STMMap.delete sessionId sessions

216283closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)

217284-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.

218285close :: Connection -> IO ()

219−close (InProcessConnection _ sessions _) = atomically $ do

286+close (InProcessConnection _ _ sessions _) = atomically $ do

220287 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)

221288 pure ()

289+

222290close (RemoteProcessConnection localNode serverProcessId) = do

223291 runProcessResult localNode $ do

224292 call serverProcessId Logout

@@ -232,7 +300,7 @@ runProcessResult localNode proc = do

232300 takeMVar ret

233301

234302remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b

235−remoteCall (InProcessConnection _ _ _) _ = error "remoteCall called on local connection"

303+remoteCall (InProcessConnection _ _ _ _) _ = error "remoteCall called on local connection"

236304remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do

237305 ret <- safeCall serverProcessId arg

238306 case ret of

@@ -246,7 +314,7 @@ sessionForSessionId sessionId sessions = do

246314

247315-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.

248316executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

249−executeRelationalExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do

317+executeRelationalExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do

250318 eSession <- sessionForSessionId sessionId sessions

251319 case eSession of

252320 Left err -> pure $ Left err

@@ -255,7 +323,7 @@ executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = rem

255323

256324-- | 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.

257325executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Maybe RelationalError)

258−executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do

326+executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do

259327 eSession <- sessionForSessionId sessionId sessions

260328 case eSession of

261329 Left err -> pure $ Just err

@@ -268,14 +336,11 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = a

268336 return Nothing

269337executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)

270338

271−-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

272−executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

273−executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphExpr = do

274− freshUUID <- nextRandom

275− manip <- atomically $ do

276− graph <- readTVar graphTvar

277− eSession <- sessionForSessionId sessionId sessions

278− case eSession of

339+executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)

340+executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do

341+ graph <- readTVar graphTvar

342+ eSession <- sessionForSessionId sessionId sessions

343+ case eSession of

279344 Left err -> pure $ Left err

280345 Right (Session discon) -> case evalGraphOp freshUUID discon graph graphExpr of

281346 Left err -> pure $ Left err

@@ -284,23 +349,76 @@ executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphE

284349 let newSession = Session discon'

285350 STMMap.insert newSession sessionId sessions

286351 pure $ Right graph'

352+

353+-- process notifications for commits

354+executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)

355+executeCommitExprSTM_ oldContext newContext nodes = do

356+ let nots = notifications oldContext

357+ fireNots = notificationChanges nots oldContext newContext

358+ evaldNots = M.map mkEvaldNot fireNots

359+ mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) oldContext }

360+ pure (evaldNots, nodes)

361+

362+

363+-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

364+executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

365+executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar) graphExpr = do

366+ freshUUID <- nextRandom

367+ manip <- atomically $ do

368+ eSession <- sessionForSessionId sessionId sessions

369+ oldGraph <- readTVar graphTvar

370+ case eSession of

371+ Left err -> pure (Left err)

372+ Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do

373+ eGraph <- executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar

374+ case eGraph of

375+ Left err -> pure (Left err)

376+ Right newGraph -> do

377+ if graphExpr == Commit then

378+ case transactionForUUID parentUUID oldGraph of

379+ Left err -> pure $ Left err

380+ Right (Transaction _ _ previousContext) -> do

381+ (evaldNots, nodes) <- executeCommitExprSTM_ previousContext currentContext clientNodes

382+ nodesToNotify <- toList (STMSet.stream nodes)

383+ pure $ Right (evaldNots, nodesToNotify, newGraph)

384+ else

385+ pure $ Right (M.empty, [], newGraph)

287386 case manip of

288387 Left err -> return $ Just err

289− Right newGraph -> do

388+ Right (notsToFire, nodesToNotify, newGraph) -> do

290389 --update filesystem database, if necessary

291390 --this should really grab a lock at the beginning of the method to be threadsafe

292391 processPersistence strat newGraph

392+ sendNotifications nodesToNotify notsToFire

293393 return Nothing

294394executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)

295−

296−-- | After modifying a session, 'commit' the transaction to the transaction graph at the head which the session is referencing.

395+

396+{-

397+commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)

398+commitSTM_ freshUUID sessionId sessions graph = do

399+ mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph

400+ pure $ case mErr of

401+ Left err -> Just err

402+ Right _ -> Nothing

403+-}

404+

405+-- | 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.

297406commit :: SessionId -> Connection -> IO (Maybe RelationalError)

298−commit sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Commit

407+commit sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Commit

299408commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)

409+

410+sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()

411+sendNotifications pids nots = mapM_ sendNots pids

412+ where

413+ sendNots remoteClientPid = do

414+ eLocalNode <- commonLocalNode

415+ case eLocalNode of

416+ Left err -> error ("Failed to get local node: " ++ show err)

417+ Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)

300418

301419-- | 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.

302420rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

303−rollback sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Rollback

421+rollback sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Rollback

304422rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

305423

306424processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

@@ -310,11 +428,11 @@ processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist

310428

311429-- | 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.

312430typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

313−typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr

431+typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr

314432typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)

315433

316434typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)

317−typeForRelationalExprSTM sessionId (InProcessConnection _ sessions _) relExpr = do

435+typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _) relExpr = do

318436 eSession <- sessionForSessionId sessionId sessions

319437 case eSession of

320438 Left err -> pure $ Left err

@@ -324,7 +442,7 @@ typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-l

324442

325443-- | Return a 'Map' of the database's constraints at the context of the session and connection.

326444inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))

327−inclusionDependencies sessionId (InProcessConnection _ sessions _) = do

445+inclusionDependencies sessionId (InProcessConnection _ _ sessions _) = do

328446 atomically $ do

329447 eSession <- sessionForSessionId sessionId sessions

330448 case eSession of

@@ -335,7 +453,7 @@ inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall

335453

336454-- | 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.

337455planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)

338−planForDatabaseContextExpr sessionId (InProcessConnection _ sessions _) dbExpr = do

456+planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) dbExpr = do

339457 atomically $ do

340458 eSession <- sessionForSessionId sessionId sessions

341459 case eSession of

@@ -349,7 +467,7 @@ planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =

349467-- * id- UUID attribute of the transaction

350468-- * parents- a relation-valued attribute which contains a relation of UUIDs which are parent transaction to the transaction

351469transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)

352−transactionGraphAsRelation sessionId (InProcessConnection _ sessions tvar) = do

470+transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar) = do

353471 atomically $ do

354472 eSession <- sessionForSessionId sessionId sessions

355473 case eSession of

@@ -362,7 +480,7 @@ transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remote

362480

363481-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.

364482headTransactionUUID :: SessionId -> Connection -> IO (Maybe UUID)

365−headTransactionUUID sessionId (InProcessConnection _ sessions _) = do

483+headTransactionUUID sessionId (InProcessConnection _ _ sessions _) = do

366484 atomically $ do

367485 eSession <- sessionForSessionId sessionId sessions

368486 case eSession of

@@ -370,10 +488,8 @@ headTransactionUUID sessionId (InProcessConnection _ sessions _) = do

370488 Right session -> pure $ Just (sessionParentUUID session)

371489headTransactionUUID sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionUUID sessionId)

372490

373−-- | Returns Just the name of the head of the current disconnected transaction or Nothing.

374−headName :: SessionId -> Connection -> IO (Maybe HeadName)

375−headName sessionId (InProcessConnection _ sessions graphTvar) = do

376− atomically $ do

491+headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)

492+headNameSTM_ sessionId sessions graphTvar = do

377493 graph <- readTVar graphTvar

378494 eSession <- sessionForSessionId sessionId sessions

379495 case eSession of

@@ -381,6 +497,11 @@ headName sessionId (InProcessConnection _ sessions graphTvar) = do

381497 Right session -> pure $ case transactionForUUID (sessionParentUUID session) graph of

382498 Left _ -> Nothing

383499 Right parentTrans -> headNameForTransaction parentTrans graph

500+

501+-- | Returns Just the name of the head of the current disconnected transaction or Nothing.

502+headName :: SessionId -> Connection -> IO (Maybe HeadName)

503+headName sessionId (InProcessConnection _ _ sessions graphTvar) = do

504+ atomically (headNameSTM_ sessionId sessions graphTvar)

384505headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)

385506

386507

ProjectM36/DatabaseContext.hs+6−4

@@ -6,13 +6,15 @@ import qualified Data.HashSet as HS

66empty :: DatabaseContext

77empty = DatabaseContext { inclusionDependencies = M.empty,

88 relationVariables = M.empty,

9+ notifications = M.empty,

910 atomFunctions = HS.empty }

1011

1112-- | convert an existing database context into its constituent expression.

1213databaseContextAsDatabaseContextExpr :: DatabaseContext -> DatabaseExpr

13−databaseContextAsDatabaseContextExpr (DatabaseContext incDeps relVars funcs) = MultipleExpr $ incDepsExprs ++ relVarsExprs ++ funcsExprs

14+databaseContextAsDatabaseContextExpr context = MultipleExpr $ incDepsExprs ++ relVarsExprs ++ funcsExprs

1415 where

15− relVarsExprs = map (\(name, rel) -> Assign name (ExistingRelation rel)) (M.toList relVars)

16+ relVarsExprs = map (\(name, rel) -> Assign name (ExistingRelation rel)) (M.toList (relationVariables context))

1617 incDepsExprs :: [DatabaseExpr]

17− incDepsExprs = map (\(name, dep) -> AddInclusionDependency name dep) (M.toList incDeps)

18− funcsExprs = [] -- map (\func -> ) (HS.toList funcs) -- there are no databaseExprs to add atom functions yet

18+ incDepsExprs = map (\(name, dep) -> AddInclusionDependency name dep) (M.toList (inclusionDependencies context))

19+ funcsExprs = [] -- map (\func -> ) (HS.toList funcs) -- there are no databaseExprs to add atom functions yet

20+

ProjectM36/Error.hs+2

@@ -39,6 +39,8 @@ data RelationalError = NoSuchAttributeNamesError (S.Set AttributeName)

3939 | AtomTypeCountError [AtomType] [AtomType]

4040 | AtomFunctionTypeError AtomFunctionName Int AtomType AtomType --arg number

4141 | RelationValuedAttributesNotSupportedError [AttributeName]

42+ | NotificationNameInUseError NotificationName

43+ | NotificationNameNotInUseError NotificationName

4244 | ImportError T.Text -- really? This should be broken out into some other error type- this has nothing to do with relational algebra

4345 | ExportError T.Text

4446 deriving (Show,Eq,Generic,Binary,Typeable)

ProjectM36/Notifications.hsnew file+17

@@ -0,0 +1,17 @@

1+module ProjectM36.Notifications where

2+import ProjectM36.Base

3+import ProjectM36.RelationalExpression

4+import Control.Monad.State

5+import qualified Data.Map as M

6+import Data.Either (isRight)

8+-- | Returns the notifications which should be triggered based on the transition from the first 'DatabaseContext' to the second 'DatabaseContext'.

9+notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

10+notificationChanges nots context1 context2 = M.filter notificationFilter nots

11+ where

12+ notificationFilter (Notification chExpr _) = let oldChangeEval = evalChangeExpr chExpr context1

13+ newChangeEval = evalChangeExpr chExpr context2 in

14+ oldChangeEval /= newChangeEval && isRight oldChangeEval

15+ evalChangeExpr chExpr = evalState (evalRelationalExpr chExpr)

16+

17+

ProjectM36/RelationalExpression.hs+35−14

@@ -116,19 +116,22 @@ evalRelationalExpr (Extend tupleExpression relExpr) = do

116116emptyDatabaseContext :: DatabaseContext

117117emptyDatabaseContext = DatabaseContext { inclusionDependencies = M.empty,

118118 relationVariables = M.empty,

119− atomFunctions = HS.empty

119+ atomFunctions = HS.empty,

120+ notifications = M.empty

120121 }

121122

122123basicDatabaseContext :: DatabaseContext

123124basicDatabaseContext = DatabaseContext { inclusionDependencies = M.empty,

124125 relationVariables = M.fromList [("true", relationTrue),

125126 ("false", relationFalse)],

126− atomFunctions = basicAtomFunctions

127+ atomFunctions = basicAtomFunctions,

128+ notifications = M.empty

127129 }

128130

129131dateExamples :: DatabaseContext

130132dateExamples = DatabaseContext { inclusionDependencies = dateIncDeps,

131133 relationVariables = M.union (relationVariables basicDatabaseContext) dateRelVars,

134+ notifications = M.empty,

132135 atomFunctions = basicAtomFunctions}

133136 where

134137 dateRelVars = M.fromList [("S", suppliers),

@@ -206,7 +209,7 @@ setRelVar :: RelVarName -> Relation -> DatabaseState (Maybe RelationalError)

206209setRelVar relVarName rel = do

207210 currentContext <- get

208211 let newRelVars = M.insert relVarName rel $ relationVariables currentContext

209− let potentialContext = DatabaseContext (inclusionDependencies currentContext) newRelVars (atomFunctions currentContext)

212+ potentialContext = currentContext { relationVariables = newRelVars }

210213 case checkConstraints potentialContext of

211214 Just err -> return $ Just err

212215 Nothing -> do

@@ -216,9 +219,9 @@ setRelVar relVarName rel = do

216219-- it is not an error to delete a relvar which does not exist, just like it is not an error to insert a pre-existing tuple into a relation

217220deleteRelVar :: RelVarName -> DatabaseState (Maybe RelationalError)

218221deleteRelVar relVarName = do

219− currstate <- get

220− let newRelVars = M.delete relVarName (relationVariables currstate)

221− put $ DatabaseContext (inclusionDependencies currstate) newRelVars (atomFunctions currstate)

222+ currcontext <- get

223+ let newRelVars = M.delete relVarName (relationVariables currcontext)

224+ put $ currcontext { relationVariables = newRelVars }

222225 return Nothing

223226

224227evalContextExpr :: DatabaseExpr -> DatabaseState (Maybe RelationalError)

@@ -284,7 +287,7 @@ evalContextExpr (AddInclusionDependency newDepName newDep) = do

284287 if M.member newDepName currDeps then

285288 return $ Just (InclusionDependencyNameInUseError newDepName)

286289 else do

287− let potentialContext = DatabaseContext newDeps (relationVariables currContext) (atomFunctions currContext)

290+ let potentialContext = currContext { inclusionDependencies = newDeps }

288291 case checkConstraints potentialContext of

289292 Just err -> return $ Just err

290293 Nothing -> do

@@ -298,7 +301,30 @@ evalContextExpr (RemoveInclusionDependency depName) = do

298301 if M.notMember depName currDeps then

299302 return $ Just (InclusionDependencyNameNotInUseError depName)

300303 else do

301− put $ DatabaseContext newDeps (relationVariables currContext) (atomFunctions currContext)

304+ put $ currContext {inclusionDependencies = newDeps }

305+ return Nothing

306+

307+-- | Add a notification which will send the resultExpr when triggerExpr changes between commits.

308+evalContextExpr (AddNotification notName triggerExpr resultExpr) = do

309+ currentContext <- get

310+ let nots = notifications currentContext

311+ if M.member notName nots then

312+ return $ Just (NotificationNameInUseError notName)

313+ else do

314+ let newNotifications = M.insert notName newNotification nots

315+ newNotification = Notification { changeExpr = triggerExpr,

316+ reportExpr = resultExpr }

317+ put $ currentContext { notifications = newNotifications }

318+ return Nothing

319+

320+evalContextExpr (RemoveNotification notName) = do

321+ currentContext <- get

322+ let nots = notifications currentContext

323+ if M.notMember notName nots then

324+ return $ Just (NotificationNameNotInUseError notName)

325+ else do

326+ let newNotifications = M.delete notName nots

327+ put $ currentContext { notifications = newNotifications }

302328 return Nothing

303329

304330evalContextExpr (MultipleExpr exprs) = do

@@ -309,9 +335,6 @@ evalContextExpr (MultipleExpr exprs) = do

309335 [] -> return $ Nothing

310336 err:_ -> return $ Just err

311337

312−-- restrict relvar to get affected tuples, update tuples, delete restriction from relvar, relvar = relvar union updated tuples

313−--evalRelVarExpr (Update relVarName updateMap) = do

314−

315338--run verification on all constraints

316339checkConstraints :: DatabaseContext -> Maybe RelationalError

317340checkConstraints context = case failures of

@@ -360,11 +383,9 @@ typeForRelationalExpr expr = do

360383--returns a database context with all tuples removed

361384--this is useful for type checking and optimization

362385contextWithEmptyTupleSets :: DatabaseContext -> DatabaseContext

363−contextWithEmptyTupleSets contextIn = DatabaseContext incDeps relVars funcs

386+contextWithEmptyTupleSets contextIn = contextIn { relationVariables = relVars }

364387 where

365− incDeps = inclusionDependencies contextIn

366388 relVars = M.map (\rel -> Relation (attributes rel) emptyTupleSet) (relationVariables contextIn)

367− funcs = atomFunctions contextIn

368389

369390{- used for restrictions- take the restrictionpredicate and return the corresponding filter function -}

370391predicateRestrictionFilter :: DatabaseContext -> Attributes -> RestrictionPredicateExpr -> Either RelationalError (RelationTuple -> Bool)

ProjectM36/Server.hs+7−3

@@ -29,7 +29,7 @@ serverDefinition = defaultProcess {

2929 handleCall (\conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr) -> handleRetrievePlanForDatabaseContextExpr sessionId conn dbExpr),

3030 handleCall (\conn (RetrieveHeadTransactionUUID sessionId) -> handleRetrieveHeadTransactionUUID sessionId conn),

3131 handleCall (\conn (RetrieveTransactionGraph sessionId) -> handleRetrieveTransactionGraph sessionId conn),

32− handleCall (\conn Login -> handleLogin conn),

32+ handleCall (\conn (Login procId) -> handleLogin conn procId),

3333 handleCall (\conn (CreateSessionAtHead headn) -> handleCreateSessionAtHead headn conn),

3434 handleCall (\conn (CreateSessionAtCommit commitId) -> handleCreateSessionAtCommit commitId conn),

3535 handleCall (\conn (CloseSession sessionId) -> handleCloseSession sessionId conn)

@@ -54,11 +54,15 @@ registerDB dbname = do

5454 let dbname' = remoteDBLookupName dbname

5555 register dbname' self

5656 liftIO $ putStrLn $ "registered " ++ (show self) ++ " " ++ dbname'

57−

57+

58+-- | A notification callback which logs the notification to stderr and does nothing else.

59+loggingNotificationCallback :: NotificationCallback

60+loggingNotificationCallback notName evaldNot = hPutStrLn stderr $ "Notification received \"" ++ show notName ++ "\": " ++ show evaldNot

61+

5862-- | A synchronous function to start the project-m36 daemon given an appropriate 'ServerConfig'. Note that this function only returns if the server exits. Returns False if the daemon exited due to an error. If the second argument is not Nothing, the port is put after the server is ready to service the port.

5963launchServer :: ServerConfig -> Maybe (MVar Port) -> IO (Bool)

6064launchServer daemonConfig mPortMVar = do

61− econn <- connectProjectM36 (InProcessConnectionInfo (persistenceStrategy daemonConfig))

65+ econn <- connectProjectM36 (InProcessConnectionInfo (persistenceStrategy daemonConfig) loggingNotificationCallback)

6266 case econn of

6367 Left err -> do

6468 hPutStrLn stderr ("Failed to create database connection: " ++ show err)

ProjectM36/Server/EntryPoints.hs+5−3

@@ -2,7 +2,7 @@ module ProjectM36.Server.EntryPoints where

22import ProjectM36.Base hiding (inclusionDependencies)

33import ProjectM36.Client

44import ProjectM36.Error

5−import Control.Distributed.Process (Process)

5+import Control.Distributed.Process (Process, ProcessId)

66import Control.Distributed.Process.ManagedProcess (ProcessReply)

77import Control.Distributed.Process.ManagedProcess.Server (reply)

88import Control.Monad.IO.Class (liftIO)

@@ -24,8 +24,10 @@ handleExecuteHeadName sessionId conn = do

2424 ret <- liftIO $ headName sessionId conn

2525 reply ret conn

2626

27−handleLogin :: Connection -> Process (ProcessReply Bool Connection)

28−handleLogin conn = reply True conn

27+handleLogin :: Connection -> ProcessId -> Process (ProcessReply Bool Connection)

28+handleLogin conn newClientProcessId = do

29+ liftIO (addClientNode conn newClientProcessId)

30+ reply True conn

2931

3032handleExecuteGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> Process (ProcessReply (Maybe RelationalError) Connection)

3133handleExecuteGraphExpr sessionId conn graphExpr = do

ProjectM36/Server/RemoteCallTypes.hs+4−1

@@ -6,9 +6,12 @@ import ProjectM36.Session

66import GHC.Generics

77import Data.Binary

88import Data.UUID

9+import Control.Distributed.Process (ProcessId)

910

10−data Login = Login

11+-- | The initial login message. The argument should be the process id of the initiating client. This ProcessId will receive notification callbacks.

12+data Login = Login ProcessId

1113 deriving (Binary, Generic)

14+

1215data Logout = Logout

1316 deriving (Binary, Generic)

1417data ExecuteRelationalExpr = ExecuteRelationalExpr SessionId RelationalExpr

ProjectM36/StaticOptimizer.hs+12

@@ -121,6 +121,18 @@ applyStaticDatabaseOptimization dep@(AddInclusionDependency _ _) = return $ Righ

121121

122122applyStaticDatabaseOptimization (RemoveInclusionDependency name) = return $ Right (RemoveInclusionDependency name)

123123

124+applyStaticDatabaseOptimization (AddNotification name triggerExpr resultExpr) = do

125+ eTriggerExprOpt <- applyStaticRelationalOptimization triggerExpr

126+ case eTriggerExprOpt of

127+ Left err -> pure $ Left err

128+ Right triggerExprOpt -> do

129+ eResultExprOpt <- applyStaticRelationalOptimization resultExpr

130+ case eResultExprOpt of

131+ Left err -> pure $ Left err

132+ Right resultExprOpt -> pure (Right (AddNotification name triggerExprOpt resultExprOpt))

133+

134+applyStaticDatabaseOptimization notif@(RemoveNotification _) = pure (Right notif)

135+

124136--optimization: from pgsql lists- check for join condition referencing foreign key- if join projection project away the referenced table, then it does not need to be scanned

125137

126138--applyStaticDatabaseOptimization (MultipleExpr exprs) = return $ Right $ MultipleExpr exprs

ProjectM36/Transaction/Persist.hs+1

@@ -50,6 +50,7 @@ readTransaction dbdir transUUID = do

5050 let atomFuncs = basicAtomFunctions

5151 let newContext = DatabaseContext { inclusionDependencies = incDeps,

5252 relationVariables = relvars,

53+ notifications = M.empty,

5354 atomFunctions = atomFuncs }

5455

5556 return $ Right $ Transaction transUUID transInfo newContext

ProjectM36/TransactionGraph.hs+2−1

@@ -247,4 +247,5 @@ evalROGraphOp :: DisconnectedTransaction -> TransactionGraph -> ROTransactionGra

247247evalROGraphOp discon graph ShowGraph = do

248248 graphRel <- graphAsRelation discon graph

249249 return graphRel

250−-}

250+-}

251+

README.markdown+4−3

@@ -32,10 +32,11 @@ Project:M36 supports multiple frontends which target different audiences.

32321. [Adding New Data Types](docs/new_datatypes.markdown)

33331. [ACID Database Properties](docs/acid_assessment.markdown)

34341. [Serving Remote ProjectM36 Databases](docs/server_mode.markdown)

35−

35+1. [Using Notifications](docs/using_notifications.markdown)

36+

3637## Development

3738

38−Project:M36 is developed in Haskell and compiled with GHC.

39+Project:M36 is developed in Haskell and compiled with GHC.

3940

4041Important modules:

4142

@@ -53,4 +54,4 @@ Other modules include:

5354* relational expression cost-based optimizer

5455* relational error enumeration

5556* import of CSV as a relation and export of a relation to CSV

56−* import of TutorialD from files

57+* import of TutorialD from files

TutorialD/Interpreter.hs+9

@@ -25,6 +25,7 @@ import System.Console.Haskeline

2525import System.Directory (getHomeDirectory)

2626import qualified Data.Text as T

2727import Data.Maybe (fromMaybe)

28+import System.IO (hPutStrLn, stderr)

2829

2930{-

3031context ops are read-only operations which only operate on the database context (relvars and constraints)

@@ -117,6 +118,14 @@ evalTutorialD sessionId conn expr = case expr of

117118data InterpreterConfig = LocalInterpreterConfig PersistenceStrategy HeadName|

118119 RemoteInterpreterConfig C.NodeId C.DatabaseName HeadName

119120

121+outputNotificationCallback :: C.NotificationCallback

122+outputNotificationCallback notName evaldNot = hPutStrLn stderr $ "Notification received " ++ show notName ++ ":\n" ++ show (reportExpr (C.notification evaldNot)) ++ "\n" ++ prettyEvaluatedNotification evaldNot

123+

124+prettyEvaluatedNotification :: C.EvaluatedNotification -> String

125+prettyEvaluatedNotification eNotif = case C.reportRelation eNotif of

126+ Left err -> show err

127+ Right reportRel -> T.unpack (showRelation reportRel)

128+

120129reprLoop :: InterpreterConfig -> C.SessionId -> C.Connection -> IO ()

121130reprLoop config sessionId conn = do

122131 homeDirectory <- getHomeDirectory

TutorialD/Interpreter/DatabaseExpr.hs+17

@@ -24,6 +24,8 @@ databaseExprP = insertP

2424 <|> defineP

2525 <|> undefineP

2626 <|> assignP

27+ <|> addNotificationP

28+ <|> removeNotificationP

2729

2830assignP :: Parser DatabaseExpr

2931assignP = do

@@ -107,6 +109,20 @@ attributeAssignmentP = do

107109 reservedOp ":="

108110 atom <- stringAtomP <|> intAtomP

109111 return $ (attrName, atom)

112+

113+addNotificationP :: Parser DatabaseExpr

114+addNotificationP = do

115+ reserved "notify"

116+ notName <- identifier

117+ triggerExpr <- relExprP

118+ resultExpr <- relExprP

119+ return $ AddNotification (T.pack notName) triggerExpr resultExpr

120+

121+removeNotificationP :: Parser DatabaseExpr

122+removeNotificationP = do

123+ reserved "unnotify"

124+ notName <- identifier

125+ return $ RemoveNotification (T.pack notName)

110126

111127databaseExprOpP :: Parser DatabaseExpr

112128databaseExprOpP = multipleDatabaseExprP

@@ -131,3 +147,4 @@ interpretNO context tutdstring = case parseString tutdstring of

131147 Left err -> (Just err, context)

132148 Right parsed -> runState (evalContextExpr parsed) context

133149-}

150+

TutorialD/tutd.hs+3−3

@@ -49,8 +49,8 @@ opts :: ParserInfo InterpreterConfig

4949opts = info parseArgs idm

5050

5151connectionInfoForConfig :: InterpreterConfig -> ConnectionInfo

52−connectionInfoForConfig (LocalInterpreterConfig pStrategy _) = InProcessConnectionInfo pStrategy

53−connectionInfoForConfig (RemoteInterpreterConfig remoteNodeId remoteDBName _) = RemoteProcessConnectionInfo remoteDBName remoteNodeId

52+connectionInfoForConfig (LocalInterpreterConfig pStrategy _) = InProcessConnectionInfo pStrategy outputNotificationCallback

53+connectionInfoForConfig (RemoteInterpreterConfig remoteNodeId remoteDBName _) = RemoteProcessConnectionInfo remoteDBName remoteNodeId outputNotificationCallback

5454

5555headNameForConfig :: InterpreterConfig -> HeadName

5656headNameForConfig (LocalInterpreterConfig _ headn) = headn

@@ -58,7 +58,7 @@ headNameForConfig (RemoteInterpreterConfig _ _ headn) = headn

5858

5959errDie :: String -> IO ()

6060errDie err = hPutStrLn stderr err >> exitFailure

61−

61+

6262main :: IO ()

6363main = do

6464 interpreterConfig <- execParser opts

docs/using_notifications.markdownnew file+89

@@ -0,0 +1,89 @@

1+# Using Notifications in Project:M36

3+Interactive applications should strive to show the most up-to-date data to the user. To that end, Project:M36 supports asynchronously-fired notifications sent from the database server to the client which fire when user-specified relational expressions change between transactions.

5+As an example, consider an email application. Users of the applications will want to know right away when new mail arrives, so polling every five minutes can cause a worst-case latency of five minutes for a new email to appear. Using asynchronous notifications in the database, the application is notified immediately upon transaction commit and can update accordingly without delay.

7+## What is a Notification?

9+A Project:M36 notification is comprised of three user-selected parts:

10+

11+1. an arbitrary notification name: used to uniquely identify the notification in the server and client

12+2. a trigger relational expression: when the outcome of this relational expression changes between two transactions, the notification is fired

13+3. a report relational expression: when the notification fires, the expression is evaluated and passed along

14+

15+By separating the "trigger" and "report" expressions, database users control what kind of reporting they receive when the outcome of a relational expression changes.

16+

17+Currently, notifications are sent to all connected clients and not only to the client which created the notification. This may change in the future.

18+

19+## Creating a Notification

20+

21+Notifications can be created in the ```tutd``` TutorialD interpreter with the following syntax:

22+

23+```

24+notify <notification name> <trigger relational expression> <report relational expression>

25+```

26+

27+## Deleting a Notification

28+

29+Notifications are uniquely identified by name and are shared among all clients. To delete a notification for all clients:

30+

31+```

32+unnotify <notification name>

33+```

34+

35+## A Motivating Example

36+

37+Consider the following ```person``` relation variable where "name" is a key:

38+

39+```

40+TutorialD (master): person:=relation{tuple{name "Steve",address "Main St."},tuple{name "John", address "Elm St."}}

41+TutorialD (master): :showexpr person

42+┌────────┬─────┐

43+│address │name │

44+├────────┼─────┤

45+│Elm St. │John │

46+│Main St.│Steve│

47+└────────┴─────┘

48+```

49+

50+If a remote application is currently the address in the tuple for "Steve", then it would install a notification for that tuple's key:

51+

52+```

53+TutorialD (master): notify steve_change person where name="Steve" (person where name="Steve"){address}

54+TutorialD (master): :commit

55+```

56+

57+Then, when any client including the initiating client updates the "Steve" tuple, an asynchronous notification is fired:

58+

59+```

60+TutorialD (master): update person where name="Steve" (address:="Grove St.")

61+TutorialD (master): :commit

62+TutorialD (master): Notification received "steve_change": ...

63+```

64+

65+The asynchronously-received notification contains the result from the evaluated report relational expression. The expression, in this case, is the result of the report expression evaluated against the parent transactoin

66+

67+```

68+TutorialD (master): Notification received "steve_change":

69+Project (AttributeNames (fromList ["address"])) (Restrict (AttributeEqualityPredicate "name" (NakedAtomExpr Atom "Steve")) (RelationVariable "person"))

70+┌────────┐

71+│address │

72+├────────┤

73+│Main St.│

74+└────────┘

75+```

76+

77+## Notifications with ProjectM36.Client

78+

79+First, implement a ```NotificationCallback``` function to perform an IO action when an asynchronous notification is received from the DBMS. Note that this function will not be called on any specific thread; specifically, the callback is *not* triggered by calling into ProjectM36.Client functions.

80+

81+Each ProjectM36 Connection can only have one callback and it cannot be changed after Connection creation.

82+

83+Then, create a DatabaseExpr using the AddNotification constructor:

84+

85+```

86+AddNotification <notification name::Text> <change expression::RelationalExpr> <report expression::RelationalExpr>

87+```

88+

89+and execute it using ```executeDatabaseContextExpr```. In order for a notification to take effect, it must be committed as well.

project-m36.cabalmode 100755 → 100644

Only the mode changed.

test/Server/Main.hs+25−8

@@ -15,8 +15,8 @@ import Control.Concurrent

1515import Data.Either (isRight)

1616import Data.Maybe (isJust)

1717

18−testList :: SessionId -> Connection -> Test

19−testList sessionId conn = TestList $ map (\t -> t sessionId conn) [

18+testList :: SessionId -> Connection -> MVar () -> Test

19+testList sessionId conn notificationTestMVar = TestList $ map (\t -> t sessionId conn) [

2020 testRelationalExpr,

2121 testDatabaseContextExpr,

2222 testGraphExpr,

@@ -25,25 +25,27 @@ testList sessionId conn = TestList $ map (\t -> t sessionId conn) [

2525 testTransactionGraphAsRelation,

2626 testHeadTransactionUUID,

2727 testHeadName,

28− testSession

28+ testSession,

29+ testNotification notificationTestMVar

2930 ]

3031

3132main :: IO ()

3233main = do

3334 port <- launchTestServer

34− eTestConn <- testConnection port

35+ notificationTestMVar <- newEmptyMVar

36+ eTestConn <- testConnection port notificationTestMVar

3537 case eTestConn of

3638 Left err -> putStrLn (show err) >> exitFailure

3739 Right (session, testConn) -> do

38− tcounts <- runTestTT (testList session testConn)

40+ tcounts <- runTestTT (testList session testConn notificationTestMVar)

3941 if errors tcounts + failures tcounts > 0 then exitFailure else exitSuccess

4042

4143testDatabaseName :: DatabaseName

4244testDatabaseName = "test"

4345

44−testConnection :: Port -> IO (Either ConnectionError (SessionId, Connection))

45−testConnection port = do

46− let connInfo = RemoteProcessConnectionInfo testDatabaseName (createNodeId "127.0.0.1" port)

46+testConnection :: Port -> MVar () -> IO (Either ConnectionError (SessionId, Connection))

47+testConnection port mvar = do

48+ let connInfo = RemoteProcessConnectionInfo testDatabaseName (createNodeId "127.0.0.1" port) (testNotificationCallback mvar)

4749 eConn <- connectProjectM36 connInfo

4850 case eConn of

4951 Left err -> pure $ Left err

@@ -137,3 +139,18 @@ testSession _ conn = TestCase $ do

137139 eSessionId2 <- createSessionAtCommit headUUID conn

138140 assertBool ("invalid session: " ++ show eSessionId2) (isRight eSessionId2)

139141 closeSession sessionId1 conn

142+

143+testNotificationCallback :: MVar () -> NotificationCallback

144+testNotificationCallback mvar _ _ = putMVar mvar ()

145+

146+-- create a relvar x, add a notification on x, update x and wait for the notification

147+testNotification :: MVar () -> SessionId -> Connection -> Test

148+testNotification mvar sess conn = TestCase $ do

149+ let relvarx = RelationVariable "x"

150+ check x = x >>= maybe (pure ()) (\err -> assertFailure (show err))

151+ check $ executeDatabaseContextExpr sess conn (Assign "x" (ExistingRelation relationTrue))

152+ check $ executeDatabaseContextExpr sess conn (AddNotification "test notification" relvarx relvarx)

153+ check $ commit sess conn

154+ check $ executeDatabaseContextExpr sess conn (Assign "x" (ExistingRelation relationFalse))

155+ check $ commit sess conn

156+ takeMVar mvar

test/TutorialD/Interpreter.hs+26−13

@@ -26,6 +26,7 @@ import Data.Time.Clock (UTCTime(..))

2626import Data.Time.Calendar (fromGregorian)

2727import Data.Interval (interval, Interval, Extended(Finite))

2828import Control.Concurrent.STM

29+import Control.Concurrent

2930import qualified STMContainers.Map as STMMap

3031

3132main :: IO ()

@@ -33,7 +34,7 @@ main = do

3334 tcounts <- runTestTT (TestList tests)

3435 if errors tcounts + failures tcounts > 0 then exitFailure else exitSuccess

3536 where

36− tests = map (\(tutd, expected) -> TestCase $ assertTutdEqual basicDatabaseContext expected tutd) simpleRelTests ++ map (\(tutd, expected) -> TestCase $ assertTutdEqual dateExamples expected tutd) dateExampleRelTests ++ [transactionGraphBasicTest, transactionGraphAddCommitTest, transactionRollbackTest, transactionJumpTest, transactionBranchTest, simpleJoinTest]

37+ tests = map (\(tutd, expected) -> TestCase $ assertTutdEqual basicDatabaseContext expected tutd) simpleRelTests ++ map (\(tutd, expected) -> TestCase $ assertTutdEqual dateExamples expected tutd) dateExampleRelTests ++ [transactionGraphBasicTest, transactionGraphAddCommitTest, transactionRollbackTest, transactionJumpTest, transactionBranchTest, simpleJoinTest, testNotification]

3738 simpleRelTests = [("x:=true", Right relationTrue),

3839 ("x:=false", Right relationFalse),

3940 ("x:=true union false", Right relationTrue),

@@ -128,14 +129,14 @@ assertTutdEqual databaseContext expected tutd = assertEqual tutd expected interp

128129

129130transactionGraphBasicTest :: Test

130131transactionGraphBasicTest = TestCase $ do

131− (_, dbconn) <- dateExamplesConnection

132+ (_, dbconn) <- dateExamplesConnection emptyNotificationCallback

132133 graph <- transactionGraph dbconn

133134 assertEqual "validate bootstrapped graph" (validateGraph graph) Nothing

134135

135136--add a new transaction to the graph, validate it is in the graph

136137transactionGraphAddCommitTest :: Test

137138transactionGraphAddCommitTest = TestCase $ do

138− (sessionId, dbconn) <- dateExamplesConnection

139+ (sessionId, dbconn) <- dateExamplesConnection emptyNotificationCallback

139140 case parseTutorialD "x:=S" of

140141 Left err -> assertFailure (show err)

141142 Right parsed -> do

@@ -152,7 +153,7 @@ transactionGraphAddCommitTest = TestCase $ do

152153

153154transactionRollbackTest :: Test

154155transactionRollbackTest = TestCase $ do

155− (sessionId, dbconn) <- dateExamplesConnection

156+ (sessionId, dbconn) <- dateExamplesConnection emptyNotificationCallback

156157 graph <- transactionGraph dbconn

157158 maybeErr <- executeDatabaseContextExpr sessionId dbconn (Assign "x" (RelationVariable "S"))

158159 case maybeErr of

@@ -167,7 +168,7 @@ transactionRollbackTest = TestCase $ do

167168--commit a new transaction with "x" relation, jump to first transaction, verify that "x" is not present

168169transactionJumpTest :: Test

169170transactionJumpTest = TestCase $ do

170− (sessionId, dbconn) <- dateExamplesConnection

171+ (sessionId, dbconn) <- dateExamplesConnection emptyNotificationCallback

171172 (DisconnectedTransaction firstUUID _) <- disconnectedTransaction sessionId dbconn

172173 maybeErr <- executeDatabaseContextExpr sessionId dbconn (Assign "x" (RelationVariable "S"))

173174 case maybeErr of

@@ -190,7 +191,7 @@ maybeFail Nothing = return ()

190191--branch from the first transaction and verify that there are two heads

191192transactionBranchTest :: Test

192193transactionBranchTest = TestCase $ do

193− (sessionId, dbconn) <- dateExamplesConnection

194+ (sessionId, dbconn) <- dateExamplesConnection emptyNotificationCallback

194195 mapM_ (\x -> x >>= maybeFail) [executeGraphExpr sessionId dbconn (Branch "test"),

195196 executeDatabaseContextExpr sessionId dbconn (Assign "x" (RelationVariable "S")),

196197 commit sessionId dbconn,

@@ -224,12 +225,11 @@ simpleJoinTest = TestCase $ assertTutdEqual dateExamples joinedRel "x:=S join SP

224225 [stringAtom "London", intAtom 200, stringAtom "P2", stringAtom "S4", stringAtom "Clark", intAtom 20]

225226 ]

226227transactionGraph :: Connection -> IO TransactionGraph

227−transactionGraph (InProcessConnection _ _ tvar) = atomically $ readTVar tvar

228−

228+transactionGraph (InProcessConnection _ _ _ tvar) = atomically $ readTVar tvar

229229transactionGraph _ = error "remote connection used"

230230

231231disconnectedTransaction :: SessionId -> Connection -> IO DisconnectedTransaction

232−disconnectedTransaction sessionId (InProcessConnection _ sessions _) = do

232+disconnectedTransaction sessionId (InProcessConnection _ _ sessions _) = do

233233 mSession <- atomically $ do

234234 STMMap.lookup sessionId sessions

235235 case mSession of

@@ -243,9 +243,9 @@ inclusionDependencies (InProcessConnection (DisconnectedTransaction _ context))

243243inclusionDependencies _ = error "remote connection used"

244244-}

245245

246−dateExamplesConnection :: IO (SessionId, Connection)

247−dateExamplesConnection = do

248− dbconn <- connectProjectM36 (InProcessConnectionInfo NoPersistence)

246+dateExamplesConnection :: NotificationCallback -> IO (SessionId, Connection)

247+dateExamplesConnection callback = do

248+ dbconn <- connectProjectM36 (InProcessConnectionInfo NoPersistence callback)

249249 let incDeps = Base.inclusionDependencies dateExamples

250250 case dbconn of

251251 Left err -> error (show err)

@@ -259,5 +259,18 @@ dateExamplesConnection = do

259259 --skipping atom functions for now- there are no atom function manipulation operators yet

260260 commit sessionId conn >>= maybeFail

261261 pure (sessionId, conn)

262−

263262

263+-- test notifications over the InProcessConnection

264+testNotification :: Test

265+testNotification = TestCase $ do

266+ notifmvar <- newEmptyMVar

267+ let notifCallback mvar = \_ _ -> putMVar mvar ()

268+ relvarx = RelationVariable "x"

269+ (sess, conn) <- dateExamplesConnection (notifCallback notifmvar)

270+ let check' x = x >>= maybe (pure ()) (\err -> assertFailure (show err))

271+ check' $ executeDatabaseContextExpr sess conn (Assign "x" (ExistingRelation relationTrue))

272+ check' $ executeDatabaseContextExpr sess conn (AddNotification "test notification" relvarx relvarx)

273+ check' $ commit sess conn

274+ check' $ executeDatabaseContextExpr sess conn (Assign "x" (ExistingRelation relationFalse))

275+ check' $ commit sess conn

276+ takeMVar notifmvar

test/persistent-driver/test.hs+1−1

@@ -39,7 +39,7 @@ main = do

3939 if errors tcounts + failures tcounts > 0 then exitFailure else exitSuccess

4040

4141inProcSettings :: ConnectionInfo

42−inProcSettings = InProcessConnectionInfo NoPersistence

42+inProcSettings = InProcessConnectionInfo NoPersistence emptyNotificationCallback

4343

4444defPersonRel :: SessionId -> Connection -> IO ()

4545defPersonRel sessionId conn = do