57faeae · project-m36 · git.iko.soy

git.iko.soy/project-m36

on master help sign in

WIP compilation checkpoint

57faeaeAgentM9 years ago

project-m36.cabal+4−1

compare with master

unifiedsplit

@@ -66,6 +66,8 @@ Library

66 ,list-t

67 ,optparse-applicative

68 ,Glob

69+--used for hashing the transaction graph file to check for differences

70+ ,cryptonite >= 0.23

69 Exposed-Modules: ProjectM36.Error,

71 Exposed-Modules: ProjectM36.Error,

70 ProjectM36.Transaction,

72 ProjectM36.Transaction,

71 ProjectM36.TransactionGraph,

73 ProjectM36.TransactionGraph,

@@ -125,7 +127,8 @@ Library

125 ProjectM36.Sessions,

127 ProjectM36.Sessions,

126 ProjectM36.Transaction.Persist,

128 ProjectM36.Transaction.Persist,

127 ProjectM36.TypeConstructor,

129 ProjectM36.TypeConstructor,

128− ProjectM36.TypeConstructorDef

130+ ProjectM36.TypeConstructorDef,

131+ ProjectM36.FileLock

129 GHC-Options: -Wall

132 GHC-Options: -Wall

130 if os(windows)

133 if os(windows)

131 Build-Depends: Win32 >= 2.3 && < 2.4

134 Build-Depends: Win32 >= 2.3 && < 2.4

src/lib/ProjectM36/Client.hs+121−47

@@ -20,7 +20,7 @@ module ProjectM36.Client

20 typeForRelationalExpr,

21 inclusionDependencies,

22 planForDatabaseContextExpr,

23− processPersistence,

23+ processTransactionGraphPersistence,

24 currentSchemaName,

25 SchemaName,

26 HeadName,

@@ -101,6 +101,7 @@ import ProjectM36.TransactionGraph.Persist

101import ProjectM36.Attribute hiding (atomTypes)

102import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)

103import ProjectM36.Persist (DiskSync(..))

104+import ProjectM36.FileLock

104import ProjectM36.Notifications

105import ProjectM36.Notifications

105import ProjectM36.Server.RemoteCallTypes

106import ProjectM36.Server.RemoteCallTypes

106import qualified ProjectM36.DisconnectedTransaction as Discon

107import qualified ProjectM36.DisconnectedTransaction as Discon

@@ -108,6 +109,8 @@ import ProjectM36.Relation (typesAsRelation)

108import ProjectM36.ScriptSession (initScriptSession, ScriptSession)

109import ProjectM36.ScriptSession (initScriptSession, ScriptSession)

109import qualified ProjectM36.Relation as R

110import qualified ProjectM36.Relation as R

110import qualified ProjectM36.DatabaseContext as DBC

111import qualified ProjectM36.DatabaseContext as DBC

112+import Control.Exception.Base

113+import GHC.Conc.Sync

111

114

112import Network.Transport (Transport(closeTransport))

115import Network.Transport (Transport(closeTransport))

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

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

@@ -117,7 +120,6 @@ import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

117import Control.Distributed.Process (NodeId(..), reconnect)

120import Control.Distributed.Process (NodeId(..), reconnect)

118

121

119import Data.UUID.V4 (nextRandom)

122import Data.UUID.V4 (nextRandom)

120−import Control.Concurrent.STM

121import Data.Word

123import Data.Word

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

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

123import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)

125import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)

@@ -134,6 +136,7 @@ import Data.Binary (Binary)

134import GHC.Generics (Generic)

136import GHC.Generics (Generic)

135import Control.DeepSeq (force)

137import Control.DeepSeq (force)

136import System.IO

138import System.IO

139+import Crypto.Hash

137

140

138type Hostname = String

141type Hostname = String

139

142

