diff --git a/hadoop-rpc.cabal b/hadoop-rpc.cabal --- a/hadoop-rpc.cabal +++ b/hadoop-rpc.cabal @@ -1,5 +1,5 @@ name: hadoop-rpc -version: 1.0.0.0 +version: 1.0.0.1 synopsis: Use the Hadoop RPC interface from Haskell. @@ -26,7 +26,7 @@ hs-source-dirs: src default-language: Haskell2010 - ghc-options: -funbox-strict-fields + ghc-options: -funbox-strict-fields -Wall exposed-modules: Data.Hadoop.Configuration @@ -51,6 +51,7 @@ , cereal >= 0.4 , exceptions >= 0.6 , hashable >= 1.2.1 + , monad-loops >= 0.4 , network >= 2.5 , protobuf >= 0.2.0.4 , random >= 1.0.1.1 @@ -64,13 +65,24 @@ , vector >= 0.10 , xmlhtml >= 0.2 -executable test-hadoop-rpc +flag travis + description: Travis CI can't run some tests that depend on HDFS being present, + so this flag disables those. + default: False + +test-suite hdfs-tests + type: exitcode-stdio-1.0 main-is: Main.hs hs-source-dirs: test default-language: Haskell2010 + if flag(travis) + buildable: False + build-depends: base >= 4.7 && < 5 , hadoop-rpc , protobuf + , tasty + , tasty-hunit >= 0.9 , vector diff --git a/src/Network/Hadoop/Hdfs.hs b/src/Network/Hadoop/Hdfs.hs --- a/src/Network/Hadoop/Hdfs.hs +++ b/src/Network/Hadoop/Hdfs.hs @@ -132,8 +132,7 @@ -> TBQueue (Maybe (HdfsPath, Either SomeException (V.Vector FileStatus))) -> HdfsPath -> IO () -getListingRecursive' conn outstanding queue rootPath = do - getInitial rootPath +getListingRecursive' conn outstanding queue = getInitial where getInitial path = getPartial path B.empty @@ -264,7 +263,7 @@ setPermissions mode path = ignore <$> hdfsInvoke "setPermission" P.SetPermissionRequest { P.chmodPath = putField (T.decodeUtf8 path) - , P.chmodMode = putField $ P.FilePermission { fpPerm = putField mode } + , P.chmodMode = putField P.FilePermission{ fpPerm = putField mode } } where ignore :: P.SetPermissionResponse -> () diff --git a/src/Network/Hadoop/Read.hs b/src/Network/Hadoop/Read.hs --- a/src/Network/Hadoop/Read.hs +++ b/src/Network/Hadoop/Read.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -14,43 +13,41 @@ , hdfsCat ) where -import Control.Applicative (Applicative(..), (<$>)) -import Control.Exception (SomeException, throwIO) -import Control.Monad (guard, foldM) +import Control.Applicative ((<$>), (<$)) +import Control.Exception (IOException, throwIO) +import Control.Monad (foldM) import Control.Monad.Catch (MonadMask, catch) import Control.Monad.IO.Class (MonadIO(..)) -import Data.Attoparsec.Text (Parser, char, decimal, parseOnly) -import Data.Bits +import Control.Monad.Loops (iterateUntilM) +import Control.Monad.Trans.State +import Control.Monad.Trans.Class (lift) import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B -import Data.List (intercalate) import Data.Maybe (fromMaybe) import Data.ProtocolBuffers import Data.ProtocolBuffers.Orphans () import qualified Data.Text as T import qualified Data.Text.Encoding as T -import Data.Word (Word32, Word64) +import Data.Word (Word64) -import Data.Serialize.Get -import Data.Serialize.Put +import qualified Data.Serialize.Get as Get +import qualified Data.Serialize.Put as Put import Data.Hadoop.Protobuf.ClientNameNode import Data.Hadoop.Protobuf.DataTransfer import Data.Hadoop.Protobuf.Hdfs -import Data.Hadoop.Protobuf.Security import Data.Hadoop.Types import Network.Hadoop.Hdfs import Network.Hadoop.Rpc import qualified Network.Hadoop.Socket as S -import Network.Hadoop.Stream (Stream) import qualified Network.Hadoop.Stream as Stream +import Text.Printf (printf) +import Prelude hiding (rem) + ------------------------------------------------------------------------ -data HdfsReadHandle = HdfsReadHandle - { readProxy :: Maybe SocksProxy - , readLocations :: LocatedBlocks - } +data HdfsReadHandle = HdfsReadHandle (Maybe SocksProxy) LocatedBlocks -- | Open an HDFS path for reading. -- This retrieves the block locations for an HDFS path, but does not @@ -75,62 +72,59 @@ maybe (return ()) (hdfsMapM_ (liftIO . B.putStr)) h -- | Map an action over the contents of a file on HDFS. -hdfsMapM_ :: (MonadIO m, MonadMask m) => +hdfsMapM_ :: (Functor m, MonadIO m, MonadMask m) => (ByteString -> m ()) -> HdfsReadHandle -> m () hdfsMapM_ f = hdfsFoldM (\_ x -> f x) () -hdfsFoldM :: (MonadIO m, MonadMask m) => +hdfsFoldM :: (Functor m, MonadIO m, MonadMask m) => (a -> ByteString -> m a) -> a -> HdfsReadHandle -> m a hdfsFoldM f acc0 (HdfsReadHandle proxy l) = do let len = getField . lbFileLength $ l - foldM (procBlock f proxy len) acc0 (getField . lbBlocks $ l) + foldM (flip (execStateT . procBlock f proxy len)) acc0 (getField $ lbBlocks l) -procBlock :: (MonadIO m, MonadMask m) => - (a -> ByteString -> m a) -> Maybe SocksProxy -> Word64 -> a -> LocatedBlock -> m a -procBlock f proxy len acc0 b = do - let extended = getField . lbExtended $ b - token = getField . lbToken $ b - case getField . lbLocations $ b of +procBlock :: (Functor m, MonadIO m, MonadMask m) => + (a -> ByteString -> m a) -> Maybe SocksProxy -> Word64 -> LocatedBlock -> StateT a m () +procBlock f proxy blockSize block = do + let extended = getField . lbExtended $ block + token = getField . lbToken $ block + case getField . lbLocations $ block of [] -> error $ "No locations for block " ++ show extended - ls -> failover (error $ "All locations failed for block " ++ show extended) - (map (getLoc proxy len extended token) ls) + ls -> failover [] ls (getLoc extended token) $ \es -> + error $ printf "All locations failed for block %s:\n%s" + (show es) (show extended) where - failover err [] = err - failover err (x:xs) = catch x f - where f (_ :: SomeException) = failover err xs + failover es [] _ err = err es + failover es (x:xs) fn err = catch (fn x) handler + where handler (e :: IOException) = failover ((e,x):es) xs fn err - getLoc proxy len extended token l = do + getLoc extended token l = do let i = getField (dnId l) - Right addr = parseOnly parseIPv4 . getField . dnIpAddr $ i host = getField . dnHostName $ i port = getField . dnXferPort $ i - let endpoint = Endpoint host (fromIntegral port) - runBlock proxy endpoint 0 len extended token + endpoint = Endpoint host (fromIntegral port) + runBlock endpoint 0 extended token - runBlock proxy endpoint offset len0 extended token = do - let len = fromMaybe len0 . getField . ebNumBytes $ extended + runBlock endpoint offset extended token = do + let len = fromMaybe blockSize . getField . ebNumBytes $ extended S.bracketSocket proxy endpoint $ readBlock offset len extended token - readBlock offset len extended token sock = go 0 offset len acc0 + readBlock offset0 len extended token sock = () <$ go (0, offset0, len) where - go nread0 offset0 rem0 acc = do - (len, acc') <- readBlockPart offset0 rem0 extended token sock acc - let offset = offset0 + len - nread = nread0 + len - rem = rem0 - len - if rem > 0 then go nread offset rem acc' else return acc' + go = iterateUntilM (\(_, _, rem) -> rem <= 0) $ \(nread, offset, rem) -> do + len' <- readBlockPart offset rem extended token sock + return (nread + len', offset + len', rem - len') - readBlockPart offset rem extended token sock acc = do + readBlockPart offset rem extended token sock = do stream <- liftIO $ Stream.mkSocketStream sock b <- liftIO $ do Stream.runPut stream putReadRequest Stream.runGet stream decodeLengthPrefixedMessage - readPackets b rem stream acc + readPackets b rem stream where putReadRequest = do - putWord16be 28 -- Data Transfer Protocol Version, 2 bytes - putWord8 81 -- READ_BLOCK + Put.putWord16be 28 -- Data Transfer Protocol Version, 2 bytes + Put.putWord8 81 -- READ_BLOCK encodeLengthPrefixedMessage opReadBlock opReadBlock = OpReadBlock @@ -149,69 +143,37 @@ , bhToken = putField (Just token) } - showBlockOpResponse :: BlockOpResponse -> String - showBlockOpResponse = show - - readPackets BlockOpResponse{..} len stream acc1 = go 0 len acc1 + readPackets BlockOpResponse{..} len stream = fst <$> go (0, len) where - go nread0 rem0 acc = do - (len, acc') <- readPacket b rem0 stream acc - let rem = rem0 - len - nread = nread0 + len - if rem > 0 then go nread rem acc' else return (nread, acc') + go = iterateUntilM (\(_, rem) -> rem <= 0) $ \(nread, rem) -> do + len' <- readPacket (fromIntegral <$> b) stream + return (nread + len', rem - len') m = getField borReadOpChecksumInfo c = getField . rociChecksum <$> m - b = fromIntegral . getField . csBytesPerChecksum <$> c + b = getField . csBytesPerChecksum <$> c - readPacket bytesPerChecksum remaining stream acc = do + readPacket bytesPerChecksum stream = do (dataLen, d) <- liftIO $ do - len <- Stream.runGet stream getWord32be - sz <- Stream.runGet stream getWord16be - - bs <- Stream.runGet stream $ getByteString (fromIntegral sz) + _len <- Stream.runGet stream Get.getWord32be + sz <- Stream.runGet stream Get.getWord16be + bs <- Stream.runGet stream $ Get.getByteString (fromIntegral sz) ph <- decodeBytes bs let numChunks = countChunks ph - dataLen = fromIntegral . getField . phDataLen $ ph - - _ <- Stream.runGet stream (getByteString (4*numChunks)) - (fromIntegral dataLen,) <$> - Stream.runGet stream (getByteString dataLen) - - acc' <- f acc d - return (fromIntegral dataLen, acc') + dataLen = getField $ phDataLen ph + _ <- Stream.runGet stream (Get.getByteString (4*numChunks)) + (dataLen,) <$> + Stream.runGet stream (Get.getByteString (fromIntegral dataLen)) + get >>= \x -> lift (f x d) >>= put + return (fromIntegral dataLen) where - showPacketHeader :: PacketHeader -> String - showPacketHeader = show - countChunks :: PacketHeader -> Int countChunks PacketHeader{..} = (dataLen + b - 1) `div` b where - b = fromIntegral $ fromMaybe 512 bytesPerChecksum + b = fromMaybe 512 bytesPerChecksum dataLen = fromIntegral $ getField phDataLen - decodeBytes bs = case runGetState decodeMessage bs 0 of + decodeBytes bs = case Get.runGetState decodeMessage bs 0 of Left err -> throwIO (RemoteError "DecodeError" (T.pack err)) Right (x, "") -> return x Right (_, _) -> throwIO (RemoteError "DecodeError" "decoded response but did not consume enough bytes") - ------------------------------------------------------------------------- - -parseIPv4 :: Parser Word32 -parseIPv4 = d >>= dd >>= dd >>= dd - where - d = do - x <- decimal - guard $ x < 256 - return x - dd acc = do - x <- char '.' *> d - return $ (acc `shiftL` 8) .|. x - -showIPv4 :: Word32 -> String -showIPv4 x = intercalate "." . map show $ [a,b,c,d] - where - a = (x .&. 0xFF000000) `shiftR` 24 - b = (x .&. 0xFF0000) `shiftR` 16 - c = (x .&. 0xFF00) `shiftR` 8 - d = (x .&. 0xFF) diff --git a/src/Network/Hadoop/Rpc.hs b/src/Network/Hadoop/Rpc.hs --- a/src/Network/Hadoop/Rpc.hs +++ b/src/Network/Hadoop/Rpc.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} diff --git a/src/Network/Hadoop/Stream.hs b/src/Network/Hadoop/Stream.hs --- a/src/Network/Hadoop/Stream.hs +++ b/src/Network/Hadoop/Stream.hs @@ -45,7 +45,7 @@ -- | Lightweight abstraction over an input/output stream. data Stream = Stream { streamIn :: IO (Maybe B.ByteString) - , streamOut :: (Maybe L.ByteString -> IO ()) + , streamOut :: Maybe L.ByteString -> IO () , streamState :: !(IORef StreamState) } diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -4,13 +4,20 @@ import Data.Hadoop.Types import Network.Hadoop.Hdfs +import Test.Tasty (TestTree, defaultMain, testGroup) +import Test.Tasty.HUnit (testCase) ------------------------------------------------------------------------ main :: IO () -main = do - files <- runHdfs' config (getListing' "/") - print files +main = defaultMain testTree + +testTree :: TestTree +testTree = testGroup "Commands" [listing] + where + listing = testCase "Listing root" $ do + files <- runHdfs' config (getListing' "/") + print files config :: HadoopConfig config = HadoopConfig {