bugfix and continue implementing rd client.

import Control.Concurrent.Chan

import System.Directory (setCurrentDirectory)

import qualified Data.Text as T

import qualified Data.Text.Lazy as LT

import Web.Scotty

import Network.Wai.Handler.Warp

import Formatting

import Log

            Just dir -> config {webRoot=dir}

            Nothing -> config

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

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

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

  return newConfig

main :: IO ()

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

          }

  fileChan <- newChan

  let runtimeConfig = RDRuntimeConfig { config=config

                                      , redisConn=conn

                                      , fileChan=fileChan}

  let runtimeConfig = RDRuntimeConfig { rcConfig=config

                                      , rcRedisConn=conn

                                      , rcFileChan=fileChan}

  startWorkers runtimeConfig

  runLogT "Main" logger $

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

import Control.Exception

import System.IO.Error

import System.Exit

import System.Environment (lookupEnv)

import Data.List (isPrefixOf)

import System.FilePath ((</>))

import Data.Text.Encoding (decodeUtf8)

import Control.Concurrent (threadDelay)

import qualified Data.Text as T

import qualified Data.ByteString as B

import qualified Data.ByteString.Lazy as LB

import qualified Data.ByteString.Lazy.Char8 as L8

import qualified Data.ByteString.Char8 as Char8

import qualified Data.HashMap.Strict as M

import Data.Aeson (Value(..))

import Network.HTTP.Simple

import Network.HTTP.Client (path)

import Formatting

import Formatting hiding (bytes)

import RD.Lib (sha1sumOnBytes, guessFilename)

import Type

      filename = fbpFilename fbp

      rangeHeader = "bytes=" <> Char8.pack (show start) <> "-"

                             <> Char8.pack (show end)

  assert (sha1sum /= "pending") (return ())

  -- TODO capture http error. implement retry here.

  debug opts $ "downloading " <> filename <> " block " <> show blockId

  req <- parseRequest $ T.unpack $ fbpUrl fbp

      LB.writeFile blockTargetFile bodyLBS

      return True

  else do

      putStrLn $ "sha1sum verification failed for " <> filename <> " block " <> show blockId

      putStrLn $ "sha1sum verification failed for " <> filename <> " block " <> show blockId <> ", expect " <> show sha1sum

      return False

-- | return block target file name (just base filename, no dir info)

  formatToString ("block" % left padding '0' % "_" % stext) blockId sha1sum

-- | fetch a single block, return IO True on success

fetchBlock :: RDOptions -> T.Text -> RDResponse -> BlockID -> IO Bool

fetchBlock opts url rdResp blockId = do

  let blockWithChecksum = respBlocks rdResp !! fromIntegral blockId

      filename = guessFilename url

fetchBlock :: RDOptions -> T.Text -> RDResponse -> BlockWithChecksum -> IO Bool

fetchBlock opts url rdResp blockWithChecksum = do

  let filename = guessFilename url

      blockFileDir = tempDir opts </> filename

      blockFilename = getBlockFilename rdResp blockWithChecksum

      blockTargetFile = blockFileDir </> blockFilename

    removeDirectoryRecursive tempdir

  return True

-- | call /rd/<file> api and fetch response

getRDResponse :: RDOptions -> T.Text -> IO RDResponse

getRDResponse _opts url = do

  req <- parseRequest $ T.unpack url

  resp <- httpJSON $ req { path="/rd" <> path req }

  return $ getResponseBody resp

-- | download file at given URL using reliable download API and block based

-- downloading.

downloadFile :: RDOptions -> T.Text -> IO Bool

downloadFile opts url = do

  req <- parseRequest $ T.unpack url

  resp <- httpJSON $ req { path="/rd" <> path req }

  let rdResp = getResponseBody resp :: RDResponse

  rdResp <- getRDResponse opts url

  if not $ respOk rdResp then do

      putStrLn $ "GET /rd/ api failed: " <> show (respMsg rdResp)

      return False

      putStrLn $ "Downloading file: " <> show (respPath rdResp) <> ", "

               <> humanReadableSize (respFileSize rdResp)

               <> ", " <> show (respBlockCount rdResp) <> " blocks"

      results <- mapM (fetchBlock opts url rdResp) [0..respBlockCount rdResp - 1]

      (rdResp2, results) <- loopUntilAllBlocksReady opts url rdResp []

      if and results then

          combineBlocks opts rdResp

          combineBlocks opts rdResp2

      else do

          putStrLn $ (show . length . filter id) results <> " blocks failed."

          return False

-- | a sleep loop that check whether all blocks in rdResp is ready, if not, do

-- a GET again later and check it again. On each try, the diff of new ready

-- blocks are sent to fetchBlock function.

--

-- Return whether download is successful for each block.