@@ -195,6 +198,9 @@ defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (c

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

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

196type ClientNodes = STMSet.Set ProcessId

199type ClientNodes = STMSet.Set ProcessId

197

200

201+type TransactionGraphLockHandle = Handle

202+type LockFileHash = Digest SHA256

203+

198-- internal structure specific to in-process connections

204-- internal structure specific to in-process connections

199data InProcessConnectionConf = InProcessConnectionConf {

205data InProcessConnectionConf = InProcessConnectionConf {

200 ipPersistenceStrategy :: PersistenceStrategy,

206 ipPersistenceStrategy :: PersistenceStrategy,

@@ -203,7 +209,8 @@ data InProcessConnectionConf = InProcessConnectionConf {

203 ipTransactionGraph :: TVar TransactionGraph,

209 ipTransactionGraph :: TVar TransactionGraph,

204 ipScriptSession :: Maybe ScriptSession,

210 ipScriptSession :: Maybe ScriptSession,

205 ipLocalNode :: LocalNode,

211 ipLocalNode :: LocalNode,

206− ipTransport :: Transport -- we hold onto this so that we can close it gracefully

212+ ipTransport :: Transport, -- we hold onto this so that we can close it gracefully

213+ ipLocks :: Maybe (TransactionGraphLockHandle, MVar LockFileHash) -- nothing when NoPersistence

207 }

214 }

208

215

209data RemoteProcessConnectionConf = RemoteProcessConnectionConf {

216data RemoteProcessConnectionConf = RemoteProcessConnectionConf {

@@ -275,14 +282,16 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath

275 (localNode, transport) <- createLocalNode

282 (localNode, transport) <- createLocalNode

276 notificationPid <- startNotificationListener localNode notificationCallback

283 notificationPid <- startNotificationListener localNode notificationCallback

277 mScriptSession <- createScriptSession ghcPkgPaths

284 mScriptSession <- createScriptSession ghcPkgPaths

285+

278 let conn = InProcessConnection (InProcessConnectionConf {

286 let conn = InProcessConnection (InProcessConnectionConf {

279− ipPersistenceStrategy =strat,

287+ ipPersistenceStrategy = strat,

280 ipClientNodes = clientNodes,

288 ipClientNodes = clientNodes,

281 ipSessions = sessions,

289 ipSessions = sessions,

282 ipTransactionGraph = graphTvar,

290 ipTransactionGraph = graphTvar,

283 ipScriptSession = mScriptSession,

291 ipScriptSession = mScriptSession,

284 ipLocalNode = localNode,

292 ipLocalNode = localNode,

285− ipTransport = transport })

293+ ipTransport = transport,

294+ ipLocks = Nothing})

286 addClientNode conn notificationPid

295 addClientNode conn notificationPid

287 pure (Right conn)

296 pure (Right conn)

288 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths

297 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths

@@ -318,8 +327,8 @@ connectPersistentProjectM36 :: PersistenceStrategy ->

318connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do

327connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do

319 err <- setupDatabaseDir sync dbdir freshGraph

328 err <- setupDatabaseDir sync dbdir freshGraph

320 case err of

329 case err of

321− Just err' -> return $ Left (SetupDatabaseDirectoryError err')

330+ Left err' -> return $ Left (SetupDatabaseDirectoryError err')

322− Nothing -> do

331+ Right (lockFileH, digest) -> do

323 mScriptSession <- createScriptSession ghcPkgPaths

332 mScriptSession <- createScriptSession ghcPkgPaths

324 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession

333 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession

325 case graph of

334 case graph of

@@ -329,6 +338,7 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc

329 sessions <- STMMap.newIO

338 sessions <- STMMap.newIO

330 clientNodes <- STMSet.newIO

339 clientNodes <- STMSet.newIO

331 (localNode, transport) <- createLocalNode

340 (localNode, transport) <- createLocalNode

341+ lockMVar <- newMVar digest

332 let conn = InProcessConnection (InProcessConnectionConf {

342 let conn = InProcessConnection (InProcessConnectionConf {

333 ipPersistenceStrategy = strat,

343 ipPersistenceStrategy = strat,

334 ipClientNodes = clientNodes,

344 ipClientNodes = clientNodes,

@@ -336,7 +346,8 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc

336 ipTransactionGraph = tvarGraph,

346 ipTransactionGraph = tvarGraph,

337 ipScriptSession = mScriptSession,

347 ipScriptSession = mScriptSession,

338 ipLocalNode = localNode,

348 ipLocalNode = localNode,

339− ipTransport = transport

349+ ipTransport = transport,

350+ ipLocks = Just (lockFileH, lockMVar)

340 })

351 })

341

352

342 notificationPid <- startNotificationListener localNode notificationCallback

353 notificationPid <- startNotificationListener localNode notificationCallback

@@ -578,6 +589,7 @@ executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do

578 Right session -> case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of

589 Right session -> case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of

579 Left err -> pure $ Left err

590 Left err -> pure $ Left err

580 Right (discon', graph') -> do

591 Right (discon', graph') -> do

592+

581 writeTVar graphTvar graph'

593 writeTVar graphTvar graph'

582 let newSession = Session discon' (Sess.schemaName session)

594 let newSession = Session discon' (Sess.schemaName session)

583 STMMap.insert newSession sessionId sessions

595 STMMap.insert newSession sessionId sessions

@@ -591,53 +603,98 @@ executeCommitExprSTM_ oldContext newContext nodes = do

591 evaldNots = M.map mkEvaldNot fireNots

603 evaldNots = M.map mkEvaldNot fireNots

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

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

593 pure (evaldNots, nodes)

605 pure (evaldNots, nodes)

594−

606+

595−

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

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

608+-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file

609+-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM

610+-- perhaps keep hash of data file instead of checking if our head was updated on every write

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

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

598executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do

612executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do

599 let strat = ipPersistenceStrategy conf

613 let strat = ipPersistenceStrategy conf

600 clientNodes = ipClientNodes conf

614 clientNodes = ipClientNodes conf

601 sessions = ipSessions conf

615 sessions = ipSessions conf

602 graphTvar = ipTransactionGraph conf

616 graphTvar = ipTransactionGraph conf

617+ mLockFileH = ipLocks conf

618+ lockHandler body = case graphExpr of

619+ Commit -> case mLockFileH of

620+ Nothing -> pure Nothing

621+ Just (lockFileH, lockMVar) -> let acquireLocks = do

622+ lastWrittenDigest <- takeMVar lockMVar

623+ lockFile lockFileH WriteLock

624+ latestDigest <- readGraphTransactionIdDigest strat

625+ pure (latestDigest /= lastWrittenDigest)

626+

627+ releaseLocks _ = do

628+ --still holding the lock- get the latest digest

629+ gDigest <- readGraphTransactionIdDigest strat

630+ unlockFile lockFileH

631+ putMVar lockMVar gDigest

632+ in bracket acquireLocks releaseLocks body

633+ _ -> body False

603 freshId <- nextRandom

634 freshId <- nextRandom

635+ lockHandler $ \dbWrittenByOtherProcess -> do

636+ --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

637+ --read transaction data and compare to existing graph

638+ --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)

639+ if dbWrittenByOtherProcess then do

640+ handleGraphFromDiskUpdate sessionId conf

641+ else do

604− manip <- atomically $ do

642+ manip <- atomically $ do

643+ eSession <- sessionForSessionId sessionId sessions

644+ --handle graph update by other process

645+ oldGraph <- readTVar graphTvar

646+ case eSession of

647+ Left err -> pure (Left err)

648+ Right session -> do

649+ eGraph <- executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar

650+ case eGraph of

651+ Left err -> pure (Left err)

652+ Right newGraph -> do

653+ --handle commmit

654+ if graphExpr == Commit then

655+ case transactionForId (Sess.parentId session) oldGraph of

656+ Left err -> pure $ Left err

657+ Right previousTrans -> do

658+ (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes

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

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

661+ else

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

663+ case manip of

664+ Left err -> return $ Just err

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

666+ --update filesystem database, if necessary

667+ processTransactionGraphPersistence strat newGraph

668+ sendNotifications nodesToNotify (ipLocalNode conf) notsToFire

669+ pure Nothing

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

671+

672+-- | Called when the transaction graph on disk has been updated by a secondary process. Reject the commit and update the graph

673+handleGraphFromDiskUpdate :: SessionId -> InProcessConnectionConf -> IO (Maybe RelationalError)

674+handleGraphFromDiskUpdate sessionId conf = do

675+ let graphTVar = ipTransactionGraph conf

676+ strat = ipPersistenceStrategy conf

677+ sessions = ipSessions conf

678+ mScriptSession = ipScriptSession conf

679+ dbdir = case strat of

680+ MinimalPersistence x -> x

681+ CrashSafePersistence x -> x

682+ _ -> error "handleGraphFromDiskUpdate on unpersisted database"

683+ atomically $ do

605 eSession <- sessionForSessionId sessionId sessions

684 eSession <- sessionForSessionId sessionId sessions

606− oldGraph <- readTVar graphTvar

607 case eSession of

685 case eSession of

608− Left err -> pure (Left err)

686+ Left err -> pure (Just err)

609 Right session -> do

687 Right session -> do

610− eGraph <- executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar

688+ graph <- readTVar graphTVar

689+ let parentTid = Discon.parentId (disconnectedTransaction session)

690+ --this should be re-entrant because it only reads from the disk to create immutable structures

691+ --BUG: here, we unsafely assume that the updated graph is based on the original- that might not be true- we should check that the graph is sound and consistence with the previous graph

692+ eGraph <- unsafeIOToSTM (transactionGraphLoad dbdir graph mScriptSession)

611 case eGraph of

693 case eGraph of

612− Left err -> pure (Left err)

694+ Left err -> pure (Just (DatabaseLoadError err))

613− Right newGraph -> do

695+ Right updatedGraph -> do

614− if graphExpr == Commit then

696+ writeTVar graphTVar updatedGraph

615− case transactionForId (Sess.parentId session) oldGraph of

697+ pure (Just (TransactionIsNotAHeadError parentTid))

616− Left err -> pure $ Left err

617− Right previousTrans -> do

618− (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes

619− nodesToNotify <- toList (STMSet.stream nodes)

620− pure $ Right (evaldNots, nodesToNotify, newGraph)

621− else

622− pure $ Right (M.empty, [], newGraph)

623− case manip of

624− Left err -> return $ Just err

625− Right (notsToFire, nodesToNotify, newGraph) -> do

626− --update filesystem database, if necessary

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

628− processPersistence strat newGraph

629− sendNotifications nodesToNotify (ipLocalNode conf) notsToFire

630− return Nothing

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

632−

633−{-

634−commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)

635−commitSTM_ freshUUID sessionId sessions graph = do

636− mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph

637− pure $ case mErr of

638− Left err -> Just err

639− Right _ -> Nothing

640−-}

641

698

642-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.

699-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.

643executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)

700executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)

