hadoop-rpc 1.0.0.0 → 1.0.0.1

raw patch · 6 files changed

+86/−107 lines, 6 filesdep +monad-loopsdep +tastydep +tasty-hunitPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: monad-loops, tasty, tasty-hunit

API changes (from Hackage documentation)

Files

• hadoop-rpc.cabal +15/−3
• src/Network/Hadoop/Hdfs.hs +2/−3
• src/Network/Hadoop/Read.hs +58/−96
• src/Network/Hadoop/Rpc.hs +0/−1
• src/Network/Hadoop/Stream.hs +1/−1
• test/Main.hs +10/−3

@@ -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

@@ -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 -> ()

@@ -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)

@@ -1,4 +1,3 @@-{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} 

@@ -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)     } 

@@ -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 {