loopUntilAllBlocksReady :: RDOptions -> T.Text -> RDResponse -> [BlockID] -> IO (RDResponse, [Bool])

loopUntilAllBlocksReady opts url rdResp oldReadyBlocks = do

  let blockIsReady = (/= "pending") . getBlockSha1sum

      blocks = respBlocks rdResp

      readyBlocks = filter blockIsReady blocks

      newReadyBlocks = filter ((`notElem` oldReadyBlocks) . getBlockId) readyBlocks

      allBlocksReady = all blockIsReady blocks

  putStrLn $ (show . length) newReadyBlocks <> " new blocks ready on server side"

  results <- mapM (fetchBlock opts url rdResp) newReadyBlocks

  if allBlocksReady then

    -- loop finished

    return (rdResp, results)

  else do

    when (null newReadyBlocks) $ do

         putStrLn "No new blocks ready on server side, waiting 1s"

         threadDelay 1000000

    newRdResp <- getRDResponse opts url

    let prevResp = if respOk newRdResp then newRdResp else rdResp

    loopUntilAllBlocksReady opts url prevResp (map getBlockId readyBlocks)

cliApp :: RDOptions -> IO ()

cliApp opts = do

  debug opts $ "command line options: " <> show opts

  results <- mapM (downloadFile opts) (urls opts)

  if and results then

      putStrLn "all urls downloaded."

  else do

  else

      die $ (show . length . filter id) results <> " urls failed."

main :: IO ()

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

  stack exec rd -- -d ~/d/.blocks -o ~/d/ http://localhost:8082/ideaIC-2018.1.tar.gz

  test fresh block-not-ready state:

  redis-cli del "/home/sylecn/persist/cache/sdkman.sh_2097152_status"

  redis-cli del "/home/sylecn/persist/cache/sdkman.sh_2097152"

  stack exec rd -- -d ~/d/.blocks -o ~/d/ http://localhost:8082/sdkman.sh

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

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

* current                                                             :entry:

** 

** 2018-05-07 add reliability to client code.

- when a block is pending. keep a sleep loop to do GET /rd/file again.

- DONE If some block sha1sum is pending, go-to a sleep loop, retry until there

  are no pending blocks. Filter pending blocks when sending download task to

  thread pool.

  should I use Control.Monad.Trans.Loop or just recursive call?

  I used recursive call.

- use a thread pool to download blocks. print overall progress when some

  parts done or some time elapsed.

- retry download if sha1sum verification failed.

- catch all IO exceptions, including disk io, redis io, http io.

  fail at correct checkpoints.

- 

- problems

  - does haskell support TCO?

    https://softwareengineering.stackexchange.com/questions/144274/whats-the-difference-between-recursion-and-corecursion

    And Haskell's guarded recursion is just like tail recursion modulo cons.

    https://stackoverflow.com/a/8882745/849891

    http://en.wikipedia.org/wiki/Tail_call#Tail_recursion_modulo_cons

    too complex, read it later.

  - fix exception, fresh run fails. and the output is very frightening.

    when some block fail, it should never run combine.

    #+BEGIN_SRC sh

      sylecn@ryzen5:~/projects/reliable-download$ stack exec rd -- -d ~/d/.blocks -o ~/d/ http://localhost:8082/sdkman.sh

      GET /rd/ api ok

      Downloading file: "sdkman.sh", 0 MiB, 1 blocks

      sha1sum verification failed for sdkman.sh block 0, expect "pending"

      combining blocks to create /home/sylecn/d/sdkman.sh

      rd: /home/sylecn/d/.blocks/sdkman.sh/block0_pending: openBinaryFile: does not exist (No such file or directory)

      sylecn@ryzen5:~/projects/reliable-download$

    #+END_SRC

    server side may fail to set status to "done".

    // fixed. it's missing forever $ do on worker.

    what does mapM return on empty list?

    mapM (\x -> return True) []

    it's an empty list.

  - DONE still have race condition

    #+BEGIN_SRC sh

      sylecn@ryzen5:~/projects/reliable-download$   stack exec rd -- -d ~/d/.blocks -o ~/d/ http://localhost:8082/ideaIC-2018.1.tar.gz

      GET /rd/ api ok

      Downloading file: "ideaIC-2018.1.tar.gz", 516 MiB, 259 blocks

      0 new blocks ready on server side

      No new blocks ready on server side, waiting 1s

      259 new blocks ready on server side

      combining blocks to create /home/sylecn/d/ideaIC-2018.1.tar.gz

      rd: /home/sylecn/d/.blocks/ideaIC-2018.1.tar.gz/block000_pending: openBinaryFile: does not exist (No such file or directory)

      sylecn@ryzen5:~/projects/reliable-download$

    #+END_SRC

    I see. combining blocks get outdated rdRes.

    fixed.

  - infinite loop on server side.

    when I deleted two redis key. and do a GET /rd/file.

    no. loop is in client side. server side has no recursive call.

    fixed.

  - DONE when client side keep requesting /rd/file, the server side worker doesn't

    calculate block sha1sum! staying in "pending".

    status is "working"

    hash key is empty hash.

    I believe it's a race condition.

    // I see the problem. the Worker only read one job from queue, then it

    quits! There is no loop. haha. That's why task stays in "pending" state.

  - DONE webRoot is /nonexistent?

    fixed.