@@ -686,10 +743,27 @@ rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

686rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback

743rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback

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

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

688

745

689−processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

746+processTransactionGraphPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

690−processPersistence NoPersistence _ = return ()

747+processTransactionGraphPersistence NoPersistence _ = pure ()

691−processPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph

748+processTransactionGraphPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph >> pure ()

692−processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph

749+processTransactionGraphPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph >> pure ()

750+

751+readGraphTransactionIdDigest :: PersistenceStrategy -> IO (Digest SHA256)

752+readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"

753+readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir

754+readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir

755+

756+{-

757+sameGraphTransactionIdDigest :: PersistenceStrategy -> Digest SHA256 -> IO (Bool)

758+sameGraphTransactionIdDigest NoPersistence _ = pure True

759+sameGraphTransactionIdDigest persist lastDigest = do

760+ let dbdir = case persist of

761+ MinimalPersistence x -> x

762+ CrashSafePersistence x -> x

763+ newDigest <- readGraphTransactionIdDigest persist

764+ pure (lastDigest == newDigest)

765+

766+-}

693

767

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

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

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

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

src/lib/ProjectM36/DataTypes/DateTime.hs−1

@@ -3,7 +3,6 @@ import ProjectM36.Base

3import ProjectM36.AtomFunctionBody

