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