64a9037 · project-m36 · git.iko.soy

git.iko.soy/project-m36

WIP notifications not yet propagated

commit64a903795b2782863200158180577e4b5dcb4174 authorAgentM <agentm@themactionfaction.com> dateMon 14 Dec 2015, 17:31 (11 years ago) parent781a9f8

6 files changed+89−29

compare with master

unifiedsplit

ProjectM36/Base.hs+12−1 ProjectM36/Client.hs+56−15 ProjectM36/DatabaseContext.hs+8−1 ProjectM36/RelationalExpression.hs+7−9 ProjectM36/Server/EntryPoints.hs+4−2 ProjectM36/TransactionGraph.hs+2−1

ProjectM36/Base.hs+12−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)

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

ProjectM36/Client.hs+56−15

@@ -69,6 +69,7 @@ import qualified Data.Map as M

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

7070--import Control.Distributed.Process.Debug

7171import qualified STMContainers.Map as STMMap

72+import qualified STMContainers.Set as STMSet

7273import ProjectM36.Session

7374import ProjectM36.Sessions

7475import ListT

@@ -98,7 +99,9 @@ defaultRemoteConnectionInfo :: ConnectionInfo

9899defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort)

99100

100101-- | 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) |

102+type ClientNodes = STMSet.Set ProcessId

103+

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

102105 RemoteProcessConnection LocalNode ProcessId

103106

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

@@ -130,10 +133,12 @@ connectProjectM36 (InProcessConnectionInfo strat) = do

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

131134 NoPersistence -> do

132135 graphTvar <- newTVarIO freshGraph

136+ clientNodes <- STMSet.newIO

133137 sessions <- STMMap.newIO

134− return $ Right $ InProcessConnection strat sessions graphTvar

138+ return $ Right $ InProcessConnection strat clientNodes sessions graphTvar

135139 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

136140 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

141+

137142

138143connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

139144 connStatus <- newEmptyMVar

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

148153 case mServerProcessId of

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

150155 Just serverProcessId -> do

151− loginConfirmation <- call serverProcessId Login

156+ loginConfirmation <- call serverProcessId (Login loginNode)

152157 if not loginConfirmation then

153158 liftIO $ putMVar connStatus (Left LoginError)

154159 else do

@@ -206,7 +211,13 @@ createSessionAtHead headn conn@(InProcessConnection _ _ tvar) = do

206211 case transactionForHead headn graph of

207212 Nothing -> pure $ Left (NoSuchHeadNameError headn)

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

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

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

215+

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

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

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

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

220+

210221

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

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

@@ -268,10 +279,8 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = a

268279 return Nothing

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

270281

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

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

283+executeGraphExprSTM_ sessionId sessions graphExpr graphTvar = do

275284 manip <- atomically $ do

276285 graph <- readTVar graphTvar

277286 eSession <- sessionForSessionId sessionId sessions

@@ -284,6 +293,12 @@ executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphE

284293 let newSession = Session discon'

285294 STMMap.insert newSession sessionId sessions

286295 pure $ Right graph'

296+

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

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

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

300+ freshUUID <- nextRandom

301+ manip <- atomically $ executeGraphExprSTM_ sessionId sessions graphTvar

287302 case manip of

288303 Left err -> return $ Just err

289304 Right newGraph -> do

@@ -292,10 +307,33 @@ executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphE

292307 processPersistence strat newGraph

293308 return Nothing

294309executeGraphExpr 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.

310+

311+commitSTM_ :: SessionId -> Sessions -> STM (Maybe RelationalError)

312+commitSTM_ sessionId sessions = executeGraphExprSTM_ sessionId sessions graph Commit

313+

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

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

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

316+commit sessionId conn@(InProcessConnection _ sessions graphTvar) = do

317+ mNots <- atomically $ do

318+ graph <- readTVar graphTvar

319+ -- take the notifications of the current context and scan them for changes in the previous -> current contexts

320+ eSession <- sessionForSessionId sessionId sessions

321+ case eSession of

322+ Left err -> pure (Left err)

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

324+ --grab previous context

325+ case transactionForUUID parentUUID graph of

326+ Left err -> pure $ Left err

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

328+ let nots = notifications currentContext

329+ fireNots = notificationChanges nots previousContext currentContext

330+ commitSTM_ sessionId sessions graph

331+ pure (Right fireNots)

332+ case mNots of

333+ Left err -> pure (Just err)

334+ Right notsToFire -> do

335+ --trigger the notifications on all clients

336+

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

300338

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

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

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

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

372410

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

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

412+headNameSTM_ sessionId sessions graphTvar= do

377413 graph <- readTVar graphTvar

378414 eSession <- sessionForSessionId sessionId sessions

379415 case eSession of

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

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

382418 Left _ -> Nothing

383419 Right parentTrans -> headNameForTransaction parentTrans graph

420+

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

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

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

424+ atomically (headNameSTM_ sessionId sessions graphTvar)

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

385426

386427

ProjectM36/DatabaseContext.hs+8−1

@@ -15,4 +15,11 @@ databaseContextAsDatabaseContextExpr (DatabaseContext incDeps relVars funcs) = M

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

1616 incDepsExprs :: [DatabaseExpr]

1717 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+ funcsExprs = [] -- map (\func -> ) (HS.toList funcs) -- there are no databaseExprs to add atom functions yet

19+

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

21+notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

22+notificationChanges notifications context1 context2 = M.filter notificationFilter notifications

23+ where

24+ notificationFilter (Notification changeExpr _) = evalChangeExpr changeExpr context1 /= evalChangeExpr changeExpr context2

25+ evalChangeExpr changeExpr = evalState (evalRelationalExpr changeExpr)

ProjectM36/RelationalExpression.hs+7−9

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

206206setRelVar relVarName rel = do

207207 currentContext <- get

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

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

209+ potentialContext = currentContext { relationVariables = newRelVars }

210210 case checkConstraints potentialContext of

211211 Just err -> return $ Just err

212212 Nothing -> do

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

216216-- 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

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

218218deleteRelVar relVarName = do

219− currstate <- get

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

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

219+ currcontext <- get

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

221+ put $ currcontext { relationVariables = newRelVars }

222222 return Nothing

223223

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

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

284284 if M.member newDepName currDeps then

285285 return $ Just (InclusionDependencyNameInUseError newDepName)

286286 else do

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

287+ let potentialContext = currContext { inclusionDependencies = newDeps }

288288 case checkConstraints potentialContext of

289289 Just err -> return $ Just err

290290 Nothing -> do

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

298298 if M.notMember depName currDeps then

299299 return $ Just (InclusionDependencyNameNotInUseError depName)

300300 else do

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

301+ put $ currContext {inclusionDependencies = newDeps }

302302 return Nothing

303303

304304evalContextExpr (MultipleExpr exprs) = do

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

360360--returns a database context with all tuples removed

361361--this is useful for type checking and optimization

362362contextWithEmptyTupleSets :: DatabaseContext -> DatabaseContext

363−contextWithEmptyTupleSets contextIn = DatabaseContext incDeps relVars funcs

363+contextWithEmptyTupleSets contextIn = contextIn { relationVariables = relVars }

364364 where

365− incDeps = inclusionDependencies contextIn

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

367− funcs = atomFunctions contextIn

368366

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

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

ProjectM36/Server/EntryPoints.hs+4−2

@@ -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@(Connection _ clientNodesTvar _ _) = do

29+

30+ reply True conn

2931

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

3133handleExecuteGraphExpr sessionId conn graphExpr = do

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+