** 2018-05-06 write client side tool to actually do the download.

- make this work:

extra-source-files:

- README.md

ghc-options:

  - -Wall

  - -Werror

  # - -fprof-auto

default-extensions:

  - OverloadedStrings

  - ScopedTypeVariables

module App (mkApp, mkWaiApp, genBlocks) where

module App (mkApp, mkWaiApp) 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

import qualified Data.Text.Lazy as LT

import Network.Wai (Application)

import Web.Scotty

import Data.Aeson (Value(..), toJSON, object, (.=))

import System.FilePath (combine)

import Data.Aeson (object, (.=))

import System.FilePath ((</>))

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)

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

genBlocks fileSize blockSize = if fileSize == 0 then

                                   []

                               else

                                   go (0 :: Integer) (0 :: Integer) []

  where

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

    go blockId startByte accumulator

      | fileSize - startByte == blockSize =

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

      | fileSize - startByte > blockSize =

        go (blockId + 1) (startByte + blockSize)

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

      | startByte < fileSize =

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

      | otherwise = reverse accumulator

import RD.Lib (sha1sum, genBlocks)

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

fillSha1sum :: RDRuntimeConfig -> FillBlockParam -> IO [BlockWithChecksum]

fillSha1sum runtimeConfig fbp = do

  let filepath = fbpFilepath fbp

      blockSize = fbpBlockSize fbp

      hashKey = blockSha1sumHashKey fbp

  redisReply <- R.runRedis (redisConn runtimeConfig) $ R.hgetall hashKey

  let hashKey = blockSha1sumHashKey fbp

  redisReply <- R.runRedis (rcRedisConn runtimeConfig) $ R.hgetall hashKey

  case redisReply of

    Left reply -> do

      putStrLn $ "redis hgetall " <> show hashKey <> " failed: " <> show reply

      return $ map fillBlock (fbpBlocks fbp) where

        fillBlock (blockId, start, end) = (blockId, start, end, "pending")

    Right blockIdSha1sumAlist -> do

      putStrLn $ "redis hgetall " <> show hashKey <> " ok"

      putStrLn $ "fillSha1sum: redis hgetall " <> show hashKey <> " ok"

      return $ map fillBlock (fbpBlocks fbp) where

        blockIdSha1sumMap = M.fromList blockIdSha1sumAlist

        fillBlock :: Block -> BlockWithChecksum

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

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

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

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

    liftIO $ putStrLn $ "get block metadata for " <> filepath

    let filepath = webRoot (rcConfig runtimeConfig) </> T.unpack path

    liftIO $ putStrLn $ "user request " <> filepath

    fileStatus <- liftIO $ getFileStatus filepath    -- TODO catch IO exception

    let fileSizeInByte = toInteger $ fileSize fileStatus

        blockSizeInByte = 2097152    -- 2MiB

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

        jsonRespOk = do

          blocksWithSha1sum <- liftIO $ fillSha1sum runtimeConfig fbp

          json $ RDResponse {

                        respOk=True

          json RDResponse { respOk=True

                          , respMsg=""

                          , respPath=path

                          , respFilePath=filepath

                          , respBlocks=blocksWithSha1sum }

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

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

    redisReply <- liftIO $ R.runRedis (rcRedisConn 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

            liftIO $ putStrLn $ "new file, status set to working for " <> show strKey

            liftIO $ writeChan (rcFileChan runtimeConfig) fbp

            return True

        else do

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

            liftIO $ putStrLn $ "not a new file, redis key " <> show strKey <> " exists"

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

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

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

            case redisReply2 of

              Left reply -> do

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

                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"

                    redisReply3 <- liftIO $ R.runRedis (rcRedisConn 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

                        liftIO $ writeChan (rcFileChan runtimeConfig) fbp

                        return True

                  else return True

    if statusCheckResult then

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

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

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

    let filepath = webRoot (rcConfig runtimeConfig) </> LT.unpack path

    json $ object ["ok" .= True

                  ,"path" .= path

                  ,"filepath" .= filepath]

    json $ object ["ok" .= True

                  ,"sha1sum" .= sha1]

  get "/debug/count" $ do

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

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

    json $ object ["ok" .= True

                  ,"count" .= fromRight 0 count]