{-# LANGUAGE DeriveDataTypeable, BangPatterns #-}------------------------------------------------------------------------------- |-- Module : Data.Acid.Local-- Copyright : PublicDomain---- Maintainer : lemmih@gmail.com-- Portability : non-portable (uses GHC extensions)---- AcidState container using a transaction log on disk. The term \'Event\' is-- loosely used for transactions with ACID guarantees. \'Method\' is loosely-- used for state operations without ACID guarantees (see "Data.Acid.Core").--moduleData.Acid.Local(openLocalState,openLocalStateFrom,createArchive,createCheckpointAndClose)whereimportData.Acid.LogasLogimportData.Acid.CoreimportData.Acid.CommonimportData.Acid.AbstractimportControl.Concurrent(newEmptyMVar,putMVar,takeMVar,MVar)--import Control.Exception ( evaluate )importControl.Monad.State(runState)importControl.Applicative((<$>),(<*>))importData.ByteString.Lazy(ByteString)--import qualified Data.ByteString.Lazy as Lazy ( length )importData.Serialize(runPutLazy,runGetLazy)importData.SafeCopy(SafeCopy(..),safeGet,safePut,primitive,contain)importData.Typeable(Typeable,typeOf)importSystem.FilePath((</>)){-| State container offering full ACID (Atomicity, Consistency, Isolation and Durability)
guarantees.
[@Atomicity@] State changes are all-or-nothing. This is what you'd expect of any state
variable in Haskell and AcidState doesn't change that.
[@Consistency@] No event or set of events will break your data invariants.
[@Isolation@] Transactions cannot interfere with each other even when issued in parallel.
[@Durability@] Successful transaction are guaranteed to survive system failure (both
hardware and software).
-}dataLocalStatest=LocalState{localCore::Corest,localEvents::FileLog(TaggedByteString),localCheckpoints::FileLogCheckpoint}deriving(Typeable)-- | Issue an Update event and return immediately. The event is not durable-- before the MVar has been filled but the order of events is honored.-- The behavior in case of exceptions is exactly the same as for 'update'.---- If EventA is scheduled before EventB, EventA /will/ be executed before EventB:---- @--do scheduleUpdate acid EventA-- scheduleUpdate acid EventB-- @scheduleLocalUpdate::UpdateEventevent=>LocalState(EventStateevent)->event->IO(MVar(EventResultevent))scheduleLocalUpdateacidStateevent=domvar<-newEmptyMVarletencoded=runPutLazy(safePutevent)--evaluate (Lazy.length encoded) -- It would be best to encode the event before we lock the core-- but it hurts performance /-:modifyCoreState_(localCoreacidState)$\st->dolet!(result,!st')=runStatehotMethodst-- Schedule the log entry. Very important that it happens when 'localCore' is locked-- to ensure that events are logged in the same order that they are executed.pushEntry(localEventsacidState)(methodTagevent,encoded)$putMVarmvarresultreturnst'returnmvarwherehotMethod=lookupHotMethod(coreMethods(localCoreacidState))eventscheduleLocalColdUpdate::LocalStatest->TaggedByteString->IO(MVarByteString)scheduleLocalColdUpdateacidStateevent=domvar<-newEmptyMVarmodifyCoreState_(localCoreacidState)$\st->dolet!(result,!st')=runStatecoldMethodst-- Schedule the log entry. Very important that it happens when 'localCore' is locked-- to ensure that events are logged in the same order that they are executed.pushEntry(localEventsacidState)event$putMVarmvarresultreturnst'returnmvarwherecoldMethod=lookupColdMethod(localCoreacidState)event-- | Issue a Query event and wait for its result. Events may be issued in parallel.localQuery::QueryEventevent=>LocalState(EventStateevent)->event->IO(EventResultevent)localQueryacidStateevent=domvar<-newEmptyMVarwithCoreState(localCoreacidState)$\st->dolet(result,_st)=runStatehotMethodst-- Make sure that we do not return the result before the event log has-- been flushed to disk.pushAction(localEventsacidState)$putMVarmvarresulttakeMVarmvarwherehotMethod=lookupHotMethod(coreMethods(localCoreacidState))event-- Whoa, a buttload of refactoring is needed here. 2011-11-02localQueryCold::LocalStatest->TaggedByteString->IOByteStringlocalQueryColdacidStateevent=domvar<-newEmptyMVarwithCoreState(localCoreacidState)$\st->dolet(result,_st)=runStatecoldMethodst-- Make sure that we do not return the result before the event log has-- been flushed to disk.pushAction(localEventsacidState)$putMVarmvarresulttakeMVarmvarwherecoldMethod=lookupColdMethod(localCoreacidState)event-- | Take a snapshot of the state and save it to disk. Creating checkpoints-- makes it faster to resume AcidStates and you're free to create them as-- often or seldom as fits your needs. Transactions can run concurrently-- with this call.---- This call will not return until the operation has succeeded.createLocalCheckpoint::SafeCopyst=>LocalStatest->IO()createLocalCheckpointacidState=domvar<-newEmptyMVarwithCoreState(localCoreacidState)$\st->doeventId<-askCurrentEntryId(localEventsacidState)pushAction(localEventsacidState)$doletencoded=runPutLazy(safePutst)pushEntry(localCheckpointsacidState)(CheckpointeventIdencoded)(putMVarmvar())takeMVarmvar-- | Save a snapshot to disk and close the AcidState as a single atomic-- action. This is useful when you want to make sure that no events-- are saved to disk after a checkpoint.createCheckpointAndClose::SafeCopyst=>AcidStatest->IO()createCheckpointAndCloseabstract_state=domvar<-newEmptyMVarcloseCore'(localCoreacidState)$\st->doeventId<-askCurrentEntryId(localEventsacidState)pushAction(localEventsacidState)$pushEntry(localCheckpointsacidState)(CheckpointeventId(runPutLazy(safePutst)))(putMVarmvar())takeMVarmvarcloseFileLog(localEventsacidState)closeFileLog(localCheckpointsacidState)whereacidState=downcastabstract_statedataCheckpoint=CheckpointEntryIdByteStringinstanceSafeCopyCheckpointwherekind=primitiveputCopy(CheckpointeventEntryIdcontent)=contain$dosafePuteventEntryIdsafePutcontentgetCopy=contain$Checkpoint<$>safeGet<*>safeGet-- | Create an AcidState given an initial value.---- This will create or resume a log found in the \"state\/[typeOf state]\/\" directory.openLocalState::(Typeablest,IsAcidicst)=>st-- ^ Initial state value. This value is only used if no checkpoint is-- found.->IO(AcidStatest)openLocalStateinitialState=openLocalStateFrom("state"</>show(typeOfinitialState))initialState-- | Create an AcidState given a log directory and an initial value.---- This will create or resume a log found in @directory@.-- Running two AcidState's from the same directory is an error-- but will not result in dataloss.openLocalStateFrom::(IsAcidicst)=>FilePath-- ^ Location of the checkpoint and transaction files.->st-- ^ Initial state value. This value is only used if no checkpoint is-- found.->IO(AcidStatest)openLocalStateFromdirectoryinitialState=docore<-mkCore(eventsToMethodsacidEvents)initialStateleteventsLogKey=LogKey{logDirectory=directory,logPrefix="events"}checkpointsLogKey=LogKey{logDirectory=directory,logPrefix="checkpoints"}mbLastCheckpoint<-Log.newestEntrycheckpointsLogKeyn<-casembLastCheckpointofNothing->return0Just(CheckpointeventCutOffcontent)->domodifyCoreState_core(\_oldState->caserunGetLazysafeGetcontentofLeftmsg->checkpointRestoreErrormsgRightval->returnval)returneventCutOffeventsLog<-openFileLogeventsLogKeyevents<-readEntriesFromeventsLognmapM_(runColdMethodcore)eventscheckpointsLog<-openFileLogcheckpointsLogKeyreturn$toAcidStateLocalState{localCore=core,localEvents=eventsLog,localCheckpoints=checkpointsLog}checkpointRestoreErrormsg=error$"Could not parse saved checkpoint due to the following error: "++msg-- | Close an AcidState and associated logs.-- Any subsequent usage of the AcidState will throw an exception.closeLocalState::LocalStatest->IO()closeLocalStateacidState=docloseCore(localCoreacidState)closeFileLog(localEventsacidState)closeFileLog(localCheckpointsacidState)-- | Move all log files that are no longer necessary for state restoration into the 'Archive'-- folder in the state directory. This folder can then be backed up or thrown out as you see fit.-- Reverting to a state before the last checkpoint will not be possible if the 'Archive' folder-- has been thrown out.-- -- This method is idempotent and does not block the normal operation of the AcidState.createArchive::AcidStatest->IO()createArchiveabstract_state=do-- We need to look at the last checkpoint saved to disk. Since checkpoints can be written-- in parallel with this call, we can't guarantee that the checkpoint we get really is the-- last one but that's alright.currentCheckpointId<-cutFileLog(localCheckpointsstate)-- 'currentCheckpointId' is the ID of the next checkpoint that will be written to disk.-- 'currentCheckpointId-1' must then be the ID of a checkpoint on disk (or -1, of course).letdurableCheckpointId=currentCheckpointId-1checkpoints<-readEntriesFrom(localCheckpointsstate)durableCheckpointIdcasecheckpointsof[]->return()(CheckpointentryId_content:_)->do-- 'entryId' is the lowest entryId that didn't contribute to the checkpoint.-- 'archiveFileLog' moves all files that are lower than this entryId to the archive.archiveFileLog(localEventsstate)entryId-- In the same style as above, we archive all log files that came before the log file-- which contains our checkpoint.archiveFileLog(localCheckpointsstate)durableCheckpointIdwherestate=downcastabstract_statetoAcidState::IsAcidicst=>LocalStatest->AcidStatesttoAcidStatelocal=AcidState{_scheduleUpdate=scheduleLocalUpdatelocal,scheduleColdUpdate=scheduleLocalColdUpdatelocal,_query=localQuerylocal,queryCold=localQueryColdlocal,createCheckpoint=createLocalCheckpointlocal,closeAcidState=closeLocalStatelocal,acidSubState=mkAnyStatelocal}