4import qualified Data.HashSet as HS

5import Data.Time.Clock.POSIX

6−import Data.Time.Clock

8dateTimeAtomFunctions :: AtomFunctions

7dateTimeAtomFunctions :: AtomFunctions

9dateTimeAtomFunctions = HS.fromList [ AtomFunction {

8dateTimeAtomFunctions = HS.fromList [ AtomFunction {

src/lib/ProjectM36/Error.hs+2−1

@@ -81,6 +81,7 @@ data RelationalError = NoSuchAttributeNamesError (S.Set AttributeName)

81 | MergeTransactionError MergeError

82 | ScriptError ScriptCompilationError

83 | DatabaseContextFunctionUserError DatabaseContextFunctionError

84+ | DatabaseLoadError PersistenceError

85 | SubschemaNameInUseError SchemaName

86 | SubschemaNameInUseError SchemaName

86 | SubschemaNameNotInUseError SchemaName

87 | SubschemaNameNotInUseError SchemaName

@@ -94,7 +95,7 @@ data RelationalError = NoSuchAttributeNamesError (S.Set AttributeName)

95data PersistenceError = InvalidDirectoryError FilePath |

96data PersistenceError = InvalidDirectoryError FilePath |

96 MissingTransactionError TransactionId

97 MissingTransactionError TransactionId

97− deriving (Show, Eq, Generic)

98+ deriving (Show, Eq, Generic, Binary, NFData)

99--collapse list of errors into normal error- if there is just one, just return one

100--collapse list of errors into normal error- if there is just one, just return one

100someErrors :: [RelationalError] -> RelationalError

101someErrors :: [RelationalError] -> RelationalError

src/lib/ProjectM36/FileLock.hsnew file+36

@@ -0,0 +1,36 @@

1+--cross-platform file locking utilizing POSIX file locking on Unix/Linux and Windows file locking

2+--hackage's System.FileLock doesn't support POSIX advisory locks nor locking file based on file descriptors, hence this needless rewrite

3+module ProjectM36.FileLock where

5+#if defined(linux_HOST_OS)

6+import qualified System.Posix.IO as P

7+import System.IO

9+lockStruct :: P.LockRequest -> P.FileLock

10+lockStruct req = (req, AbsoluteSeek, 0, 0)

11+

12+--blocks on lock, if necessary

13+lockFile :: Handle -> LockType -> IO ()

14+lockFile file lock = do

15+ fd <- P.handleToFd file

16+ let lockt = case lock of

17+ WriteLock -> P.WriteLock

18+ ReadLock -> P.ReadLock

19+ P.waitToSetLock fd (lockStruct lockt)

20+

21+unlockFile :: Handle -> IO ()

22+unlockFile file = do

23+ fd <- P.handleToFd file

24+ P.waitToSetLock fd (lockStruct P.Unlock)

25+#endif

26+

27+data LockType = ReadLock | WriteLock

28+

29+{-

30+lockFileSTM :: Handle -> LockType -> STM ()

31+lockFileSTM file lock = unsafeIOToSTM $ onException (lockFile file lock) (unlockFile file)

32+

33+unlockFileSTM :: Handle -> STM ()

34+unlockFileSTM file = unsafeIOToSTM $ unlockFile file

35+-}

36+

src/lib/ProjectM36/Persist.hs+5−4

@@ -5,7 +5,8 @@ module ProjectM36.Persist (writeFileSync,

5 renameSync,

6 DiskSync(..)) where

7-- on Windows, use FlushFileBuffers and MoveFileEx

8−

8+import qualified Data.Text.IO as TIO

9+import Data.Text

9#if defined(mingw32_HOST_OS)

10#if defined(mingw32_HOST_OS)

10import qualified System.Win32 as Win32

11import qualified System.Win32 as Win32

11#else

12#else

@@ -19,7 +20,7 @@ import System.Posix.IO (handleToFd)

19import Foreign.C

20import Foreign.C

20#endif

21#endif

22−import System.IO (withFile, IOMode(WriteMode), hPutStr, Handle)

23+import System.IO (withFile, IOMode(WriteMode), Handle)

23import qualified Data.ByteString.Lazy as BS

24import qualified Data.ByteString.Lazy as BS

25#if defined(mingw32_HOST_OS)

26#if defined(mingw32_HOST_OS)

@@ -30,11 +31,11 @@ foreign import ccall unsafe "cDirectoryFsync" cHSDirectoryFsync :: CString -> IO

31data DiskSync = NoDiskSync | FsyncDiskSync

32data DiskSync = NoDiskSync | FsyncDiskSync

33−writeFileSync :: DiskSync -> FilePath -> String -> IO()

34+writeFileSync :: DiskSync -> FilePath -> Text -> IO()

34writeFileSync sync path strOut = withFile path WriteMode handler

35writeFileSync sync path strOut = withFile path WriteMode handler

35 where

36 where

36 handler handle = do

37 handler handle = do

37− hPutStr handle strOut

38+ TIO.hPutStr handle strOut

38 syncHandle sync handle

39 syncHandle sync handle

40renameSync :: DiskSync -> FilePath -> FilePath -> IO ()

41renameSync :: DiskSync -> FilePath -> FilePath -> IO ()

src/lib/ProjectM36/Transaction/Persist.hs+1−3

@@ -19,8 +19,6 @@ import Control.Exception

19import GHC

20import GHC.Paths

22−import Debug.Trace

23−

24getDirectoryNames :: FilePath -> IO [FilePath]

22getDirectoryNames :: FilePath -> IO [FilePath]

25getDirectoryNames path = do

23getDirectoryNames path = do

26 subpaths <- getDirectoryContents path

24 subpaths <- getDirectoryContents path

@@ -164,7 +162,7 @@ writeDBCFuncs sync transDir funcs = mapM_ (writeDBCFunc sync transDir) (HS.toLis

164writeDBCFunc :: DiskSync -> FilePath -> DatabaseContextFunction -> IO ()

162writeDBCFunc :: DiskSync -> FilePath -> DatabaseContextFunction -> IO ()

165writeDBCFunc sync transDir func = do

163writeDBCFunc sync transDir func = do

166 let dbcFuncPath = dbcFuncsDir transDir </> T.unpack (dbcFuncName func)

164 let dbcFuncPath = dbcFuncsDir transDir </> T.unpack (dbcFuncName func)

167− writeBSFileSync sync (traceShowId dbcFuncPath) (B.encode (dbcFuncType func, databaseContextFunctionScript func))

165+ writeBSFileSync sync dbcFuncPath (B.encode (dbcFuncType func, databaseContextFunctionScript func))

168

166

169readDBCFuncs :: FilePath -> Maybe ScriptSession -> IO DatabaseContextFunctions

167readDBCFuncs :: FilePath -> Maybe ScriptSession -> IO DatabaseContextFunctions

170readDBCFuncs transDir mScriptSession = do

168readDBCFuncs transDir mScriptSession = do

src/lib/ProjectM36/TransactionGraph.hs+4−4

@@ -17,7 +17,6 @@ import GHC.Generics

17import Data.Binary

18import ProjectM36.TransactionGraph.Merge

19import Data.Either (lefts, rights, isRight)

20−--import Debug.Trace

22-- | Record a lookup for a specific transaction in the graph.

21-- | Record a lookup for a specific transaction in the graph.

23data TransactionIdLookup = TransactionIdLookup TransactionId |

22data TransactionIdLookup = TransactionIdLookup TransactionId |

@@ -55,7 +54,10 @@ emptyTransactionGraph :: TransactionGraph

55emptyTransactionGraph = TransactionGraph M.empty S.empty

54emptyTransactionGraph = TransactionGraph M.empty S.empty

57transactionForHead :: HeadName -> TransactionGraph -> Maybe Transaction

56transactionForHead :: HeadName -> TransactionGraph -> Maybe Transaction

58−transactionForHead headName (TransactionGraph heads _) = M.lookup headName heads

57+transactionForHead headName graph = M.lookup headName (transactionHeadsForGraph graph)

58+

59+headList :: TransactionGraph -> [(HeadName, TransactionId)]

60+headList graph = map (\(k,v) -> (k, transactionId v)) (M.assocs (transactionHeadsForGraph graph))

60headNameForTransaction :: Transaction -> TransactionGraph -> Maybe HeadName

62headNameForTransaction :: Transaction -> TransactionGraph -> Maybe HeadName

61headNameForTransaction transaction (TransactionGraph heads _) = if M.null matchingTrans then

63headNameForTransaction transaction (TransactionGraph heads _) = if M.null matchingTrans then

@@ -186,7 +188,6 @@ walkChildTransactions seenTransSet graph trans =

186-- the current transaction is not part of the transaction graph until it is committed

188-- the current transaction is not part of the transaction graph until it is committed

187evalGraphOp :: TransactionId -> DisconnectedTransaction -> TransactionGraph -> TransactionGraphOperator -> Either RelationalError (DisconnectedTransaction, TransactionGraph)

189evalGraphOp :: TransactionId -> DisconnectedTransaction -> TransactionGraph -> TransactionGraphOperator -> Either RelationalError (DisconnectedTransaction, TransactionGraph)

188

190

189−--affects only disconncted transaction

190evalGraphOp _ _ graph (JumpToTransaction jumpId) = case transactionForId jumpId graph of

191evalGraphOp _ _ graph (JumpToTransaction jumpId) = case transactionForId jumpId graph of

191 Left err -> Left err

192 Left err -> Left err

192 Right parentTrans -> Right (newTrans, graph)

193 Right parentTrans -> Right (newTrans, graph)

@@ -194,7 +195,6 @@ evalGraphOp _ _ graph (JumpToTransaction jumpId) = case transactionForId jumpId

194 newTrans = DisconnectedTransaction jumpId (schemas parentTrans)

195 newTrans = DisconnectedTransaction jumpId (schemas parentTrans)

195

196

196-- switch from one head to another

197-- switch from one head to another

197−-- affects only disconnectedtransaction

198evalGraphOp _ _ graph (JumpToHead headName) =

199 case transactionForHead headName graph of

200 Just newHeadTransaction -> let disconnectedTrans = DisconnectedTransaction (transactionId newHeadTransaction) (schemas newHeadTransaction) in

src/lib/ProjectM36/TransactionGraph/Persist.hs+57−27

@@ -6,17 +6,23 @@ import ProjectM36.Transaction.Persist

6import ProjectM36.Base

7import ProjectM36.ScriptSession

8import ProjectM36.Persist (writeFileSync, renameSync, DiskSync)

9+import ProjectM36.FileLock

9import System.Directory

10import System.Directory

10import System.FilePath

11import System.FilePath

11import System.IO.Temp

12import System.IO.Temp

12−import Data.List

13+import System.IO

13import qualified Data.UUID as U

14import qualified Data.UUID as U

14import qualified Data.Set as S

15import qualified Data.Set as S

15import qualified Data.Map as M

16import qualified Data.Map as M

16import qualified Data.Text as T

17import qualified Data.Text as T

18+import Data.Text.Encoding

17import Control.Monad (foldM)

19import Control.Monad (foldM)

18import Data.Either (isRight)

20import Data.Either (isRight)

19import Data.Maybe (catMaybes)

21import Data.Maybe (catMaybes)

22+import Control.Exception.Base

23+import Crypto.Hash

24+import qualified Data.Text.IO as TIO

25+import Data.Monoid

21{-

27{-

22The "m36v1" file at the top-level of the destination directory contains the the transaction graph as a set of transaction ids referencing their parents (1 or more)

28The "m36v1" file at the top-level of the destination directory contains the the transaction graph as a set of transaction ids referencing their parents (1 or more)

@@ -31,44 +37,63 @@ transactionLogPath dbdir = dbdir </> "m36v1"

31headsPath :: FilePath -> FilePath

37headsPath :: FilePath -> FilePath

32headsPath dbdir = dbdir </> "heads"

38headsPath dbdir = dbdir </> "heads"

40+lockFilePath :: FilePath -> FilePath

41+lockFilePath dbdir = dbdir </> "lockFile"

42+

34{-

43{-

35verify that the database directory is valid or bootstrap it

44verify that the database directory is valid or bootstrap it

36-note: checking for the existence of every transaction may be prohibitively expensive

45-note: checking for the existence of every transaction may be prohibitively expensive

46+- return error or lock file handle which is already locked with a read lock

37-}

47-}

39−setupDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Maybe PersistenceError)

49+setupDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Either PersistenceError (Handle, Digest SHA256))

40setupDatabaseDir sync dbdir bootstrapGraph = do

50setupDatabaseDir sync dbdir bootstrapGraph = do

41 dbdirExists <- doesDirectoryExist dbdir

51 dbdirExists <- doesDirectoryExist dbdir

42 m36exists <- doesFileExist (transactionLogPath dbdir)

52 m36exists <- doesFileExist (transactionLogPath dbdir)

43− if dbdirExists && m36exists then

53+ if dbdirExists && m36exists then do

44− return Nothing

54+ --no directories to write, just

55+ lockFileH <- openLockFile dbdir

56+ gDigest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (readGraphTransactionIdFileDigest dbdir)

57+ pure (Right (lockFileH, gDigest))

45 else if not m36exists then do

58 else if not m36exists then do

46− bootstrapDatabaseDir sync dbdir bootstrapGraph

59+ locks <- bootstrapDatabaseDir sync dbdir bootstrapGraph

47− return Nothing

60+ pure (Right locks)

48 else

61 else

49− return $ Just (InvalidDirectoryError dbdir)

62+ pure (Left (InvalidDirectoryError dbdir))

50−

51{-

63{-

52−initialize a database directory with the graph from which to bootstrap

64+initialize a database directory with the graph from which to bootstrap- return lock file handle

53-}

65-}

54−bootstrapDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO ()

66+bootstrapDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Handle, Digest SHA256)

55bootstrapDatabaseDir sync dbdir bootstrapGraph = do

67bootstrapDatabaseDir sync dbdir bootstrapGraph = do

56 createDirectory dbdir

68 createDirectory dbdir

69+ lockFileH <- openLockFile dbdir

57− transactionGraphPersist sync dbdir bootstrapGraph

70+ digest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (transactionGraphPersist sync dbdir bootstrapGraph)

71+ pure (lockFileH, digest)

72+

73+openLockFile :: FilePath -> IO (Handle)

74+openLockFile dbdir = do

75+ lockFileH <- openFile (lockFilePath dbdir) WriteMode

76+ pure lockFileH

59{-

78{-

60incrementally updates an existing database directory

79incrementally updates an existing database directory

61--algorithm:

80--algorithm:

62−-check that all heads have been written

81+--assume that all transaction data has already been written

63-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method

82-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method

64--store the head names with a symlink to the transaction under "heads"

83--store the head names with a symlink to the transaction under "heads"

65-}

84-}

66−transactionGraphPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

85+transactionGraphPersist :: DiskSync -> FilePath -> TransactionGraph -> IO (Digest SHA256)

67transactionGraphPersist sync destDirectory graph = do

86transactionGraphPersist sync destDirectory graph = do

68− mapM_ (writeTransaction sync destDirectory) $ M.elems (transactionHeadsForGraph graph)

87+ transactionHeadTransactionsPersist sync destDirectory graph

88+ --write graph file

69− writeGraphTransactionIdFile sync destDirectory graph

89+ newDigest <- writeGraphTransactionIdFile sync destDirectory graph

90+ --write heads file

70 transactionGraphHeadsPersist sync destDirectory graph

91 transactionGraphHeadsPersist sync destDirectory graph

71− return ()

92+ pure newDigest

93+

94+-- | The incremental writer which only writes from the set of heads. New heads must be written on every commit. Most heads will already be written on every commit.

95+transactionHeadTransactionsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

96+transactionHeadTransactionsPersist sync destDirectory graphIn = mapM_ (writeTransaction sync destDirectory) $ M.elems (transactionHeadsForGraph graphIn)

73{-

98{-

74write graph heads to a file which can be atomically swapped

99write graph heads to a file which can be atomically swapped

@@ -76,12 +101,12 @@ write graph heads to a file which can be atomically swapped

76--writing the heads in a directory is a synchronization nightmare, so just write the binary to a file and swap atomically

101--writing the heads in a directory is a synchronization nightmare, so just write the binary to a file and swap atomically

77transactionGraphHeadsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

102transactionGraphHeadsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

78transactionGraphHeadsPersist sync dbdir graph = do

103transactionGraphHeadsPersist sync dbdir graph = do

79− let headFileStr :: (HeadName, Transaction) -> String

104+ let headFileStr :: (HeadName, Transaction) -> T.Text

80− headFileStr (headName, trans) = T.unpack headName ++ " " ++ U.toString (transactionId trans)

105+ headFileStr (headName, trans) = headName <> " " <> U.toText (transactionId trans)

81 withTempDirectory dbdir ".heads.tmp" $ \tempHeadsDir -> do

106 withTempDirectory dbdir ".heads.tmp" $ \tempHeadsDir -> do

82 let tempHeadsPath = tempHeadsDir </> "heads"

107 let tempHeadsPath = tempHeadsDir </> "heads"

83 headsStrLines = map headFileStr $ M.toList (transactionHeadsForGraph graph)

108 headsStrLines = map headFileStr $ M.toList (transactionHeadsForGraph graph)

84− writeFileSync sync tempHeadsPath $ intercalate "\n" headsStrLines

109+ writeFileSync sync tempHeadsPath $ T.intercalate "\n" headsStrLines

85 renameSync sync tempHeadsPath (headsPath dbdir)

110 renameSync sync tempHeadsPath (headsPath dbdir)

111

87transactionGraphHeadsLoad :: FilePath -> IO [(HeadName,TransactionId)]

112transactionGraphHeadsLoad :: FilePath -> IO [(HeadName,TransactionId)]

@@ -96,7 +121,6 @@ transactionGraphHeadsLoad dbdir = do

96load any transactions which are not already part of the incoming transaction graph

121load any transactions which are not already part of the incoming transaction graph

97-}

122-}

123

99−--ALERT I need to figure out how to manage the transaction heads (branch names)

100transactionGraphLoad :: FilePath -> TransactionGraph -> Maybe ScriptSession -> IO (Either PersistenceError TransactionGraph)

124transactionGraphLoad :: FilePath -> TransactionGraph -> Maybe ScriptSession -> IO (Either PersistenceError TransactionGraph)

101transactionGraphLoad dbdir graphIn mScriptSession = do

125transactionGraphLoad dbdir graphIn mScriptSession = do

102 --optimization: perform tail-bisection search to find last-recorded transaction in the existing stream- replay the rest

126 --optimization: perform tail-bisection search to find last-recorded transaction in the existing stream- replay the rest

@@ -131,20 +155,26 @@ readTransactionIfNecessary dbdir transId mScriptSession graphIn = do

131 Left err -> return $ Left err

155 Left err -> return $ Left err

132 Right trans' -> return $ Right $ TransactionGraph (transactionHeadsForGraph graphIn) (S.insert trans' (transactionsForGraph graphIn))

156 Right trans' -> return $ Right $ TransactionGraph (transactionHeadsForGraph graphIn) (S.insert trans' (transactionsForGraph graphIn))

133

157

134−writeGraphTransactionIdFile :: DiskSync -> FilePath -> TransactionGraph -> IO ()

158+writeGraphTransactionIdFile :: DiskSync -> FilePath -> TransactionGraph -> IO (Digest SHA256)

135−writeGraphTransactionIdFile sync destDirectory (TransactionGraph _ transSet) = writeFileSync sync graphFile uuidInfo

159+writeGraphTransactionIdFile sync destDirectory (TransactionGraph _ transSet) = writeFileSync sync graphFile uuidInfo >> pure digest

136 where

160 where

137 graphFile = destDirectory </> "m36v1"

161 graphFile = destDirectory </> "m36v1"

138− uuidInfo = intercalate "\n" graphLines

162+ uuidInfo = T.intercalate "\n" graphLines

163+ digest = hashWith SHA256 (encodeUtf8 uuidInfo)

139 graphLines = S.toList $ S.map graphLine transSet

164 graphLines = S.toList $ S.map graphLine transSet

140− graphLine trans = U.toString (transactionId trans) ++ " " ++ intercalate " " (S.toList (S.map U.toString $ transactionParentIds trans))

165+ graphLine trans = U.toText (transactionId trans) <> " " <> T.intercalate " " (S.toList (S.map U.toText $ transactionParentIds trans))

166+

167+readGraphTransactionIdFileDigest :: FilePath -> IO (Digest SHA256)

168+readGraphTransactionIdFileDigest dbdir = do

169+ graphTransactionIdData <- TIO.readFile (transactionLogPath dbdir)

170+ pure (hashWith SHA256 (encodeUtf8 graphTransactionIdData))

141

171

142readGraphTransactionIdFile :: FilePath -> IO (Either PersistenceError [(TransactionId, [TransactionId])])

172readGraphTransactionIdFile :: FilePath -> IO (Either PersistenceError [(TransactionId, [TransactionId])])

143readGraphTransactionIdFile dbdir = do

173readGraphTransactionIdFile dbdir = do

144 --read in all transactions' uuids

174 --read in all transactions' uuids

145− let grapher line = let tids = catMaybes (map U.fromString (words line)) in

175+ let grapher line = let tids = catMaybes (map U.fromText (T.words line)) in

146 (head tids, tail tids)

176 (head tids, tail tids)

147 --warning: uses lazy IO

177 --warning: uses lazy IO

148− graphTransactionIdData <- readFile (transactionLogPath dbdir)

178+ graphTransactionIdData <- TIO.readFile (transactionLogPath dbdir)

149− return $ Right (map grapher $ lines graphTransactionIdData)

179+ return $ Right (map grapher $ T.lines graphTransactionIdData)

150

180