Commit 44b76f7d authored by Yuanle Song's avatar Yuanle Song
Browse files

GET /rd/file api is fully working 

It feels great when it works on first try.
parent b6fb1542

api/Main.hs

+10 −7
Original line number Diff line number Diff line
@@ -4,6 +4,7 @@ import Network.Socket.Internal (PortNumber) 
import Data.String (fromString)

import System.Environment (lookupEnv)

import Data.Monoid ((<>))

import Control.Concurrent.Chan

import qualified Data.Text as T

import qualified Data.Text.Lazy as LT
@@ -14,8 +15,9 @@ import Log 
import Log.Backend.StandardOutput

import qualified Database.Redis as R



import App (mkApp)

import Config

import App (mkApp)

import Worker (startWorkers)



-- TODO use a proper config lib.

-- TODO support other env variables.
@@ -26,8 +28,7 @@ updateRDConfigFromEnv config = do 
          case webroot of

            Just dir -> config {webRoot=dir}

            Nothing -> config

  withSimpleStdOutLogger $ \logger -> do

    runLogT "Main" logger $ do

  withSimpleStdOutLogger $ \logger -> runLogT "Main" logger $

    logInfo_ $ "webRoot is " <> T.pack (webRoot config)

  return newConfig
@@ -36,12 +37,14 @@ main = withSimpleStdOutLogger $ \logger -> do 
  config <- updateRDConfigFromEnv defaultRDConfig

  conn <- R.checkedConnect $ R.defaultConnectInfo {

            R.connectHost=redisHost config

          , R.connectPort=R.PortNumber $ (fromIntegral (redisPort config) :: PortNumber)

          , R.connectPort=R.PortNumber (fromIntegral (redisPort config) :: PortNumber)

          }

  fileChan <- newChan

  let runtimeConfig = RDRuntimeConfig { config=config

                                      , redisConn=conn

                                      }

  runLogT "Main" logger $ do

                                      , fileChan=fileChan}

  startWorkers runtimeConfig

  runLogT "Main" logger $

    logInfo_ $ sformat ("will listen on " % string % ":" % int) (host config) (port config)

  let opts = Options { verbose=0

                     , settings=warpSettings

operational

+68 −56
Original line number Diff line number Diff line
@@ -3,6 +3,71 @@ 
Time-stamp: <2018-05-06>

#+STARTUP: content

* notes                                                               :entry:

** 2018-05-05 write the main logic of creating block metadata.

then make it work with a thread pool with a single thread.



env WEB_ROOT=/home/sylecn/persist/cache stack exec rd-api



curl -XGET http://localhost:8082/rd/ideaIC-2018.1.tar.gz



this should return json of the block metadata.



- block metadata looks like this:

  GET /rd/bigfile

  #+BEGIN_SRC sh

    {"ok": true,

     "block_size": "2MiB",    # this is a fixed value.

     "file_size": xxxx,       # file size in bytes

     "block_count": 24,

     "blocks": [

             [0, 0, 2097151, block1_sha1sum],

             [1, 2097152, 4194303, block2_sha1sum],

             ...

             [N, start, end, blockN_sha1sum]

     ]}

  #+END_SRC



** 2018-05-06 calculate sha1sum for blocks using a thread pool. design try 2.

- data protocol via redis.

  hset <filepath>_blockSize blockId sha1sum



  set <filepath>_blockSize_status working|done



- worker pool is there for calculating all blocks for one file.



         fileQueue

  Main ------------> fileWorker



  GET /rd/file

  if file status is None, push file to fileQueue.

  do normal logic.



  fileWorker:

  fetch file from fileQueue.

  start working on blocks one by one.

  if block already cached in redis, skip it.

  when all done, set <filepath>_blockSize_status done.



- this works and is easy to understand.

  WIP info is also kept in redis for each file's block.



- works on first try. excellent.



- problems

  - how to fail when redis hget or hset fail?

    just return False

    if some block fail, set status to error.

    next time a GET /rd/file, it will trigger the queue again.



    a cron job can also trigger a run.

  - mapM how to skip rest when some action failed?



    If there is a redis error halfway during calculation, I don't want to

    calculate the rest sha1sum, because the result can't be stored.

  - 



** 2018-05-06 how to run hlint

stack exec hlint -- src api

** 2018-05-05 it's impossible to do logging easily in haskell.

two problems
@@ -230,63 +295,10 @@ that way you don't need to tell rd-server the web root dir. 


* current                                                             :entry:

** 

** 2018-05-06 calculate sha1sum for blocks using a thread pool. design try 2.

- data protocol via redis.

  hset <filepath>_blockSize blockId sha1sum



  set <filepath>_blockSize_status working|done



- worker pool is there for calculating all blocks for one file.



         fileQueue

  Main ------------> fileWorker



  GET /rd/file

  if file status is None, push file to fileQueue.

  do normal logic.



  fileWorker:

  fetch file from fileQueue.

  start working on blocks one by one.

  if block already cached in redis, skip it.

  when all done, set <filepath>_blockSize_status done.



- this works and is easy to understand.

  WIP info is also kept in redis for each file's block.



- 



** 2018-05-05 write the main logic of creating block metadata.

then make it work with a thread pool with a single thread.



env WEB_ROOT=/home/sylecn/persist/cache stack exec rd-api



curl -XGET http://localhost:8082/rd/ideaIC-2018.1.tar.gz



this should return json of the block metadata.



- block metadata looks like this:

  GET /rd/bigfile

  #+BEGIN_SRC sh

    {"ok": true,

     "block_size": "2MiB",    # this is a fixed value.

     "file_size": xxxx,       # file size in bytes

     "block_count": 24,

     "blocks": [

             [0, 0, 2097151, block1_sha1sum],

             [1, 2097152, 4194303, block2_sha1sum],

             ...

             [N, start, end, blockN_sha1sum]

     ]}

  #+END_SRC



- problems

  - it hangs.

    curl -XGET http://localhost:8082/rd/ideaIC-2018.1.tar.gz

    curl -XGET http://localhost:8082/test/rd/ideaIC-2018.1.tar.gz

    fixed.



* done                                                                :entry:

** 2018-05-06 integrate hlint

stack install hlint



** 2018-05-04 check whether haskell is viable for this project.

In static file hosting nginx reverse proxy /rd/ to rd application server.

src/App.hs

+75 −52
Original line number Diff line number Diff line 
module App (mkApp, mkWaiApp, genBlocks) where



import Control.Monad.IO.Class (liftIO)

import Control.Monad (when)

import Data.Either (fromRight)

import Data.Monoid ((<>))

import Data.Text.Encoding (decodeUtf8)

import Control.Concurrent.Chan

import qualified Data.ByteString as B

import qualified Data.ByteString.Char8 as Char8

import qualified Data.Text as T
@@ -17,13 +19,10 @@ import System.Posix.Files (getFileStatus, fileSize) 
import qualified Database.Redis as R

import qualified Data.HashMap.Strict as M



import Type

import Config

import RD.Lib (sha1sum)



type BlockID = Integer

type Block = (BlockID, Integer, Integer)

type BlockWithChecksum = (BlockID, Integer, Integer, T.Text)



genBlocks :: Integer -> Integer -> [Block]

genBlocks fileSize blockSize = if fileSize == 0 then

                                   []
@@ -31,31 +30,15 @@ genBlocks fileSize blockSize = if fileSize == 0 then 
                                   go (0 :: Integer) (0 :: Integer) []

  where

    go :: Integer -> Integer -> [Block] -> [Block]

    go blockId startByte accumulator =

        if fileSize - startByte == blockSize then

    go blockId startByte accumulator

      | fileSize - startByte == blockSize =

        reverse ((blockId, startByte, fileSize - 1) : accumulator)

        else if fileSize - startByte > blockSize then

            go (blockId + 1)

               (startByte + blockSize)

      | fileSize - startByte > blockSize =

        go (blockId + 1) (startByte + blockSize)

          ((blockId, startByte, startByte + blockSize - 1) : accumulator)

        else if startByte < fileSize then

      | startByte < fileSize =

        reverse ((blockId, startByte, fileSize - 1) : accumulator)

        else

            reverse accumulator



data FillBlockParam = FillBlockParam {

      fbpFilepath :: FilePath

    , fbpBlockSize :: Integer

    , fbpFileSize :: Integer

    , fbpBlocks :: [Block]}



-- | the redis hash key used to store cached sha1sum for given FillBlockParam

blockSha1sumHashKey :: FillBlockParam -> B.ByteString

blockSha1sumHashKey fbp = Char8.pack (fbpFilepath fbp) <> "_" <> (Char8.pack . show) (fbpBlockSize fbp)



-- | the redis hash key sub key, used to store the sha1sum for that blockId.

blockIdKey :: BlockID -> B.ByteString

blockIdKey = Char8.pack . show

      | otherwise = reverse accumulator



-- | fill block sha1sum, if sha1sum is not ready yet, put "pending" there.

fillSha1sum :: RDRuntimeConfig -> FillBlockParam -> IO [BlockWithChecksum]
@@ -79,9 +62,9 @@ fillSha1sum runtimeConfig fbp = do 
-- | given a redis connection pool, return a Scotty app.

mkApp :: RDRuntimeConfig -> ScottyM ()

mkApp runtimeConfig = do

  get (literal "/rd/") $ do

    json $ object [("ok" .= True)

                  ,("app" .= ("reliable-download api" :: T.Text))]

  get (literal "/rd/") $ json $

      object ["ok" .= True

             ,"app" .= ("reliable-download api" :: T.Text)]

  get (regex "^/rd/(.*)") $ do

    path :: LT.Text <- param "1"

    let filepath = combine (webRoot (config runtimeConfig)) (LT.unpack path)
@@ -91,34 +74,74 @@ mkApp runtimeConfig = do 
        blockSizeInByte = 2097152    -- 2MiB

        blockCount = (fileSizeInByte - 1) `div` blockSizeInByte + 1

        blocks = genBlocks fileSizeInByte blockSizeInByte

    blocksWithSha1sum <- liftIO $ fillSha1sum runtimeConfig $ FillBlockParam {

                              fbpFilepath=filepath

        fbp = FillBlockParam { fbpFilepath=filepath

                             , fbpFileSize=fileSizeInByte

                             , fbpBlockSize=blockSizeInByte

                             , fbpBlocks=blocks }

    json $ object [("ok" .= True)

                  ,("block_size" .= ("2MiB" :: T.Text))

                  ,("file_size" .= fileSizeInByte)

                  ,("block_count" .= blockCount)

                  ,("blocks" .= blocksWithSha1sum)

                  ,("path" .= path)

                  ,("filepath" .= filepath)

                  ]

        strKey = fileStatusKey fbp

        jsonRespFileStatusCheckFailed = json $

          object ["ok" .= False

                 ,"path" .= path

                 ,"filepath" .= filepath

                 ,"msg" .= ("check file status in redis failed" :: T.Text)]

        jsonRespOk = do

          blocksWithSha1sum <- liftIO $ fillSha1sum runtimeConfig fbp

          json $ object ["ok" .= True

                        ,"block_size" .= ("2MiB" :: T.Text)

                        ,"file_size" .= fileSizeInByte

                        ,"block_count" .= blockCount

                        ,"blocks" .= blocksWithSha1sum

                        ,"path" .= path

                        ,"filepath" .= filepath]



    -- if this file is new or has status "error", add task to fileChan.

    redisReply <- liftIO $ R.runRedis (redisConn runtimeConfig) $ R.setnx strKey "working"

    statusCheckResult <- case redisReply of

      Left reply -> do

        liftIO $ putStrLn $ "redis setnx " <> show strKey <> " failed: " <> show reply

        return False

      Right setNxResult ->

        if setNxResult then do

            liftIO $ writeChan (fileChan runtimeConfig) fbp

            return True

        else do

            liftIO $ putStrLn $ "redis key " <> show strKey <> " exists. file is old"

            -- if status is error, set it to working, then add task to fileChan

            redisReply2 <- liftIO $ R.runRedis (redisConn runtimeConfig) $ R.get strKey

            case redisReply2 of

              Left reply -> do

                liftIO $ putStrLn $ "redis file status check for " <> show strKey <> " failed: " <> show reply

                return False

              Right statusStr ->

                liftIO $ if statusStr == Just "error"

                  then do

                    putStrLn $ "set file status to \"working\" for " <> show strKey

                    redisReply3 <- liftIO $ R.runRedis (redisConn runtimeConfig) $ R.set strKey "working"

                    case redisReply3 of

                      Left reply -> do

                        liftIO $ putStrLn $ "set file status to \"working\" failed: " <> show reply

                        return False

                      Right _ -> do

                        liftIO $ writeChan (fileChan runtimeConfig) fbp

                        return True

                  else return True

    if statusCheckResult then

        jsonRespOk

    else

        jsonRespFileStatusCheckFailed



  get (regex "^/test/rd/(.*)") $ do  -- for testing path capture

    path :: LT.Text <- param "1"

    let filepath = combine (webRoot (config runtimeConfig)) (LT.unpack path)

    json $ object [("ok" .= True)

                  ,("path" .= path)

                  ,("filepath" .= filepath)

                  ]

    json $ object ["ok" .= True

                  ,"path" .= path

                  ,"filepath" .= filepath]

  get "/debug/t1" $ do

    sha1 <- liftIO $ sha1sum "/home/sylecn/persist/cache/ideaIC-2018.1.tar.gz"

    json $ object ["ok" .= True

                  ,"sha1sum" .= sha1]

  get "/debug/count" $ do

    count <- liftIO $ R.runRedis (redisConn runtimeConfig) $ do

                        count <- R.incr "count"

                        return count

    count <- liftIO $ R.runRedis (redisConn runtimeConfig) $ R.incr "count"

    json $ object ["ok" .= True

                  ,"count" .= fromRight 0 count]

src/Config.hs

+22 −0
Original line number Diff line number Diff line 
module Config where



import qualified Database.Redis as R

import Control.Concurrent.Chan

import qualified Data.ByteString as B

import qualified Data.ByteString.Char8 as Char8

import Data.Monoid ((<>))



import Type



data RDConfig = RDConfig {

      host :: String
@@ -8,11 +14,13 @@ data RDConfig = RDConfig { 
    , redisHost :: String

    , redisPort :: Int

    , webRoot :: FilePath

    , fileWorkerCount :: Int

    } deriving (Show)



data RDRuntimeConfig = RDRuntimeConfig {

      config :: RDConfig

    , redisConn :: R.Connection

    , fileChan :: Chan FillBlockParam

    }



defaultRDConfig :: RDConfig
@@ -22,4 +30,18 @@ defaultRDConfig = RDConfig { 
                  , redisHost = "127.0.0.1"

                  , redisPort = 6379

                  , webRoot = "/nonexistent"

                  , fileWorkerCount = 2

                  }



-- | the redis hash key used to store cached sha1sum for given FillBlockParam

blockSha1sumHashKey :: FillBlockParam -> B.ByteString

blockSha1sumHashKey fbp = Char8.pack (fbpFilepath fbp) <> "_" <> (Char8.pack . show) (fbpBlockSize fbp)



-- | the redis hash key sub key, used to store the sha1sum for that blockId.

blockIdKey :: BlockID -> B.ByteString

blockIdKey = Char8.pack . show



-- | the redis string key used to track whether this file and blockSize is

-- new|working|done.

fileStatusKey :: FillBlockParam -> B.ByteString

fileStatusKey fbp = blockSha1sumHashKey fbp <> "_status"

src/RD/Lib.hs

+1 −1
Original line number Diff line number Diff line
@@ -10,7 +10,7 @@ import Crypto.Hash 


-- | get sha1sum hex string for given bytes

sha1sumOnBytes :: LB.ByteString -> LB.ByteString

sha1sumOnBytes bytes = LB.fromStrict $ digestToHexByteString $ (hashlazy bytes :: Digest SHA1)

sha1sumOnBytes bytes = LB.fromStrict $ digestToHexByteString (hashlazy bytes :: Digest SHA1)



-- | calculate sha1sum for given file

sha1sum :: LT.Text -> IO LT.Text