{-# LANGUAGE ExistentialQuantification, DeriveDataTypeable
, FlexibleInstances, MultiParamTypeClasses, OverloadedStrings, CPP #-}
module Transient.MapReduce
(
Distributable(..),distribute, getText,
getUrl, getFile,textUrl, textFile,
mapKeyB, mapKeyU, reduce,eval,
-- * internals
DDS(..),Partition(..),PartRef(..))
where
#ifdef ghcjs_HOST_OS
import Transient.Base
import Transient.Move hiding (pack)
import Transient.Logged hiding (hash)
-- dummy Transient.MapReduce module,
reduce _ _ = local stop :: Loggable a => Cloud a
mapKeyB _ _= undefined
mapKeyU _ _= undefined
distribute _ = undefined
getText _ _ = undefined
textFile _ = undefined
getUrl _ _ = undefined
textUrl _ = undefined
getFile _ _ = undefined
eval _= local stop
data Partition
data DDS= DDS
class Distributable
data PartRef a=PartRef a
#else
import Transient.Internals hiding (Ref)
import Transient.Mailboxes
import Transient.Parse
import Transient.Logged
import Transient.Move.Internals hiding (pack)
import Transient.Indeterminism
import Control.Applicative
import System.Random
import Control.Monad.State
import Control.Monad
import Data.Monoid
import Data.Typeable
import Data.Foldable
import Data.List hiding (delete, foldl')
import Control.Exception
import Control.Concurrent
--import Data.Time.Clock
import Network.HTTP
import Data.TCache hiding (onNothing)
import Data.TCache.Defs hiding (serialize,deserialize)
import Data.ByteString.Lazy.Char8 (pack,unpack, toStrict)
import Data.ByteString.Builder
import qualified Data.Map.Strict as M
import Control.Arrow (second)
import qualified Data.Vector.Unboxed as DVU
import qualified Data.Vector as DV
import Data.Hashable
import System.IO.Unsafe
import qualified Data.Foldable as F
import qualified Data.Text as Text
import Data.Text.Encoding
import Data.IORef
-- | a DDS contains a distrib. computation which return a (non-deterministic/stream/set of)
-- link/s to the generated chunk/s of data, in different nodes, thanks to the non-deterministic
-- and multithreaded nature of the Transient/Cloud comp.
data DDS a= Loggable a => DDS (Cloud (PartRef a))
-- | a link to a chunk of data, located in a node
data PartRef a= Ref Node Path Save deriving (Typeable, Read, Show)
-- | the chunk of data loaded in memory
data Partition a= Part Node Path Save a deriving (Typeable,Read,Show)
type Save= Bool
instance Loggable Text.Text where
serialize t= byteString (encodeUtf8 t)
deserialize = tTakeWhile (/= '/') >>= return . decodeUtf8 . toStrict
instance Typeable a => Loggable (PartRef a) where
serialize(Ref node path save)= serialize node <> "/" <> serialize path <> "/" <> serialize save -- <> "/"
deserialize= Ref <$> (deserialize <* slash)
<*> (deserialize <* slash)
<*> (deserialize)
where
slash= tChar '/'
instance Indexable (Partition a) where
key (Part _ string b _)= keyp string b
keyp s True= "PartP@"++s :: String
keyp s False="PartT@"++s
instance Loggable a => IResource (Partition a) where
keyResource= key
readResourceByKey k= r
where
typePart :: IO (Maybe a) -> a
typePart = undefined
r = if k !! 4 /= 'P' then return Nothing else
defaultReadByKey (defPath (typePart r) ++ k) >>= return . fmap ( read . unpack)
writeResource (s@(Part _ _ save _))=
unless (not save) $ defaultWrite (defPath s ++ key s) (pack $ show s)
eval :: DDS a -> Cloud (PartRef a)
eval (DDS mx) = mx
type Path=String
instance F.Foldable DVU.Vector where
{-# INLINE foldr #-}
foldr = foldr
{-# INLINE foldl #-}
foldl = foldl
{-# INLINE foldr1 #-}
foldr1 = foldr1
{-# INLINE foldl1 #-}
foldl1 = foldl1
--foldlIt' :: V.Unbox a => (b -> a -> b) -> b -> V.Vector a -> b
--foldlIt' f z0 xs= V.foldr f' id xs z0
-- where f' x k z = k $! f z x
--
--foldlIt1 :: V.Unbox a => (a -> a -> a) -> V.Vector a -> a
--foldlIt1 f xs = fromMaybe (error "foldl1: empty structure")
-- (V.foldl mf Nothing xs)
-- where
-- mf m y = Just (case m of
-- Nothing -> y
-- Just x -> f x y)
class (F.Foldable c, Monoid (c a), Loggable (c a),Typeable c, Typeable a) => Distributable c a where
singleton :: a -> c a
splitAt :: Int -> c a -> (c a, c a)
fromList :: [a] -> c a
instance Loggable a => Loggable (DV.Vector a) where
serialize v= intDec (DV.length v) <> "/" <> foldl' (\s x -> s <> "/" <> serialize x ) mempty v
deserialize= do
len <- int
DV.replicateM len $ tChar '/' *> deserialize
instance (Typeable a, Loggable a) => Distributable DV.Vector a where
singleton = DV.singleton
splitAt= DV.splitAt
fromList = DV.fromList
instance (Loggable a,DVU.Unbox a) => Loggable (DVU.Vector a) where
serialize v= intDec (DVU.length v) <> "/" <> serialize(v DVU.! 0) <> DVU.ifoldl' (\s _ x -> s <> "/" <> serialize x) mempty (DVU.slice 1 (DVU.length v -1) v)
deserialize= do
len <- int
DVU.replicateM len $ tChar '/' *> deserialize
instance (Typeable a, Loggable a, DVU.Unbox a) => Distributable DVU.Vector a where
singleton= DVU.singleton
splitAt= DVU.splitAt
fromList= DVU.fromList
-- | perform a map and partition the result with different keys using boxed vectors
-- The final result will be used by reduce.
mapKeyB :: (Typeable a, Loggable a, Typeable b,Loggable b, Typeable k, Loggable k,Ord k)
=> (a -> (k,b))
-> DDS (DV.Vector a)
-> DDS (M.Map k(DV.Vector b))
mapKeyB= mapKey
-- | perform a map and partition the result with different keys using unboxed vectors
-- The final result will be used by reduce.
mapKeyU :: (Typeable a, Loggable a, DVU.Unbox a, Typeable b, Loggable b, DVU.Unbox b, Typeable k, Loggable k,Ord k)
=> (a -> (k,b))
-> DDS (DVU.Vector a)
-> DDS (M.Map k(DVU.Vector b))
mapKeyU= mapKey
-- | perform a map and partition the result with different keys.
-- The final result will be used by reduce.
mapKey :: (Distributable container a,Distributable container b, Typeable k, Loggable k,Ord k)
=> (a -> (k,b))
-> DDS (container a)
-> DDS (M.Map k (container b))
mapKey f (DDS mx)= DDS $ loggedc $ do
refs <- mx
process refs !> ("process",refs)
where
-- process :: Partition a -> Cloud [Partition b]
process (ref@(Ref node path sav))= runAt node $ local $ do
xs <- getPartitionData ref !> ("CMAP", ref,node)
(generateRef $ map1 f xs)
-- map1 :: (Ord k, F.Foldable container) => (a -> (k,b)) -> container a -> M.Map k(container b)
map1 f v= F.foldl' f1 M.empty v
where
f1 map x=
let (k,r) = f x
in M.insertWith (<>) k (Transient.MapReduce.singleton r) map
{-
map :: (Distributable container a,Distributable container b, Loggable k,Ord k)
=> (a -> b)
-> DDS (container a)
-> DDS (container b)
map f (DDS mx)= DDS $ loggedc $ do
refs <- mx
process refs -- !> ("process",refs)
where
-- process :: Partition a -> Cloud [Partition b]
process (ref@(Ref node path sav))= runAt node $ local $ do
xs <- getPartitionData ref -- !> ("CMAP", ref,node)
(generateRef $ map1 f xs) // xxx
!> "MAP"
map1 :: (Ord k, F.Foldable container) => (a -> b) -> container a -> container b
map1 f v= F.foldl' f1 M.empty v
where
f1 map x=
let r = f x
in M.insertWith (<>) k (Transient.MapReduce.singleton r) map
-}
data ReduceChunk a= EndReduce | Reduce a deriving (Typeable, Read, Show)
boxids= unsafePerformIO $ newIORef (0 :: Int)
reduce :: (Hashable k,Ord k, Distributable container a, Typeable k, Loggable k, Typeable a, Loggable a)
=> (a -> a -> a) -> DDS (M.Map k (container a)) ->Cloud (M.Map k a)
reduce red (dds@(DDS mx))= loggedc $ do
mboxid <- localIO $ atomicModifyIORef boxids $ \n -> let n'= n+1 in (n',n')
nodes <- local getEqualNodes
-- return () !> ("REDUCE NODES=", nodes)
let lengthNodes = length nodes
shuffler nodes = do
localIO $ threadDelay 100000
ref@(Ref node path sav) <- mx -- return the resulting blocks of the map
runAt node $ do
localIO $ return () !> ("FOLDANDSEND","REF",ref, "runAt", node)
foldAndSend node nodes ref
stop
-- groupByDestiny :: (Hashable k, Distributable container a) => M.Map k (container a) -> M.Map Int [(k ,container a)]
groupByDestiny map = M.foldlWithKey' f M.empty map
where
-- f :: M.Map Int [(k ,container a)] -> k -> container a -> M.Map Int [(k ,container a)]
f map k vs= M.insertWith (<>) (hash1 k) [(k,vs)] map
hash1 k= abs $ hash k `rem` length nodes
-- foldAndSend :: (Hashable k, Distributable container a)=> (Int,[(k,container a)]) -> Cloud ()
foldAndSend node nodes ref= do
pairs <- onAll $ getPartitionData1 ref
<|> return (error $ "DDS computed out of his node:"++ show ref )
let mpairs = groupByDestiny pairs
length <- local . return $ M.size mpairs
let port2= nodePort node
if length == 0 then sendEnd nodes else do
nsent <- onAll $ liftIO $ newMVar 0
(i,folded) <- local $ parallelize foldthem (M.assocs mpairs)
n <- localIO $ modifyMVar nsent $ \r -> return (r+1, r+1) :: IO (Int,Int)
-- sourcenode <- local getMyNode -- XXX borrar, solo para debug
--localIO $ return () !> ("PUTMAILBOX TOSEND from",sourcenode,n,length,i,folded)
(runAt (nodes !! i) $ do
return () !> "JUST BEFORE PUTMAILBOX"
local $ (putMailbox' mboxid (Reduce folded `asTypeOf` paramOf dds))
!> ("PUTMAILBOX SENT ",n,length,i,folded))
when (n == length) $ sendEnd nodes
return ()
empty
where
foldthem (i,kvs)= async . return
$ (i,map (\(k,vs) -> (k,foldl1 red vs)) kvs)
sendEnd nodes = do
-- node <- local getMyNode -- XXX quitar. solo para debug
onNodes nodes $ local $
putMailbox' mboxid (EndReduce `asTypeOf` paramOf dds)
-- !> ("PUTMAILBOX ENDREDUCE FROM", node))
onNodes nodes f = foldr (<|>) empty $ map (\n -> runAt n f ) nodes
sumNodes nodes f= foldr (<>) mempty $ map (\n -> runAt n f ) nodes
reducer nodes= sumNodes nodes reduce1 -- a reduce1 process in each node, get the results and mappend them
-- reduce1 :: (Ord k) => Cloud (M.Map k v)
reduce1 =local $ do
reduceResults <- liftIO $ newMVar M.empty
numberSent <- liftIO $ newMVar 0
return () !> "GETMAILBOX"
minput <- getMailbox' mboxid -- get the chunk once it arrives to the mailbox
case minput of
EndReduce -> do
return () !> "ENDREDUCE"
n <- liftIO $ modifyMVar numberSent $ \r -> let r'= r+1 in return (r', r')
if n == lengthNodes
!> ("END REDUCE RECEIVEDDDD",n, lengthNodes)
then do
deleteMailbox' mboxid (EndReduce `asTypeOf` paramOf dds)
r <- liftIO $ readMVar reduceResults
return r !> ("RETURNING",r)
else stop
Reduce kvs -> do
return () !> "REDUCE RECEIVEDDDDDDDDDDDD"
let addIt (k,inp) = do
let input= inp `asTypeOf` atype dds
liftIO $ modifyMVar_ reduceResults
$ \map -> do
let maccum = M.lookup k map
return $ M.insert k (case maccum of
Just accum -> red input accum
Nothing -> input) map
mapM addIt (kvs `asTypeOf` paramOf' dds)
!> ("RECEIVED REDUCEEEEEEEEEEEEE",kvs)
stop
r <- reducer nodes <|> shuffler nodes
localIO $ return () !> "RETRETRET"
return r
where
atype ::DDS(M.Map k (container a)) -> a
atype = undefined -- type level
paramOf :: DDS (M.Map k (container a)) -> ReduceChunk [( k, a)]
paramOf = undefined -- type level
paramOf' :: DDS (M.Map k (container a)) -> [( k, a)]
paramOf' = undefined -- type level
-- parallelize :: Loggable b => (a -> Cloud b) -> [a] -> Cloud b
parallelize f xs = foldr (<|>) empty $ map f xs
mparallelize f xs = loggedc $ foldr (<>) mempty $ map f xs
getPartitionData :: (Typeable a, Loggable a) => PartRef a -> TransIO a
getPartitionData (Ref node path save) = Transient $ do
mp <- (liftIO $ atomically
$ readDBRef
$ getDBRef
$ keyp path save)
`onNothing` error ("not found DDS data: "++ keyp path save)
case mp of
(Part _ _ _ xs) -> return $ Just xs
getPartitionData1 :: (Typeable a, Loggable a) => PartRef a -> TransIO a
getPartitionData1 (Ref node path save) = Transient $ do
mp <- liftIO $ atomically
$ readDBRef
$ getDBRef
$ keyp path save
case mp of
Just (Part _ _ _ xs) -> return $ Just xs
Nothing -> return Nothing
getPartitionData2 :: (Typeable a,Loggable a) => PartRef a -> IO a
getPartitionData2 (Ref node path save) = do
mp <- ( atomically
$ readDBRef
$ getDBRef
$ keyp path save)
`onNothing` error ("not found DDS data: "++ keyp path save)
case mp of
(Part _ _ _ xs) -> return xs
-- en caso de fallo de Node, se lanza un clustered en busca del path
-- si solo uno lo tiene, se copia a otro
-- se pone ese nodo de referencia en Part
runAtP :: Loggable a => Node -> (Path -> IO a) -> Path -> Cloud a
runAtP node f uuid= do
r <- runAt node $ onAll . liftIO $ (SLast <$> f uuid) `catch` sendAnyError
case r of
SLast r -> return r
SError e -> do
nodes <- mclustered $ search uuid
when(length nodes < 1) $ asyncDuplicate node uuid
runAtP ( head nodes) f uuid
search uuid= error $ "chunk failover not yet defined. Lookin for: "++ uuid
asyncDuplicate node uuid= do
forkTo node
nodes <- onAll getEqualNodes
let node'= head $ nodes \\ [node]
content <- onAll . liftIO $ readFile uuid
runAt node' $ local $ liftIO $ writeFile uuid content
sendAnyError :: SomeException -> IO (StreamData a)
sendAnyError e= return $ SError e
-- | distribute a container of values among many nodes.
-- If the container is static and sharable, better use the get* primitives
-- since each node will load the data independently.
distribute :: (Loggable a, Distributable container a ) => container a -> DDS (container a)
distribute = DDS . distribute'
distribute' xs= loggedc $ do
nodes <- local getEqualNodes -- !> "DISTRIBUTE"
let lnodes = length nodes
let size= case F.length xs `div` (length nodes) of 0 ->1 ; n -> n
xss= split size lnodes 1 xs -- !> size
r <- distribute'' xss nodes
return r
where
split n s s' xs | s==s' = [xs]
split n s s' xs=
let (h,t)= Transient.MapReduce.splitAt n xs
in h : split n s (s'+1) t
distribute'' :: (Loggable a, Distributable container a)
=> [container a] -> [Node] -> Cloud (PartRef (container a))
distribute'' xss nodes =
parallelize move $ zip nodes xss -- !> show xss
where
move (node, xs)= runAt node $ local $ do
par <- generateRef xs
return par
!> ("move", node,xs)
-- | input data from a text that must be static and shared by all the nodes.
-- The function parameter partition the text in words
getText :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)
getText part str= DDS $ loggedc $ do
nodes <- local getEqualNodes !> "getText"
return () !> ("DISTRIBUTE TEXT IN NODES:",nodes)
let lnodes = length nodes
parallelize (process lnodes) $ zip nodes [0..lnodes-1]
where
process lnodes (node,i)=
runAt node $ local $ do
let xs = part str
size= case length xs `div` lnodes of 0 ->1 ; n -> n
xss= Transient.MapReduce.fromList $
if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs
generateRef xss
!> "GETTEXT PROCESS"
-- | get the worlds of an URL
textUrl :: String -> DDS (DV.Vector Text.Text)
textUrl= getUrl (map Text.pack . words)
-- | generate a DDS from the content of a URL.
-- The first parameter is a function that divide the text in words
getUrl :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)
getUrl partitioner url= DDS $ do
nodes <- local getEqualNodes -- !> "DISTRIBUTE"
let lnodes = length nodes
parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss
where
process lnodes (node,i)= runAt node $ local $ do
r <- liftIO . simpleHTTP $ getRequest url
body <- liftIO $ getResponseBody r
let xs = partitioner body
size= case length xs `div` lnodes of 0 ->1 ; n -> n
xss= Transient.MapReduce.fromList $
if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs
generateRef xss
!> "GETURL"
-- | get the words of a file
textFile :: String -> DDS (DV.Vector Text.Text)
textFile= getFile (map Text.pack . words)
-- | generate a DDS from a file. All the nodes must access the file with the same path
-- the first parameter is the parser that generates elements from the content
getFile :: (Loggable a, Distributable container a) => (String -> [a]) -> String -> DDS (container a)
getFile partitioner file= DDS $ do
nodes <- local getEqualNodes -- !> "DISTRIBUTE"
let lnodes = length nodes
parallelize (process lnodes) $ zip nodes [0..lnodes-1] -- !> show xss
where
process lnodes (node, i)= runAt node $ local $ do
content <- do
c <- liftIO $ readFile file
length c `seq` return c
let xs = partitioner content
size= case length xs `div` lnodes of 0 ->1 ; n -> n
xss= Transient.MapReduce.fromList $
if i== lnodes-1 then drop (i* size) xs else take size $ drop (i * size) xs
generateRef xss
!> "GETFILE"
generateRef :: (Typeable a, Loggable a) => a -> TransIO (PartRef a)
generateRef x= do
node <- getMyNode
liftIO $ do
temp <- getTempName
let reg= Part node temp True x -- False to not save
atomically $ newDBRef reg
-- syncCache
(return $ getRef reg) -- !> ("generateRef",reg,node)
getRef (Part n t s x)= Ref n t s
getTempName :: IO String
getTempName= ("DDS" ++) <$> replicateM 5 (randomRIO ('a','z'))
-------------- Distributed Datasource Streams ---------
-- | produce a stream of DDS's that can be map-reduced. Similar to spark streams.
-- each interval of time,a new DDS is produced.(to be tested)
streamDDS
:: (Loggable a, Distributable container a) =>
Int -> IO (StreamData a) -> DDS (container a)
streamDDS time io= DDS $ do
xs <- local . groupByTime time $ do
r <- parallel io
case r of
SDone -> empty
SLast x -> return [x]
SMore x -> return [x]
SError e -> error $ show e
distribute' $ Transient.MapReduce.fromList xs
#endif