Rakka/Storage/Impl.hs

   1 module Rakka.Storage.Impl
   2     ( getPage'
   3     , startIndexManager
   4     )
   5     where
   6
   7 import           Control.Concurrent
   8 import           Control.Concurrent.STM
   9 import           Control.Exception
  10 import           Control.Monad
  11 import           Data.Maybe
  12 import           Data.Set (Set)
  13 import qualified Data.Set as S
  14 import           Network.URI
  15 import           Rakka.Page
  16 import           Rakka.Storage.DefaultPage
  17 import           Rakka.Storage.Repos
  18 import           Rakka.Storage.Types
  19 import           Subversion.Types
  20 import           Subversion.FileSystem
  21 import           Subversion.Repository
  22 import           System.Directory
  23 import           System.FilePath
  24 import           System.IO
  25 import           System.Log.Logger
  26 import           System.Posix.Files
  27 import           System.Posix.Types
  28 import           System.Posix.IO
  29 import           Text.HyperEstraier hiding (WriteLock)
  30
  31
  32 logger :: String
  33 logger = "Rakka.Storage"
  34
  35
  36 getPage' :: Repository -> PageName -> Maybe RevNum -> IO (Maybe Page)
  37 getPage' repos name rev
  38     = do page <- loadPageInRepository repos name rev
  39          case page of
  40            Nothing -> loadDefaultPage name
  41            p       -> return p
  42
  43
  44 findAllPages :: Repository -> RevNum -> IO (Set PageName)
  45 findAllPages _     0   = findAllDefaultPages
  46 findAllPages repos rev = do reposPages   <- findAllPagesInRevision repos rev
  47                             defaultPages <- findAllDefaultPages
  48                             return (reposPages `S.union` defaultPages)
  49
  50
  51 findChangedPages :: Repository -> RevNum -> RevNum -> IO (Set PageName)
  52 findChangedPages repos 0      newRev = findAllPages repos newRev
  53 findChangedPages repos oldRev newRev
  54     = mapM (findChangedPagesAtRevision repos) [oldRev + 1 .. newRev]
  55       >>=
  56       return . S.unions
  57
  58
  59 getCurrentRevNum :: Repository -> IO RevNum
  60 getCurrentRevNum repos
  61     = getRepositoryFS repos >>= getYoungestRev
  62
  63
  64 startIndexManager :: FilePath -> Repository -> (Page -> IO Document) -> IO (TChan IndexReq)
  65 startIndexManager lsdir repos mkDraft
  66     = do chan  <- newTChanIO
  67          index <- openIndex indexDir revFile
  68          forkIO (loop chan index)
  69          return chan
  70     where
  71       indexDir = lsdir </> "index"
  72       revFile  = lsdir </> "indexRev"
  73
  74       loop :: TChan IndexReq -> Database -> IO ()
  75       loop chan index
  76           = do req <- atomically $ readTChan chan
  77                case req of
  78                  RebuildIndex
  79                      -> do noticeM logger "Rebuilding the H.E. index..."
  80                            closeDatabase index
  81                            removeDirectoryRecursive indexDir
  82                            index' <- openIndex indexDir revFile
  83                            syncIndex' index' revFile repos mkDraft
  84                            loop chan index'
  85
  86                  SyncIndex
  87                      -> do syncIndex' index revFile repos mkDraft
  88                            loop chan index
  89
  90                  SearchIndex cond var
  91                      -> do result <- searchIndex index cond
  92                            atomically $ putTMVar var result
  93                            loop chan index
  94
  95
  96 -- casket を R/W モードで開く。成功したらそのまま返し、失敗したら
  97 -- indexDir と revFile を削除してから casket を R/W モードで開く。
  98 openIndex :: FilePath -> FilePath -> IO Database
  99 openIndex indexDir revFile
 100     = do ret <- openDatabase indexDir (Writer [])
 101          case ret of
 102            Right index
 103                -> do debugM logger ("Opened an H.E. index on " ++ indexDir)
 104                      return index
 105
 106            Left err
 107                -> do noticeM logger ("Failed to open an H.E. index on "
 108                                      ++ indexDir ++ ": " ++ show err)
 109
 110                      indexExists <- doesDirectoryExist indexDir
 111                      when indexExists
 112                               $ removeDirectoryRecursive indexDir
 113
 114                      revFileExists <- doesFileExist revFile
 115                      when revFileExists
 116                               $ removeFile revFile
 117
 118                      Right index <- openDatabase indexDir (Writer [Create []])
 119                      addAttrIndex index "@uri"           SeqIndex
 120                      addAttrIndex index "rakka:revision" SeqIndex
 121                      noticeM logger ("Created an H.E. index on " ++ indexDir)
 122
 123                      return index
 124
 125
 126 syncIndex' :: Database -> FilePath -> Repository -> (Page -> IO Document) -> IO ()
 127 syncIndex' index revFile repos mkDraft
 128     = updateIndexRev revFile $ \ oldRev ->
 129       do debugM logger ("The index revision is currently " ++ show oldRev)
 130
 131          newRev <- getCurrentRevNum repos
 132          debugM logger ("The repository revision is currently " ++ show newRev)
 133
 134          when (oldRev == 0 || newRev /= oldRev)
 135               $ syncIndex'' oldRev newRev
 136          return newRev
 137     where
 138       syncIndex'' :: RevNum -> RevNum -> IO ()
 139       syncIndex'' oldRev newRev
 140           = do pages <- findChangedPages repos oldRev newRev
 141                mapM_ (updateIndex index repos mkDraft newRev) (S.toList pages)
 142
 143
 144 searchIndex :: Database -> Condition -> IO [(PageName, RevNum)]
 145 searchIndex index cond
 146     = searchDatabase index cond >>= mapM fromId
 147     where
 148       fromId :: DocumentID -> IO (PageName, RevNum)
 149       fromId docId
 150           = do uri <- getDocURI index docId
 151                rev <- getDocAttr index docId "rakka:revision"
 152                       >>= return . read . fromJust
 153                return (decodePageName $ uriPath uri, rev)
 154
 155
 156 updateIndex :: Database
 157             -> Repository
 158             -> (Page -> IO Document)
 159             -> RevNum
 160             -> PageName
 161             -> IO ()
 162 updateIndex index repos mkDraft rev name
 163     = do pageM <- getPage' repos name (Just rev)
 164          case pageM of
 165            -- ページが削除された
 166            Nothing
 167                -> do docIdM <- getDocIdByURI index (mkRakkaURI name)
 168                      case docIdM of
 169                        Nothing    -> return ()
 170                        Just docId -> do removeDocument index docId [CleaningRemove]
 171                                         infoM logger ("Removed page " ++ name ++ " from the index")
 172            Just page
 173                -> do draft <- mkDraft page
 174                      putDocument index draft [CleaningPut]
 175                      infoM logger ("Indexed page " ++ name ++ " of revision " ++ show (pageRevision page))
 176
 177
 178 updateIndexRev :: FilePath -> (RevNum -> IO RevNum) -> IO ()
 179 updateIndexRev revFile f = bracket acquireLock releaseLock update
 180     where
 181       acquireLock :: IO Fd
 182       acquireLock
 183           = do fd <- openFd revFile ReadWrite (Just stdFileMode) defaultFileFlags
 184                waitToSetLock fd (WriteLock, AbsoluteSeek, 0, 0)
 185                return fd
 186
 187       releaseLock :: Fd -> IO ()
 188       releaseLock fd
 189           = setLock fd (Unlock, AbsoluteSeek, 0, 0)
 190
 191       update :: Fd -> IO ()
 192       update fd
 193           = do fdSeek fd AbsoluteSeek 0
 194                size <- return . fromIntegral . fileSize =<< getFdStatus fd
 195                (revStr, gotSize) <- fdRead fd size
 196                when (size /= gotSize) $ fail ("read " ++ show gotSize ++
 197                                               " bytes but expected " ++ show size ++ " bytes")
 198
 199                let rev = case revStr of
 200                            "" -> 0
 201                            _  -> read revStr
 202
 203                rev' <- f rev
 204
 205                let revStr' = show rev' ++ "\n"
 206                    size'   = fromIntegral $ length revStr'
 207
 208                fdSeek fd AbsoluteSeek 0
 209                setFdSize fd 0
 210                wroteSize <- fdWrite fd revStr'
 211                when (size' /= wroteSize) $ fail ("wrote " ++ show wroteSize ++
 212                                                  " bytes but expected " ++ show size' ++ " bytes")