Holumbus-MapReduce (empty) → 0.0.1
raw patch · 48 files changed
+7038/−0 lines, 48 filesdep +Holumbus-Distributiondep +Holumbus-Storagedep +basesetup-changed
Dependencies added: Holumbus-Distribution, Holumbus-Storage, base, binary, bytestring, containers, directory, haskell98, hslogger, hxt, network, time, unix
Files
- Examples/Makefile +19/−0
- Examples/MapReduce/Crawler/Config.hs +306/−0
- Examples/MapReduce/Crawler/Crawl.hs +394/−0
- Examples/MapReduce/Crawler/CrawlerClient.hs +87/−0
- Examples/MapReduce/Crawler/CrawlerWorker.hs +74/−0
- Examples/MapReduce/Crawler/Index.hs +340/−0
- Examples/MapReduce/Crawler/Makefile +33/−0
- Examples/MapReduce/DummyFiles/ClientMain.hs +55/−0
- Examples/MapReduce/DummyFiles/WorkerMain.hs +64/−0
- Examples/MapReduce/Grep/ClientMain.hs +64/−0
- Examples/MapReduce/Grep/Grep.hs +210/−0
- Examples/MapReduce/Grep/Makefile +33/−0
- Examples/MapReduce/Grep/WorkerMain.hs +64/−0
- Examples/MapReduce/Makefile +18/−0
- Examples/MapReduce/Sort/ClientMain.hs +62/−0
- Examples/MapReduce/Sort/Makefile +33/−0
- Examples/MapReduce/Sort/Sort.hs +157/−0
- Examples/MapReduce/Sort/WorkerMain.hs +64/−0
- Examples/MapReduce/Standalone/Makefile +32/−0
- Examples/MapReduce/Standalone/StandaloneMain.hs +56/−0
- Examples/MapReduce/WordFrequency/ClientMain.hs +63/−0
- Examples/MapReduce/WordFrequency/Makefile +35/−0
- Examples/MapReduce/WordFrequency/WordFrequency.hs +142/−0
- Examples/MapReduce/WordFrequency/WorkerMain.hs +62/−0
- Examples/Utils/PickleTest/Makefile +31/−0
- Examples/Utils/PickleTest/PickleTest.hs +35/−0
- Holumbus-MapReduce.cabal +89/−0
- LICENSE +21/−0
- Makefile +58/−0
- Programs/Makefile +17/−0
- Programs/MapReduceMaster/Makefile +30/−0
- Programs/MapReduceMaster/Master.hs +60/−0
- README +119/−0
- Setup.hs +2/−0
- source/Holumbus/Distribution/DMapReduce.hs +233/−0
- source/Holumbus/Distribution/Master.hs +39/−0
- source/Holumbus/Distribution/Master/MasterData.hs +391/−0
- source/Holumbus/Distribution/Master/MasterPort.hs +169/−0
- source/Holumbus/Distribution/Messages.hs +208/−0
- source/Holumbus/Distribution/Worker.hs +39/−0
- source/Holumbus/Distribution/Worker/WorkerData.hs +192/−0
- source/Holumbus/Distribution/Worker/WorkerPort.hs +102/−0
- source/Holumbus/MapReduce/JobController.hs +761/−0
- source/Holumbus/MapReduce/MapReduce.hs +75/−0
- source/Holumbus/MapReduce/TaskProcessor.hs +545/−0
- source/Holumbus/MapReduce/Types.hs +961/−0
- source/Holumbus/MapReduce/UserInterface.hs +151/−0
- source/Holumbus/Standalone/SMapReduce.hs +243/−0
+ Examples/Makefile view
@@ -0,0 +1,19 @@+# Making all examples++FILESYSTEM = FileSystem+MAPREDUCE = MapReduce+UTILS = Utils/Filehandling Utils/PickleTest+ALLEXAMPLES = $(FILESYSTEM) $(MAPREDUCE) $(UTILS)+++all :+ $(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i all ;)++mr :+ $(foreach i,$(MAPREDUCE),$(MAKE) -C $i all ;)+ +wc :+ @wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+ $(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i $@ ;)
+ Examples/MapReduce/Crawler/Config.hs view
@@ -0,0 +1,306 @@+module Examples.MapReduce.Crawler.Config+(+ getConfig+, MRCrawlerConfig(..)+)+where+++import Text.XML.HXT.Arrow+import Holumbus.Build.Config+import Holumbus.Index.Documents+import Holumbus.Build.Crawl+import Holumbus.Build.Index+import Holumbus.Index.Documents +import Holumbus.Utility++import Data.Maybe+import Data.List+++data MRCrawlerConfig d a = MRCrawlerConfig {+ cc_CrawlerState :: CrawlerState d a+ , cc_TraceLevel :: Int+ , cc_DocsPerCrawl :: Int+ , cc_IndexerConfig :: IndexerConfig+ }+ + +getConfig = MRCrawlerConfig+ (initialCrawlerState idxConfig emptyDocuments customFunction) + traceLevel+ docsPerCrawl+ ic_test+ where+ idxConfig = ic_test+ traceLevel = 0+ docsPerCrawl = 10+++customFunction :: ArrowXml a => a XmlTree (Maybe Int)+customFunction = constA Nothing +++ic_test :: IndexerConfig+ic_test = ic_fhw {ic_startPages = ["http://april"], ic_fCrawlFilter = const True}+++ic_fhw :: IndexerConfig+ic_fhw + = IndexerConfig+ { ic_startPages = [ "http://www.fh-wedel.de/"+ , "http://www.fh-wedel.de/sonstiges/sitemap/"+ , "http://www.fh-wedel.de/wir-ueber-uns/mitarbeiter-innen/?no_cache=1"+ ]+ , ic_tempPath = Just "/tmp/"+ , ic_indexPath = "/tmp/fhw"+ , ic_indexerTimeOut = 10 * 60 * 1000000 + , ic_contextConfigs = ccs_fhw+ , ic_readAttributes = standardReadDocumentAttributes+ , ic_fCrawlFilter = simpleCrawlFilter -- [ "^http://www\\.fh-wedel\\.de"] -- + ["^http://[a-z]*\\.?fh-wedel\\.de" ] -- allow+ (["tx_fhwunternehmensforum_pi3" -- deny+ , "http://asta.fh-wedel.de" -- slow+ , "http://biblserv.fh-wedel.de" -- slow+ , "http://darcs.fh-wedel.de" -- hackers only+ , "http://stud.fh-wedel.de" -- boring+ , "http://holumbus.fh-wedel.de/branches"+ , "http://holumbus.fh-wedel.de/cgi-bin"+ , "/HXmlToolbox/hdoc", "si/doc/javadoc/docs"+ , "~herbert/html", "/java/jdk1.1.1/docs"+-- , "/~", "/%7E", "http://www.fh-wedel.de/mitarbeiter/"+ , "\\?L=0", "\\&L=0"+ , ".pdf$", ".jpg$", ".gif$", ".png$", ".tar.gz$"+ , ".ppt$", ".exe$", ".txt$", ".zip$", ".doc$"+ , ".dot$", ".png$", ".ps$", ".ps.gz$", ".nb$"+ , ".swf$", ".JPG$", ".tex$", ".rss$", ".mpg$"+ , ".mp3$", ".m3u$", ".java$", ".tgz$", ".svg", ".mdb$" + , ".PDF$", ".xls$", ".dta$", ".lst$", ".rar", ".avi$", ".mp4$" + , "%7Edi", "/~di"+ , "ws99/Ausarbeitung/mico/Beispiel"+ , "/rundgang/id=", "/vorlesungsplan/id="+ , "/vorlesungsplan/sem=", "/tv-infosystem/", "/~splan/"+ , "http://www\\.fh-wedel\\.de/index\\.php\\?eID=tx_cms_showpic"+ , "http://www.fh-wedel.de/fileadmin/mitarbeiter/ne/CG/opengl_man"+ , "http://www.fh-wedel.de/%7Esi/vorlesungen/c/beispiele"+ , "http://www.fh-wedel.de/~si/vorlesungen/c/beispiele"+ , "http://www.fh-wedel.de/~wol/fhtml" -- very slow and boring pages+ , "http://www.fh-wedel.de/%7Esi/vorlesungen/internet/TclTk/program1.html" -- slow+ ] ++ list404)+ }+ + + +list404 :: [String]+list404 = + [ "http://www.fh-wedel.de/work/medienlabor-der-fh-wedel/raeumlichkeiten/rechenzentrum/"+ , "http://www.fh-wedel.de/typo3"+ , "http://www.fh-wedel.de/~ue"+ , "http://www.fh-wedel.de/~wk"+ , "http://www.fh-wedel.de/work/medienlabor-der-fh-wedel/raeumlichkeiten/videostudio/"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/team.php"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/research/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/streams.php"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/vrlab/index.html"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/refs/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/specials.php"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/virtual-reality-labor/index.html"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/kontakt.php"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/virtual-reality-labor/cobench/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/online-tutorials/vlc-usage.htm"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/index.php"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/courses/index.html"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/virtual-reality-labor/cave/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/online-tutorials/toc_index.htm"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/vrlab/cobench/index.html"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/news/index.html"+ , "http://www.fh-wedel.de/wir-ueber-uns/labore-fe/informatik/contact/index.html"+ , "http://www.fh-wedel.de/studierende/campus-infos/fh-stream/faq.php"+ , "http://www.fh-wedel.de/internationaloffice/Deutsch/"+ , "http://www.fh-wedel.de/%7Ege/Seminar.htm"+ , "http://www.fh-wedel.de/HXmlToolbox/index.html"+ , "http://www.fh-wedel.de/Literatur_08.html"+ , "http://www.fh-wedel.de/Seminar_Algorithmische_Graphentheorie.html"+ , "http://www.fh-wedel.de/Seminar_Kombinatorik.html"+ , "http://www.fh-wedel.de/absolventen/members/absolventensuche.html"+ , "http://www.fh-wedel.de/analysis-im-Netz.htm"+ , "http://www.fh-wedel.de/bba/index.html"+ , "http://www.fh-wedel.de/buecher.html"+ , "http://www.fh-wedel.de/buero_n15.html"+ , "http://www.fh-wedel.de/c/index.html"+ , "http://www.fh-wedel.de/cg1/index.html"+ , "http://www.fh-wedel.de/cg2/index.html"+ , "http://www.fh-wedel.de/cprog.html"+ , "http://www.fh-wedel.de/ctrl.html"+ , "http://www.fh-wedel.de/dbprog.html"+ , "http://www.fh-wedel.de/diplom.html"+ , "http://www.fh-wedel.de/download.htm"+ , "http://www.fh-wedel.de/downloads.html"+ , "http://www.fh-wedel.de/dozent.html"+ , "http://www.fh-wedel.de/faelle.html"+ , "http://www.fh-wedel.de/gesetze.html"+ , "http://www.fh-wedel.de/gisela.html"+ , "http://www.fh-wedel.de/handout.html"+ , "http://www.fh-wedel.de/index.html"+ , "http://www.fh-wedel.de/index/index.html"+ , "http://www.fh-wedel.de/infos/mitarbeiter"+ , "http://www.fh-wedel.de/infos/mitarbeiter/index.html"+ , "http://www.fh-wedel.de/infos/sprechzeiten/"+ , "http://www.fh-wedel.de/klausuren.html"+ , "http://www.fh-wedel.de/~si/seminare/ws05/Ausarbeitung/7.concurrent/index.html"+ , "http://www.fh-wedel.de/cis/semesterinfo/richtlinien/doku-richtl.html"+ , "http://www.fh-wedel.de/~si/seminare/ws06/Ausarbeitung/04.UserModeLinux/index.html"+ , "http://www.fh-wedel.de/cis/semesterinfo/studienordnung/index.html"+ , "http://www.fh-wedel.de/~ge/Datenbanken.htm"+ , "http://www.fh-wedel.de/~si/seminare/ws06/Ausarbeitung/14.Cocoon/cocoon0.htm"+ , "http://www.fh-wedel.de/~si/seminare/ws00/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/~si/seminare/ws01/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/%7Efi/odb/index.html"+ , "http://www.fh-wedel.de/~si/seminare/ws02/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/%7Ekap/"+ , "http://www.fh-wedel.de/~si/seminare/ws99/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/%7Eps2/programmierstil/delphi_progstil.html"+ , "http://www.fh-wedel.de/%7Eps2/programmierstil/inline_doku.html"+ , "http://www.fh-wedel.de/%7Eps2/programmierstil/pascal_progstil.html"+ , "http://www.fh-wedel.de/~mo/lectures/seminar-sose03.html"+ , "http://www.fh-wedel.de/~si/seminare/ws03/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/~si/HXmlToolbox/hdoc/index.htm"+ , "http://www.fh-wedel.de/archiv/iw/Lehrveranstaltungen/WS2006/GdT.html"+ , "http://www.fh-wedel.de/~si/seminare/ws04/Ausarbeitung/index.html"+ , "http://www.fh-wedel.de/~si/seminare/ws04/Termine/www.wikipedia.org"+ , "http://www.fh-wedel.de/~ki/ia/sem3/seminar1/code.htm"+ , "http://www.fh-wedel.de/klausuren/Compilerbau.html"+ , "http://www.fh-wedel.de/klausuren/CundOOP.html"+ , "http://www.fh-wedel.de/klausuren/SD.html"+ , "http://www.fh-wedel.de/klausuren/Unix.html"+ , "http://www.fh-wedel.de/klausuren/index.html"+ , "http://www.fh-wedel.de/klr2.html"+ , "http://www.fh-wedel.de/kontakt.html"+ , "http://www.fh-wedel.de/lebenslauf.html"+ , "http://www.fh-wedel.de/links.htm"+ , "http://www.fh-wedel.de/links.html"+ , "http://www.fh-wedel.de/material.html"+ , "http://www.fh-wedel.de/mec_MInf.html"+ , "http://www.fh-wedel.de/mitarbeiter/iw/lehrveranstaltungen-im-naechsten-semester/informatik-seminar-algorithmen/"+ , "http://www.fh-wedel.de/oop/index.html"+ , "http://www.fh-wedel.de/or.html"+ , "http://www.fh-wedel.de/personal.html"+ , "http://www.fh-wedel.de/phys1_TInf.html"+ , "http://www.fh-wedel.de/phys1_WIng.html"+ , "http://www.fh-wedel.de/planspiel.html"+ , "http://www.fh-wedel.de/planspiel2.html"+ , "http://www.fh-wedel.de/pmod.html"+ , "http://www.fh-wedel.de/praktika/MultimediaProjekte/SVG/SVG_Tutorial_mi3794/index.htm"+ , "http://www.fh-wedel.de/praktika/MultimediaProjekte/index.html"+ , "http://www.fh-wedel.de/praktika/SoftwarePraktikum/index.html"+ , "http://www.fh-wedel.de/projekt.html"+ , "http://www.fh-wedel.de/projekte/aktuell.htm"+ , "http://www.fh-wedel.de/pse.html"+ , "http://www.fh-wedel.de/rechnernetze/nwslab.htm"+ , "http://www.fh-wedel.de/seite2.htm"+ , "http://www.fh-wedel.de/seminare.html"+ , "http://www.fh-wedel.de/seminare/index.html"+ , "http://www.fh-wedel.de/seminare/ss03/Termine/Themen.html"+ , "http://www.fh-wedel.de/seminare/ss04/Termine/Themen.html"+ , "http://www.fh-wedel.de/seminare/ws03/Termine/Themen.html"+ , "http://www.fh-wedel.de/seminare/ws04/Termine/Themen.html"+ , "http://www.fh-wedel.de/seminare/ws05/Termine/Themen.html"+ , "http://www.fh-wedel.de/seminare/ws06/Termine/Themen.html"+ , "http://www.fh-wedel.de/sprechstunde.htm"+ , "http://www.fh-wedel.de/suche/"+ , "http://www.fh-wedel.de/termine.htm"+ , "http://www.fh-wedel.de/termine/Klausureinsicht.html"+ , "http://www.fh-wedel.de/termine/Sprechstunde.html"+ , "http://www.fh-wedel.de/termine/index.html"+ , "http://www.fh-wedel.de/toc_index.htm"+ , "http://www.fh-wedel.de/uebungen/hinweise/hinweise.html"+ , "http://www.fh-wedel.de/uebungen/oop/aufgaben/aufg04.html"+ , "http://www.fh-wedel.de/uebungen/oop/aufgaben/aufg05.html"+ , "http://www.fh-wedel.de/uebungen/oop/index.html"+ , "http://www.fh-wedel.de/uebungen/oop/richtlinien.html"+ , "http://www.fh-wedel.de/uebungen/unix/index.html"+ , "http://www.fh-wedel.de/veranstaltungen.htm"+ , "http://www.fh-wedel.de/vl.html"+ , "http://www.fh-wedel.de/vlc-usage.htm"+ , "http://www.fh-wedel.de/vorlesungen/c/c.html"+ , "http://www.fh-wedel.de/vorlesungen/cb/cb.html"+ , "http://www.fh-wedel.de/vorlesungen/fp/fp.html"+ , "http://www.fh-wedel.de/vorlesungen/index.html"+ , "http://www.fh-wedel.de/vorlesungen/internet/Literatur/Buecher.html"+ , "http://www.fh-wedel.de/vorlesungen/internet/internet.html"+ , "http://www.fh-wedel.de/vorlesungen/java/Literatur/Buecher.html"+ , "http://www.fh-wedel.de/vorlesungen/java/java.html"+ , "http://www.fh-wedel.de/vorlesungen/softwaredesign/Literatur/Buecher.html"+ , "http://www.fh-wedel.de/vorlesungen/softwaredesign/design.html"+ , "http://www.fh-wedel.de/vr/index.html"+ , "http://www.fh-wedel.de/xml2html.2/xml2html/xml/index.xml"+ , "http://www.fh-wedel.de/zettelkasten/etc/archiv.html"+ , "http://www.fh-wedel.de/zettelkasten/etc/index.html"+ , "http://www.fh-wedel.de/zettelkasten/etc/schwob.html"+ , "http://www.fh-wedel.de/zettelkasten/fortune/fhspruch.html"+ , "http://www.fh-wedel.de/zettelkasten/fortune/index.html"+ , "http://www.fh-wedel.de/zettelkasten/fortune/index.html?KEY=1147779735"+ ] ++ +-- | The list of context configurations for the Fh Wedel pages+ccs_fhw :: [ContextConfig]+ccs_fhw = []+ ++ [cc_title]+ ++ [cc_meta]+ ++ [cc_content]+ ++ [cc_raw]+ ++ -- | Context for title-tags+cc_title :: ContextConfig+cc_title+ = ContextConfig+ { cc_name = "title"+ , cc_preFilter = this+ , cc_fExtract = getXPathTrees "/html/head/title" + , cc_fTokenize = map (stripWith (=='.')) . (parseWords isWordChar)+ , cc_fIsStopWord = const False -- (\s -> length s < 2)+ , cc_addToCache = False+ }+ +-- | Context for meta information. Description and keywords will be indexed+cc_meta :: ContextConfig+cc_meta+ = ContextConfig+ { cc_name = "meta"+ , cc_preFilter = this+ , cc_fExtract = getXPathTrees "/html/head/meta[@name='description' or @name='keywords']/@content" + , cc_fTokenize = map (stripWith (=='.')) . (parseWords isWordChar)+ , cc_fIsStopWord = (\s -> length s < 2)+ , cc_addToCache = False+ } + +-- | Context for normal page content. This indexes everything that is inside a div element with id+-- "col2_content" as it is defined in the central fhw template. Pages that do not use the normal+-- template will only be indexed in the raw context+cc_content :: ContextConfig+cc_content+ = ContextConfig+ { cc_name = "content"+ , cc_preFilter = this+ , cc_fExtract = getXPathTrees "//div[@id='col2_content']" + , cc_fTokenize = map (stripWith (=='.')) . (parseWords isWordChar)+ , cc_fIsStopWord = (\s -> length s < 2)+ , cc_addToCache = True+ }++-- | Raw context. This indexes all the text that is listed on the page without check if it is in+-- a menu or if it is real text. This context will have very low importance for search results+cc_raw :: ContextConfig+cc_raw+ = ContextConfig+ { cc_name = "raw"+ , cc_preFilter = this+ , cc_fExtract = getXPathTrees "//body" + , cc_fTokenize = map (stripWith (=='.')) . (parseWords isWordChar)+ , cc_fIsStopWord = (\s -> length s < 2)+ , cc_addToCache = False+ }+
+ Examples/MapReduce/Crawler/Crawl.hs view
@@ -0,0 +1,394 @@+-- ----------------------------------------------------------------------------++{- |+ Module : Holumbus.Build.Crawl+ Copyright : Copyright (C) 2008 Sebastian M. Schlatt+ License : MIT+ + Maintainer : Sebastian M. Schlatt (sms@holumbus.org)+ Stability : experimental+ Portability: portable+ Version : 0.1+ + This module provides some functions to crawl Web Pages recursively++-}++-- ----------------------------------------------------------------------------+{-# OPTIONS -fglasgow-exts -fno-warn-orphans #-}+-- -----------------------------------------------------------------------------++module Examples.MapReduce.Crawler.Crawl+ (+ -- * CrawlerState type+ CrawlerState (..)++ , crawlerAction ++ -- * Crawling+ , crawl+ + -- * Initializing+ , initialCrawlerState+-- , getRefs+ + -- * Persistence+ , loadCrawlerState+ , saveCrawlerState+ )+where++import Control.Monad hiding (when)++import Data.Binary+import Data.Char+import Data.List+import Data.Maybe+import qualified Data.Map as M+import qualified Data.Set as S+-- import qualified Data.IntMap as IM+import Data.Digest.MD5+import Data.ByteString.Lazy.Char8(pack)++import Holumbus.Build.Config+-- import Holumbus.Control.MapReduce.ParallelWithClass -- Persistent+-- import Holumbus.Control.MapReduce.MapReducible++import Holumbus.Index.Common+import Holumbus.Utility++import Text.XML.HXT.Arrow hiding (getXPathTrees)+import Text.XML.HXT.Arrow.XPathSimple+-- import System.Directory+import System.Time++import qualified Holumbus.FileSystem.FileSystem as FS+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce++import Examples.MapReduce.Crawler.Config++-- import Control.Parallel.Strategies++type MD5Hash = String+++-- because of IO(Maybe v3) --> has to instanciate the typeclass show+instance Show (CrawlerState d a) where+ show _ = "CrawlerState"++-- ----------------------------------------------------------------------------+-- Map-Action+-- ----------------------------------------------------------------------------+++crawlerAction+ :: (Binary a, Show a, HolDocuments d a)+ => MRCrawlerConfig d a -- whitness and configuration+ -> ActionConfiguration + (Int, CrawlerState d a) -- state+ DocId String -- k1, v1+ () (MD5Hash, Maybe (Document a), S.Set URI) -- k2, v2+ (MD5Hash, Maybe (Document a), S.Set URI) -- v3 == v2+ (CrawlerState d a) -- v4+crawlerAction cc+ = (defaultActionConfiguration "CRAWL")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration (mapCrawl cc)) + reduceAction+ = (defaultReduceConfiguration (reduceCrawl cc))+ { rc_Merge = mergeCrawl }+++mapCrawl+ :: (Binary a, Show a, HolDocuments d a)+ => MRCrawlerConfig d a -- ^ static configuration+ -> ActionEnvironment+ -> (Int, CrawlerState d a) -- ^ state+ -> DocId -> String -- ^ key - value+ -> IO [((), (MD5Hash, Maybe (Document a), S.Set URI))]+mapCrawl cc env (traceLevel,_) docId theUri+ = do+ let cs = cc_CrawlerState cc+ attrs = cs_readAttributes cs + tmpPath = cs_tempPath cs+ getRefs = cs_fGetReferences cs+ getCustom = cs_fGetCustom cs+ fileSystem = ae_FileSystem env++ runX (traceMsg 0 ("DocId: " ++ show docId ++ " - Uri: " ++ show theUri))+ -- putStrLn "################"+ -- putStrLn $ show attrs+ -- putStrLn "################"+ res <- crawlDoc traceLevel fileSystem attrs tmpPath getRefs getCustom docId theUri+ runX (traceMsg 0 ("Result: "))+ runX (traceMsg 0 (show res))+ return res+++-- ----------------------------------------------------------------------------+-- Reduce-Action+-- ----------------------------------------------------------------------------++mergeCrawl+ :: (Binary a, Show a, HolDocuments d a)+ -- => CrawlerState d a -- ^ type Whitness+ => ActionEnvironment+ -> (Int,CrawlerState d a) -- ^ state+ -> [((), (MD5Hash, Maybe (Document a), S.Set URI))]+ -> IO [((), [(MD5Hash, Maybe (Document a), S.Set URI)])]+mergeCrawl _ _ ls = return [((), ls')]+ where+ ls' = foldr (\(_,v) vs -> v:vs) [] ls+++reduceCrawl+ :: (Binary a, Show a, HolDocuments d a)+ => MRCrawlerConfig d a -- ^ type Whitness + -> ActionEnvironment+ -> (Int,CrawlerState d a) -- ^ state+ -> () -> [(MD5Hash, Maybe (Document a), S.Set URI)]+ -> IO (Maybe (CrawlerState d a))+reduceCrawl cc _ (_,cs) k ls = processCrawlResults cs' cs k ls+ where+ cs' = cc_CrawlerState cc++++-- | The crawl function. MapReduce is used to scan the documents for references to other documents+-- that have to be added to the 'Documents', too.+crawl+ :: (Show a, HolDocuments d a, Binary a, Binary (d a), XmlPickler (d a)+ , MapReduce mr) + => MRCrawlerConfig d a -- whitness and configuration+ -> Int -> Int+ -> CrawlerState d a+ -> mr+ -> IO (d a)+crawl config traceLevel maxDocs cs mr = + if S.null ( cs_toBeProcessed cs ) -- if no more documents have to be processed, + then do+ runX (traceMsg 1 (" no more documents to processed"))+ return (cs_docs cs) -- the Documents are returned+ else do -- otherwise, the crawling is continued+ runX (traceMsg 1 (" new crawler-cycle"))+ -- get the Docs that have to be processed and add docIds+ -- for the MapReduce Computation+ d <- return $ if maxDocs == 0 + then cs_toBeProcessed cs+ else (S.fromList . (take maxDocs) . S.toList) (cs_toBeProcessed cs)+ d' <- return $ zip [(cs_nextDocId cs)..] (S.toList d) -- (S.toList $ cs_toBeProcessed cs)+ runX (traceMsg 1 (" next docs to be pulled:"))+ runX (traceMsg 1 (show d'))+ + -- saveCrawlerState ( (fromMaybe "/tmp/" (cs_tempPath cs) ) ++ "CrawlerState.bin") cs+ -- writeToXmlFile ( (fromMaybe "/tmp/" (cs_tempPath cs) ) ++ "Docs.xml") (cs_docs cs)+ + runX (traceMsg 1 (" Status: already processed: " ++ show (S.size $ cs_wereProcessed cs) ++ + ", to be processed: " ++ show (S.size $ cs_toBeProcessed cs)))+ + cs' <- return $ cs { cs_nextDocId = cs_nextDocId cs + length d'+ , cs_wereProcessed = S.union d (cs_wereProcessed cs)+ , cs_toBeProcessed = S.difference (cs_toBeProcessed cs) d+ }++ (res,_) <- doMapReduce (crawlerAction config) (traceLevel,cs') d' [] 1 5 1 1 TOTRawTuple mr+ + runX (traceMsg 1 (" result of this cycle: ")) + + let (_,csnew) = head res+ + runX (traceMsg 1 (show $ cs_toBeProcessed csnew))+ + crawl config traceLevel maxDocs csnew mr+++-- | The REDUCE function for the crawling MapReduce computation. The 'Documents' and their contained+-- links are used to modify the 'CrawlerState'. New documents are added to the list of +-- unprocessed docs and the data of the already crawled documents are added to the 'Documents'. +processCrawlResults :: (HolDocuments d a, Binary a) => + CrawlerState d a+ -> CrawlerState d a -- ^ state before last MapReduce job+ -> () -- ^+ -> [(MD5Hash, Maybe (Document a), S.Set URI)] -- ^ data produced in the crawl phase+ -> IO (Maybe (CrawlerState d a))+processCrawlResults cc oldCs _ l = + do + cs' <- foldM process oldCs l+ runX (traceMsg 1 (" ----------------------"))+ runX (traceMsg 1 (" old cs: "))+ runX (traceMsg 1 (show $ cs_toBeProcessed oldCs))+ runX (traceMsg 1 (" ----------------------"))+ runX (traceMsg 1 (" new cs: "))+ runX (traceMsg 1 (show $ cs_toBeProcessed cs'))+ return $ Just cs'+ where + process :: (HolDocuments d a, Binary a) => + CrawlerState d a -> (String, Maybe (Document a), S.Set URI) -> IO (CrawlerState d a)+ process cs (theMD5, mdoc, refs) + = if isJust mdoc+ then + do+ if isJust (cs_docHashes cs) && M.member theMD5 (fromJust $ cs_docHashes cs)+ then do+ let hashes = fromJust $ cs_docHashes cs+ old <- M.lookup theMD5 hashes+ new <- return $ uri $ fromJust mdoc+ (newDocs, newHashes) <- update (cs_docs cs) hashes theMD5 old new+ return cs { cs_docs = newDocs + , cs_toBeProcessed = S.union + (cs_toBeProcessed cs) + (S.difference (S.filter (cs_fCrawlFilter cc) refs) + (cs_wereProcessed cs)+ )+ , cs_docHashes = Just $ newHashes+ }+ else do+ return $ + cs { cs_toBeProcessed = S.union (cs_toBeProcessed cs) + (S.difference (S.filter (cs_fCrawlFilter cc) refs) + (cs_wereProcessed cs)+ )+ , cs_docs = snd (insertDoc (cs_docs cs) (fromJust mdoc)) + , cs_docHashes = if isJust (cs_docHashes cs)+ then Just $ M.insert theMD5 (uri $ fromJust mdoc) (fromJust $ cs_docHashes cs)+ else Nothing+ }+ else return cs+ update :: (HolDocuments d a, Binary a) => + d a -> M.Map String URI -> String -> URI -> URI -> IO (d a, M.Map String URI)+ update docs hashes md5String oldUri newUri + = if length oldUri <= length newUri + then return $ (docs, hashes)+ else do + docId <- lookupByURI docs oldUri+ oldDoc <- lookupById docs docId+ newDoc <- return oldDoc {uri = newUri}+ return (updateDoc docs docId newDoc, M.insert md5String newUri hashes)++++-- | Wrapper function for the "real" crawlDoc functions. The current time is+-- traced (to identify documents that take a lot of time to be processed while+-- testing) and the crawlDoc'-arrow is run+crawlDoc :: (Binary a, Show a) => + Int + -> FS.FileSystem+ -> Attributes+ -> Maybe String+ -> IOSArrow XmlTree [URI]+ -> Custom a+ -> DocId + -> String + -> IO [((), (MD5Hash, Maybe (Document a), S.Set URI))]+crawlDoc traceLevel fileSystem attrs tmpPath getRefs getCustom docId theUri + = do+ clt <- getClockTime -- get Clock Time for debugging+ cat <- toCalendarTime clt -- and convert it to "real" date and time+ r <- runX ( setTraceLevel traceLevel+ >>> traceMsg 1 (calendarTimeToString cat) + >>> ( constA () -- this is needed to fit the MapReduce abstraction+ &&& crawlDoc' traceLevel fileSystem attrs tmpPath getRefs getCustom (docId, theUri) + )+ >>^ (\(i, (h, d, u)) -> (i, (h, d, S.fromList u))) + )+-- putStrLn (show r)+ return $ r ++-- | Download & read document and compute document title and included refs to+-- other documents +crawlDoc' :: (Binary a) =>+ Int+ -> FS.FileSystem+ -> Attributes -- ^ options for readDocument+ -> Maybe String -- ^ path for serialized tempfiles+ -> IOSArrow XmlTree [URI]+-- -> [String]+ -> Custom a+ -> (DocId, URI) -- ^ DocId, URI+ -> IOSArrow c (String, Maybe (Document a), [URI])+crawlDoc' traceLevel fileSystem attrs tmpPath getRefs getCustom (docId, theUri) =+ traceMsg 1 (" crawling document: " ++ show docId ++ " -> " ++ show theUri )+ >>> withTraceLevel (traceLevel - traceOffset) (readDocument attrs theUri)+ >>> (+ documentStatusOk `guards` -- make sure that the document could be accessed+ -- if it is set in the crawler options, write a temporary copy of the document to+ -- the hard disk+ ( ( if isJust tmpPath+ then withTraceLevel (traceLevel - traceOffset) $+ (replaceBaseElement $< computeDocBase) -- make sure that there ist a <base> tag+ -- a b c >>> a c d -> a b d+ >>> -- writeDocument :: Attributes -> String -> IOStateArrow s XmlTree XmlTree+ -- writeDocument standardWriteTmpDocumentAttributes+ -- (fromJust tmpPath ++ tmpFile docId theUri)+ + -- writeDocumentToString :: Attributes -> IOStateArrow s XmlTree String + (writeDocumentToString standardWriteTmpDocumentAttributes+ >>> + -- arrIO :: (b -> IO c) -> a b c+ arrIO (\s -> FS.createFile + ({- fromJust tmpPath ++ -} tmpFile docId theUri) + (encode s)+ fileSystem))+ &&& (returnA)+ >>> arr (\(_,t) -> t)+ --IOStateArrow s XmlTree XmlTree+ -- arr b c arr c d arr b d+ --IOStateArrow s XmlTree String >>> IOStateArrow String () >>> IOSTateArrow XmlTree () + --arr b c >>> arr c d >> arr b c+ else this+ ) + -- compute a pair of Document b (Holumbus datatype) and a list of contained links+ >>> + ( ( xshow (this) >>^ (show . md5 . pack) )+ &&& getDocument theUri getCustom+ &&& ( getRefs >>> strictA >>> perform (theTrace $< arr length) )+ ) >>^ (\(a,(b,c)) -> (a,b,c))+ )+ `orElse` ( -- if an error occurs with the current document, the global + traceMsg 0 ( "something went wrong with doc: \"" ++ theUri ++"\"") + >>> clearErrStatus -- error status has to be reset, else the crawler would stop+ >>> constA (show ( md5 (pack "foo")) , Nothing, []) -- Nothing indicates the error, the empty list shows+ ) -- that - caused by the error - no new links were found+ )+ where + theTrace i = traceMsg 2 ("found " ++ (show i) ++ " references")+++-- | Replaces the <base> tag in the Document. If there is no <base> tag, a new one is inserted.+-- This is especially usefull when saving the document to a local file. With a <base> tag+-- links in the document can be properly resolved.+replaceBaseElement :: ArrowXml a => String -> a XmlTree XmlTree+replaceBaseElement base = processTopDownUntil ( (isElem >>> hasName "html") `guards` processHtml )+ where+ processHtml = processTopDownUntil ( (isElem >>> hasName "head") `guards` processHead)+ processHead = ifA (getChildren >>> isElem >>> hasName "base") + (processTopDownUntil ( (isElem >>> hasName "base") `guards` processAttr) )+ addBaseElem+ processAttr = processAttrl ( changeAttrValue (\_ -> base ) `when` hasName "href")+ addBaseElem = replaceChildren (getChildren <+> aelem "base" [sattr "href" base]) + ++-- | extract the Title of a Document (for web pages the <title> tag) and combine+-- it with the document uri+getDocument :: (Binary b) => + URI -- ^ URI of the 'Document' + -> Custom b -- ^ Function to extract the custom Data from the 'Document'+ -> IOSArrow XmlTree (Maybe (Document b))+getDocument theUri getCustom + = mkDoc $<<< ( -- get the document title. to avoid failure of the arrow if the title tag is+ -- absent, a default value is supplied+ ( withDefault (getXPathTrees "/html/head/title/text()" >>> getText) "") + &&& constA theUri+ &&& getCustom+ )+ where+ mkDoc t u c = constA $ Just $ Document t u c +++
+ Examples/MapReduce/Crawler/CrawlerClient.hs view
@@ -0,0 +1,87 @@+module Main(main)+where++-- import qualified Data.IntMap as IM+-- import qualified Data.Set as S+-- import Data.Maybe+-- import Data.Binary+-- import Data.List+++-- Holumbus (SearchEngine)+import Control.Monad+import Text.XML.HXT.Arrow+-- import Holumbus.Build.Config+-- import Holumbus.Build.Crawl+-- import Holumbus.Build.Index+import Holumbus.Index.Common+-- import Holumbus.Index.Documents +-- import Holumbus.Index.Inverted.Memory(emptyInverted)+-- import Holumbus.Index.Cache+import Holumbus.Utility+++-- Holumbus (MapReduce)+import Holumbus.Common.Logging+import Holumbus.Network.PortRegistry.PortRegistryPort+import Holumbus.Distribution.DMapReduce++++import Examples.MapReduce.Crawler.Crawl+import Examples.MapReduce.Crawler.Index+import Examples.MapReduce.Crawler.Config++main :: IO ()+main + = do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ + let cc = getConfig+ traceLevel = cc_TraceLevel cc+ docsPerCrawl = cc_DocsPerCrawl cc+ crawlerState = cc_CrawlerState cc+ idxConfig = cc_IndexerConfig cc+ -- ---------------------------------------------------------------------------------------------+ -- CRAWLING+ -- ---------------------------------------------------------------------------------------------+ + mr <- mkMapReduceClient defaultMRClientConfig + + runX (traceMsg 0 (" crawling ----------------------------- " ))+ docs <- crawl cc traceLevel docsPerCrawl crawlerState mr+ localDocs <- return $ tmpDocs "" {- (fromMaybe "/tmp" (ic_tempPath idxConfig)) -} docs++ -- writeToBinFile ( (ic_indexPath idxConfig) ++ "-docs.bin") (docs)+ + runX (traceMsg 0 (" docs ----------------------------- " ))+ runX (traceMsg 0 (show docs)) ++ + -- ---------------------------------------------------------------------------------------------+ -- INDEXING+ -- ---------------------------------------------------------------------------------------------+ runX (traceMsg 0 (" indexing ----------------------------- " ))++ -- c <- createCache ((ic_indexPath idxConfig) ++ "-cache.db")++ idx <- buildIndex cc {- workerThreads traceLevel -} localDocs {- idxConfig emptyInverted (Just c) -} mr++ runX (traceMsg 0 (" writing index to file ----------------------------- " ))+ + writeToXmlFile ( (ic_indexPath idxConfig) ++ "-index.xml") idx+ -- writeToBinFile ( (ic_indexPath idxConfig) ++ "-index.bin") idx++ return()+{- where+ mergeIndexes' i1 f = do+ i2 <- loadFromBinFile (f ++ "-index.bin")+ return $ mergeIndexes i1 i2+-}+{- +fromDocuments :: Binary a => CrawlerState Documents a -> Documents a -> CrawlerState Documents a+fromDocuments cs ds = cs { cs_toBeProcessed = S.fromList ( map (uri . snd) ( IM.toList $ toMap ds )) } +-} +
+ Examples/MapReduce/Crawler/CrawlerWorker.hs view
@@ -0,0 +1,74 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Control.Exception++-- import Text.XML.HXT.Arrow ++import qualified Holumbus.Data.KeyMap as KMap+import Holumbus.MapReduce.Types++import Holumbus.Common.Logging+import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++import Examples.MapReduce.Crawler.Config+import Examples.MapReduce.Crawler.Crawl+import Examples.MapReduce.Crawler.Index++version :: String+version = "Holumbus-Distribution Standalone-Worker 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handle (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version + deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do+ fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig++ let cc = getConfig + let actions = KMap.insert (readActionConfiguration $ indexerOccurrencesAction cc) $+ KMap.insert (readActionConfiguration $ indexerBuildIndexAction) $+ KMap.insert (readActionConfiguration $ crawlerAction cc) $ + KMap.empty+ + let config = MR.defaultMRWorkerConfig+ mr <- MR.mkMapReduceWorker fs actions config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs+
+ Examples/MapReduce/Crawler/Index.hs view
@@ -0,0 +1,340 @@+-- ----------------------------------------------------------------------------++{- |+ Module : Holumbus.Build.Crawl+ Copyright : Copyright (C) 2008 Sebastian M. Schlatt+ License : MIT+ + Maintainer : Sebastian M. Schlatt (sms@holumbus.org)+ Stability : experimental+ Portability: portable+ Version : 0.1+ + Indexer functions++-}++-- -----------------------------------------------------------------------------+{-# OPTIONS -fglasgow-exts #-}+-- -----------------------------------------------------------------------------++module Examples.MapReduce.Crawler.Index + (+ indexerOccurrencesAction+ , indexerBuildIndexAction+ -- * Building indexes+ , buildIndex+ -- , buildIndexM+ + -- * Indexer Configuration+ , IndexerConfig (..)+ -- , ContextConfig (..)+ -- , mergeIndexerConfigs+ + -- , getTexts+ )++where++import Data.List+import Data.Binary hiding (Word)+import qualified Data.Map as M+import qualified Data.IntMap as IM+import qualified Data.IntSet as IS+import Data.Maybe++-- import Control.Exception+import Control.Monad+import Control.Parallel.Strategies++import Holumbus.Build.Config+-- import Holumbus.Control.MapReduce.ParallelWithClass +import Holumbus.Index.Common+import Holumbus.Index.Inverted.Memory+import Holumbus.Utility++import System.Time++import Text.XML.HXT.Arrow hiding (getXPathTrees) -- import all stuff for parsing, validating, and transforming XML++import qualified Holumbus.FileSystem.FileSystem as FS+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce++import Examples.MapReduce.Crawler.Config++++indexerOccurrencesAction+ :: MRCrawlerConfig d a+ -> ActionConfiguration + () -- state+ DocId String -- k1, v1+ (Context,Word) Occurrences -- k2, v2+ Occurrences -- v3 == v2+ Occurrences -- v4+indexerOccurrencesAction cc+ = (defaultActionConfiguration "INDEX_OCCURRENCES")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration (mapIndex cc))+ reduceAction+ = (defaultReduceConfiguration (reduceOccurrences))+++mapIndex + :: MRCrawlerConfig d a+ -> ActionEnvironment -> ()+ -> DocId -> String+ -> IO [((Context, Word), Occurrences)]+mapIndex cc env _ docId theUri+ = do+ let idxConfig = cc_IndexerConfig cc+ traceLevel = cc_TraceLevel cc+ fromTmp = (isJust $ ic_tempPath idxConfig)+ contextConfigs = (ic_contextConfigs idxConfig)+ attrs = (ic_readAttributes idxConfig)+ fileSystem = ae_FileSystem env+ computeOccurrences traceLevel fileSystem fromTmp contextConfigs attrs {- cache -} docId theUri+++reduceOccurrences + :: ActionEnvironment -> ()+ -> (Context, Word) -> [Occurrences]+ -> IO (Maybe Occurrences)+reduceOccurrences _ _ _ os+ = do+ let os' = (IM.unionsWith IS.union os)+ return $ Just $ os'+++indexerBuildIndexAction+ :: ActionConfiguration + () -- state+ () ((Context, Word),Occurrences) -- k1, v1+ () ((Context, Word),Occurrences) -- k2, v2+ ((Context, Word),Occurrences) -- v3 == v2+ Inverted -- v4+indexerBuildIndexAction+ = (defaultActionConfiguration "INDEX_BUILD")+ { ac_Map = Nothing+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ reduceAction+ = (defaultReduceConfiguration (reduceBuildIndex))+++reduceBuildIndex + :: ActionEnvironment -> ()+ -> () -> [((Context, Word),Occurrences)]+ -> IO (Maybe Inverted)+reduceBuildIndex _ _ _ os+ = do+ let i = emptyInverted+ let idx = foldl (\i' ((c,w),o) -> insertOccurrences c w o i') i os+ _ = rnf idx+ return $ Just $ idx +++-- -----------------------------------------------------------------------------++buildIndex :: (HolDocuments d a, {- HolIndex i -} MapReduce mr {-, HolCache c -}) + => MRCrawlerConfig d a+ -- Int -- ^ Number of parallel threads for MapReduce+ -- Int -- ^ TraceLevel for Arrows+ -> d a -- ^ List of input Data+ -- -> IndexerConfig -- ^ Configuration for the Indexing process+ -- -> Inverted -- ^ An empty HolIndex. This is used to determine which kind of index to use.+ -> mr+ -> IO [Inverted] -- ^ returns a HolIndex+buildIndex config {- workerThreads traceLevel -} docs {- idxConfig emptyIndex -} mr+ = do+ let docs' = (map (\(i,d) -> (i, uri d)) (IM.toList $ toMap docs))+ + runX (traceMsg 1 (" run indexer phase 1: "))+ + (res,_) <- doMapReduce (indexerOccurrencesAction config) () docs' [] 1 5 1 1 TOTRawTuple mr + + runX (traceMsg 1 (" result of phase 1: ")) + + runX (traceMsg 1 (" num of tuples: "))+ runX (traceMsg 1 (show $ length res))++ let os' = map (\t -> ((),t)) res++ runX (traceMsg 1 (" run indexer phase 2: "))+ + (res',_) <- doMapReduce (indexerBuildIndexAction) () os' [] 1 1 1 1 TOTRawTuple mr+ + runX (traceMsg 1 (" result of the indexer: ")) + + runX (traceMsg 1 (" num of indexes: "))+ runX (traceMsg 1 (show $ length res'))+ + let idx = map (snd) res'+ + return idx+++{-+buildIndexM :: (HolDocuments d a, HolIndexM m i, HolCache c) => + Int -- ^ Number of parallel threads for MapReduce+ -> Int -- ^ TraceLevel for Arrows+ -> d a -- ^ List of input Data+ -> IndexerConfig -- ^ Configuration for the Indexing process+ -> i -- ^ An empty HolIndexM. This is used to determine which kind of index to use.+ -> Maybe c -- ^ Just HolCache switches cache construction on+ -> IO i -- ^ returns a HolIndexM+buildIndexM workerThreads traceLevel docs idxConfig emptyIndex cache+ = let docs' = (map (\(i,d) -> (i, uri d)) (IM.toList $ toMap docs)) in+ -- assert ((sizeWords emptyIndex) == 0) + (mapReduce + workerThreads+-- (ic_indexerTimeOut idxConfig)+-- (( fromMaybe "/tmp/" (ic_tempPath idxConfig)) ++ "MapReduce.db")+ emptyIndex+ (computeOccurrences traceLevel + (isJust $ ic_tempPath idxConfig)+ (ic_contextConfigs idxConfig) + (ic_readAttributes idxConfig)+ cache+ )+ docs'+ )+-}+++-- | The MAP function in a MapReduce computation for building indexes.+-- The first three parameters have to be passed to the function to receive+-- a function with a valid MapReduce-map signature.+--+-- The function optionally outputs some debug information and then starts+-- the processing of a file by passing it together with the configuration+-- for different contexts to the @processDocument@ function where the file+-- is read and then the interesting parts configured in the+-- context configurations are extracted.++computeOccurrences :: -- HolCache c =>+ Int -> FS.FileSystem -> Bool -> [ContextConfig] -> Attributes -- -> Maybe c + -> DocId -> String -> IO [((Context, Word), Occurrences)]+computeOccurrences traceLevel fileSystem fromTmp contextConfigs attrs {- cache -} docId theUri+ = do+ clt <- getClockTime+ cat <- toCalendarTime clt+ res <- runX ( setTraceLevel traceLevel+ >>> traceMsg 1 ((calendarTimeToString cat) ++ " - indexing document: " + ++ show docId ++ " -> "+ ++ show theUri)+ >>> processDocument traceLevel fileSystem attrs' contextConfigs {- cache -} docId theUri+-- >>> arr (\ (c, w, d, p) -> (c, (w, d, p)))+ >>> strictA+ )+ return $ buildPositions res + where+ attrs' = if fromTmp then addEntries standardReadTmpDocumentAttributes attrs else attrs+ buildPositions :: [(Context, Word, DocId, Position)] -> [((Context, Word), Occurrences)]+ buildPositions l = M.foldWithKey (\(c,w,d) ps acc -> ((c,w),IM.singleton d ps) : acc) [] $+ foldl (\m (c,w,d,p) -> M.insertWith IS.union (c,w,d) (IS.singleton p) m) M.empty l + ++-- -----------------------------------------------------------------------------+ +-- | Downloads a document and calls the function to process the data for the+-- different contexts of the index+processDocument :: -- HolCache c => + Int+ -> FS.FileSystem+ -> Attributes+ -> [ContextConfig]+-- -> Maybe c+ -> DocId + -> URI+ -> IOSLA (XIOState s) b (Context, Word, DocId, Position)+processDocument traceLevel fileSystem attrs ccs {- cache -} docId theUri =+ withTraceLevel (traceLevel - traceOffset) + -- (readDocument attrs theUri) + (arrIO (\_ -> do+ c <- FS.getFileContent theUri fileSystem+ case c of+ (Just cs) -> return $ decode cs+ (Nothing) -> return "" -- TODO mark error here+ )+ ) + >>> (readFromString attrs)+ >>> (catA $ map (processContext {- cache -} docId) ccs ) -- process all context configurations + +++-- | Process a Context. Applies the given context configuration to extract information from+-- the XmlTree that is passed in the arrow.+processContext :: +-- ( HolCache c) => +-- Maybe c+ DocId+ -> ContextConfig+ -> IOSLA (XIOState s) XmlTree (Context, Word, DocId, Position)++processContext {-cache-} docId cc+ = cc_preFilter cc -- convert XmlTree+ >>>+ fromLA extractWords+ >>>+{- ( if ( isJust cache+ &&+ cc_addToCache cc+ )+ then perform (arrIO storeInCache)+ else this+ )+ >>>+-} arrL genWordList+ >>>+ strictA+ where+ extractWords :: LA XmlTree [String]+ extractWords+ = listA+ ( xshow ( (cc_fExtract cc) -- extract interesting parts+ >>>+ getTexts+ )+ >>>+ arrL ( filter (not . null) . cc_fTokenize cc )+ )++ genWordList :: [String] -> [(Context, Word, DocId, Position)]+ genWordList+ = zip [1..] -- number words+ >>> -- arrow for pure functions+ filter (not . (cc_fIsStopWord cc) . snd) -- delete boring words+ >>>+ map ( \ (p, w) -> (cc_name cc, w, docId, p) ) -- attach context and docId+{-+ storeInCache s+ = let t = unwords s in + if t /= "" then putDocText (fromJust cache) (cc_name cc) docId t+ else return()+-}++getTexts :: LA XmlTree XmlTree+getTexts -- select all text nodes+ = choiceA+ [ isElem :-> ( space -- substitute tags by a single space+ <+> -- so tags act as word delimiter+ (getChildren >>> getTexts)+ <+>+ space+ ) -- tags are interpreted as word delimiter+ , isText :-> this -- take the text nodes+ , this :-> none -- ignore anything else+ ]+ where+ space = txt " "+ +
+ Examples/MapReduce/Crawler/Makefile view
@@ -0,0 +1,33 @@+# Making Distribution-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = CrawlerClient CrawlerWorker++OUTPUT = output++STORAGE = storage++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT) $(STORAGE)+
+ Examples/MapReduce/DummyFiles/ClientMain.hs view
@@ -0,0 +1,55 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Control.Exception++import Holumbus.Common.Logging+import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI ++ +version :: String+version = "Holumbus-Distribution Standalone-Client 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handle (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p + d <- initializeData+ UI.runUI d version + deinitializeData d+++initializeData :: IO (MR.DMapReduce)+initializeData + = do+ let config = MR.defaultMRClientConfig+ MR.mkMapReduceClient config+++deinitializeData :: MR.DMapReduce -> IO ()+deinitializeData mr+ = do+ MR.closeMapReduce mr+
+ Examples/MapReduce/DummyFiles/WorkerMain.hs view
@@ -0,0 +1,64 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Control.Exception++import Holumbus.Common.Logging+import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++++version :: String+version = "Holumbus-Distribution Standalone-Worker 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handle (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version + deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do+ fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig+ + -- insert your own actions here + let actions = undefined+ let config = MR.defaultMRWorkerConfig+ + mr <- MR.mkMapReduceWorker fs actions config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs+
+ Examples/MapReduce/Grep/ClientMain.hs view
@@ -0,0 +1,64 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Distribution.DMapReduce as MR+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce ++import Examples.MapReduce.Grep.Grep++ +version :: String+version = "Holumbus-Distribution Standalone-Client 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p + mr <- initializeData+ + (ls,_) <- doMapReduce (grepAction) "Simpson" namesList [] 1 5 1 1 TOTRawTuple mr+ + putStrLn "Result: "+ putStrLn $ show ls+ + deinitializeData mr+++initializeData :: IO (MR.DMapReduce)+initializeData + = do+ let config = MR.defaultMRClientConfig+ MR.mkMapReduceClient config+++deinitializeData :: MR.DMapReduce -> IO ()+deinitializeData mr+ = do+ MR.closeMapReduce mr+ +-- ----------------------------------------------------------------------------
+ Examples/MapReduce/Grep/Grep.hs view
@@ -0,0 +1,210 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Examples.MapReduce.Grep.Grep+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Examples.MapReduce.Grep.Grep+(+ grepAction+, grepActionMap+, namesList+, grepDemoJob+, createGrepDemoFiles+)+where++import Data.Binary+import Data.Maybe++import System.Log.Logger++import Text.Regex++import qualified Holumbus.FileSystem.FileSystem as FS++import qualified Holumbus.Data.KeyMap as KMap+import Holumbus.MapReduce.Types+++localLogger :: String+localLogger = "Examples.MapReduce.Grep.Grep"+++-- ----------------------------------------------------------------------------+-- Distributed Grep+-- ----------------------------------------------------------------------------++-- grep for the namelist example+mapGrep :: ActionEnvironment -> String -> () -> String -> IO [((), String)]+mapGrep _ a _ v+ = do + infoM localLogger "mapGrep"+ debugM localLogger $ "input: a: " ++ show a ++ " - Text: " ++ show v+ res <- if (matches a v)+ then return [((),v)]+ else return []+ debugM localLogger $ "output: " ++ show res+ return res+ where+ matches e n = isJust $ matchRegex (mkRegex e) n++{-+-- grep for (Filename,Text)+mapGrep :: ActionEnvironment -> String -> String -> String -> IO [((),(String,String))]+mapGrep _ e k v+ = do+ return $ mapMaybe (getMatch) $ lines v+ where+ getMatch l + = if (isJust $ matchRegex (mkRegex e) l)+ then Just ((),(k,l))+ else Nothing +-} ++-- reduce for the namelist example+reduceGrep :: ActionEnvironment -> String -> () -> [String] -> IO (Maybe [String])+reduceGrep _ _ _ vs + = do+ infoM localLogger "reduce/combine Grep"+ debugM localLogger $ "input: " ++ show vs+ return (Just vs)++{-+-- reduce for (Filename,Text)+reduceGrep+ :: ActionEnvironment -> String -> () -> [(String,String)] -> IO (Maybe [(String,String)])+reduceGrep _ _ _ vs + = do+ infoM localLogger "reduce/combine Grep"+ debugM localLogger $ "input: " ++ show vs+ return (Just vs)+-}++-- ----------------------------------------------------------------------------+-- Actions+-- ----------------------------------------------------------------------------++-- Action for the namelist example+grepAction+ :: ActionConfiguration + String -- state+ () String -- k1, v1+ () String -- k2, v2+ String -- v3 == v2+ [String] -- v4+grepAction+ = (defaultActionConfiguration "GREP")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration mapGrep)+ reduceAction+ = (defaultReduceConfiguration reduceGrep)++{-+-- Action for (Filename,Text)+grepAction+ :: ActionConfiguration + String -- state+ String String -- k1, v1+ () (String,String) -- k2, v2+ (String, String) -- v3 == v2+ [(String, String)] -- v4+grepAction+ = (defaultActionConfiguration "GREP")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration mapGrep)+ reduceAction+ = (defaultReduceConfiguration reduceGrep)+-}+++grepActionMap :: ActionMap+grepActionMap+ = KMap.insert (readActionConfiguration grepAction) $+ KMap.empty+++-- ----------------------------------------------------------------------------+-- DemoJob+-- ----------------------------------------------------------------------------++ +grepDemoJob :: JobInfo+grepDemoJob = + createJobInfoFromConfiguration+ grepAction -- action config+ "Simpson" -- options+ namesList -- input (Tuples)+ [] -- input (Files)+ 1 -- number of splitters+ 2 -- number of mappers+ 1 -- number of reducers+ 1 -- number of results+ TOTRawTuple -- type of the result (file of raw)+ +namesList :: [((),String)]+namesList+ = map (\d -> ((),d)) names+ where+ names =+ [("Abraham Simpson"),+ ("Apu Nahasapeemapetilon"),+ ("Waylon Smithers"),+ ("Barney Gumble"),+ ("Bart Simpson"),+ ("Carl Carlson"),+ ("Homer Simpson"),+ ("Jacqueline Bouvier"),+ ("Kirk van Houten"),+ ("Lenny Leonard"),+ ("Lionel Hutz"),+ ("Lisa Simpson"),+ ("Luann van Houten"),+ ("Maggie Simpson"),+ ("Manjula Nahasapeemapetilon"),+ ("Marge Simpson"),+ ("Martin Prince"),+ ("Maude Flanders"),+ ("Milhouse van Houten"),+ ("Moe Syzslak"),+ ("Montgomery Burns"),+ ("Ned Flanders"),+ ("Patty Bouvier"),+ ("Rod Flanders"),+ ("Selma Bouvier"),+ ("Seymour Skinner"),+ ("Todd Flanders"),+ ("Troy McClure")]+ ++-- ----------------------------------------------------------------------------+-- DemoFiles+-- ----------------------------------------------------------------------------+++createGrepDemoFiles :: FS.FileSystem -> IO ()+createGrepDemoFiles fs+ = do+ -- let c = S.BinaryFile (encode ("foo","a aa aaa b bb bbb"))+ -- let c = S.TextFile "harddisk file"+ let c = encode namesList+ FS.createFile "namesList.txt" c fs
+ Examples/MapReduce/Grep/Makefile view
@@ -0,0 +1,33 @@+# Making Distribution-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = WorkerMain ClientMain++OUTPUT = output++STORAGE = storage++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT) $(STORAGE)+
+ Examples/MapReduce/Grep/WorkerMain.hs view
@@ -0,0 +1,64 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++import Examples.MapReduce.Grep.Grep++version :: String+version = "Holumbus-Distribution Standalone-Worker 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version+ deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do+ fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig+ + let actions = grepActionMap+ let config = MR.defaultMRWorkerConfig+ + mr <- MR.mkMapReduceWorker fs actions config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs+ +-- ----------------------------------------------------------------------------
+ Examples/MapReduce/Makefile view
@@ -0,0 +1,18 @@+# Making all examples++ALLEXAMPLES = Crawler Grep Sort WordFrequency+STANDALONE = Standalone+++all :+ $(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i all ;)++standalone : + $(foreach i,$(STANDALONE),$(MAKE) -C $i all ;)+ +wc :+ @wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+ $(foreach i,$(ALLEXAMPLES),$(MAKE) -C $i $@ ;)+ $(foreach i,$(STANDALONE),$(MAKE) -C $i $@ ;)
+ Examples/MapReduce/Sort/ClientMain.hs view
@@ -0,0 +1,62 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Distribution.DMapReduce as MR +import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce ++import Examples.MapReduce.Sort.Sort++version :: String+version = "Holumbus-Distribution Standalone-Client 0.1"++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p + mr <- initializeData+ + (ls,_) <- doMapReduce (sortAction) () namesList [] 1 5 1 1 TOTRawTuple mr+ + putStrLn "Result: "+ putStrLn $ show ls+ + deinitializeData mr+++initializeData :: IO (MR.DMapReduce)+initializeData + = do+ let config = MR.defaultMRClientConfig+ MR.mkMapReduceClient config+++deinitializeData :: MR.DMapReduce -> IO ()+deinitializeData mr+ = do+ MR.closeMapReduce mr+ +-- ----------------------------------------------------------------------------
+ Examples/MapReduce/Sort/Makefile view
@@ -0,0 +1,33 @@+# Making Distribution-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = WorkerMain ClientMain++OUTPUT = output++STORAGE = storage++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT) $(STORAGE)+
+ Examples/MapReduce/Sort/Sort.hs view
@@ -0,0 +1,157 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Examples.MapReduce.Sort.Sort+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Examples.MapReduce.Sort.Sort+(+ sortAction+, sortActionMap+, namesList+, sortDemoJob+, createSortDemoFiles+)+where++import Data.Binary++import System.Log.Logger++import qualified Holumbus.FileSystem.FileSystem as FS++import qualified Holumbus.Data.KeyMap as KMap+import Holumbus.MapReduce.Types+++localLogger :: String+localLogger = "Examples.MapReduce.Sort.Sort"+++-- ----------------------------------------------------------------------------+-- Distributed Sort+-- ----------------------------------------------------------------------------+++mapSort :: ActionEnvironment -> () -> () -> (String, String) -> IO [(String, (String, String))]+mapSort _ _ _ v+ = do + infoM localLogger "mapSort"+ debugM localLogger $ "input: " ++ show v+ let res = [(snd v, v)]+ debugM localLogger $ "output: " ++ show res+ return res+++reduceSort :: ActionEnvironment -> () -> String -> [(String, String)] -> IO (Maybe [(String, String)])+reduceSort _ _ k vs + = do+ infoM localLogger "reduce/combine Sort"+ debugM localLogger $ "input: " ++ k ++ " - " ++ show vs+ return (Just vs)+++-- ----------------------------------------------------------------------------+-- Actions+-- ----------------------------------------------------------------------------++sortAction+ :: ActionConfiguration + () -- state+ () (String, String) -- k1, v1+ String (String, String) -- k2, v2+ (String, String) -- v3 == v2+ [(String, String)] -- v4+sortAction+ = (defaultActionConfiguration "SORT")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration mapSort)+ reduceAction+ = (defaultReduceConfiguration reduceSort)+ ++sortActionMap :: ActionMap+sortActionMap+ = KMap.insert (readActionConfiguration sortAction) $+ KMap.empty+++-- ----------------------------------------------------------------------------+-- DemoJob+-- ----------------------------------------------------------------------------++ +sortDemoJob :: JobInfo+sortDemoJob = + createJobInfoFromConfiguration+ sortAction -- action config+ () -- options+ namesList -- input (Tuples)+ [] -- input (Files)+ 1 -- number of splitters+ 2 -- number of mappers+ 1 -- number of reducers+ 1 -- number of results+ TOTRawTuple -- type of the result (file of raw)+ +namesList :: [((),(String, String))]+namesList+ = map (\d -> ((),d)) names+ where+ names =+ [("Abraham","Simpson"),+ ("Apu","Nahasapeemapetilon"),+ ("Waylon","Smithers"),+ ("Barney","Gumble"),+ ("Bart","Simpson"),+ ("Carl","Carlson"),+ ("Homer","Simpson"),+ ("Jacqueline","Bouvier"),+ ("Kirk","van Houten"),+ ("Lenny","Leonard"),+ ("Lionel","Hutz"),+ ("Lisa","Simpson"),+ ("Luann","van Houten"),+ ("Maggie","Simpson"),+ ("Manjula","Nahasapeemapetilon"),+ ("Marge","Simpson"),+ ("Martin","Prince"),+ ("Maude","Flanders"),+ ("Milhouse","van Houten"),+ ("Moe","Syzslak"),+ ("Montgomery","Burns"),+ ("Ned","Flanders"),+ ("Patty","Bouvier"),+ ("Rod","Flanders"),+ ("Selma","Bouvier"),+ ("Seymour","Skinner"),+ ("Todd","Flanders"),+ ("Troy","McClure")]+ ++-- ----------------------------------------------------------------------------+-- DemoFiles+-- ----------------------------------------------------------------------------+++createSortDemoFiles :: FS.FileSystem -> IO ()+createSortDemoFiles fs+ = do+ -- let c = S.BinaryFile (encode ("foo","a aa aaa b bb bbb"))+ -- let c = S.TextFile "harddisk file"+ let c = encode namesList+ FS.createFile "namesList.txt" c fs
+ Examples/MapReduce/Sort/WorkerMain.hs view
@@ -0,0 +1,64 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++import Examples.MapReduce.Sort.Sort++version :: String+version = "Holumbus-Distribution Standalone-Worker 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version + deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do+ fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig+ + let actions = sortActionMap+ let config = MR.defaultMRWorkerConfig+ + mr <- MR.mkMapReduceWorker fs actions config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs+ +-- ----------------------------------------------------------------------------
+ Examples/MapReduce/Standalone/Makefile view
@@ -0,0 +1,32 @@+# Making FileSystem-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = StandaloneMain++OUTPUT = output++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ rm -rf storage+ $(RM) $(PROG) $(OUTPUT) *.port+
+ Examples/MapReduce/Standalone/StandaloneMain.hs view
@@ -0,0 +1,56 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Control.Exception++import qualified Holumbus.Common.Logging as LOG+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Standalone.SMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++import Examples.MapReduce.WordFrequency.WordFrequency++version :: String+version = "Holumbus-Standalone MapReducer 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handle (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ LOG.initializeLogging []+ sa <- initializeData+ UI.runUI sa version+ deinitializeData sa+++initializeData :: IO MR.SMapReduce+initializeData+ = do+ fs <- FS.mkStandaloneFileSystem FS.defaultFSStandaloneConfig+ createWordFrequencyDemoFiles fs+ let actions = wordFrequencyActionMap+ let config = MR.defaultStandaloneConfig+ MR.newSMapReduce fs actions config+++deinitializeData :: MR.SMapReduce -> IO ()+deinitializeData sa+ = do+ MR.closeMapReduce sa
+ Examples/MapReduce/WordFrequency/ClientMain.hs view
@@ -0,0 +1,63 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.Distribution.DMapReduce as MR+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce ++import Examples.MapReduce.WordFrequency.WordFrequency+ +version :: String+version = "Holumbus-Distribution Standalone-Client 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p + mr <- initializeData++ (ls,_) <- doMapReduce (wordFrequencyAction) () textList [] 1 5 1 1 TOTRawTuple mr+ + putStrLn "Result: "+ putStrLn $ show ls+ + deinitializeData mr+++initializeData :: IO (MR.DMapReduce)+initializeData + = do+ let config = MR.defaultMRClientConfig+ MR.mkMapReduceClient config+++deinitializeData :: MR.DMapReduce -> IO ()+deinitializeData mr+ = do+ MR.closeMapReduce mr+ +-- ----------------------------------------------------------------------------
+ Examples/MapReduce/WordFrequency/Makefile view
@@ -0,0 +1,35 @@+# Making Distribution-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = WorkerMain ClientMain++OUTPUT = output++STORAGE = storage++all : $(PROG)++standalone : WFStandalone++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT) $(STORAGE)+
+ Examples/MapReduce/WordFrequency/WordFrequency.hs view
@@ -0,0 +1,142 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Examples.MapReduce.WordFrequency.WordFrequency+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Examples.MapReduce.WordFrequency.WordFrequency+(+ wordFrequencyAction+, wordFrequencyActionMap+, wordFrequencyDemoJob+, createWordFrequencyDemoFiles+, textList+)+where++import Data.Binary++import System.Log.Logger++import qualified Holumbus.FileSystem.FileSystem as FS++import qualified Holumbus.Data.AccuMap as AMap+import qualified Holumbus.Data.KeyMap as KMap+import Holumbus.MapReduce.Types+++localLogger :: String+localLogger = "Examples.MapReduce.WordFrequency.WordFrequency"+++-- ----------------------------------------------------------------------------+-- Word-Frequency+-- ----------------------------------------------------------------------------+++mapWordFrequency :: ActionEnvironment -> () -> String -> String -> IO [(String, Integer)]+mapWordFrequency _ _ k v+ = do + infoM localLogger "mapWordFrequency"+ debugM localLogger $ show ("input: " ++ k ++ " - " ++ show v)+ let v' = map (\s -> (s,1)) $ words v+ debugM localLogger $ show $ "output: " ++ show v'+ return v'+++reduceWordFrequency :: ActionEnvironment -> () -> String -> [Integer] -> IO (Maybe Integer)+reduceWordFrequency _ _ k vs + = do+ infoM localLogger "reduce/combine WordFrequency"+ debugM localLogger $ show ("input: " ++ k ++ " - " ++ show vs)+ let s = sum vs+ debugM localLogger $ show $ "output: " ++ show s+ return (Just s)+ + +partitionWordFrequency :: ActionEnvironment -> () -> Int -> [(String, Integer)] -> IO [(Int,[(String, Integer)])]+partitionWordFrequency _ _ _ ls + = do+ infoM localLogger "partitionCountWords"+ debugM localLogger $ show ls+ -- calculate partition-Values+ let markedList = map (\t@(k,_) -> (length k,t)) ls+ -- merge them+ let resultList = AMap.toList $ AMap.fromTupleList markedList+ return resultList+++-- ----------------------------------------------------------------------------+-- Actions+-- ----------------------------------------------------------------------------++wordFrequencyAction+ :: ActionConfiguration + () -- state+ String String -- k1, v1+ String Integer -- k2, v2+ Integer -- v3 == v2+ Integer -- v4+wordFrequencyAction+ = (defaultActionConfiguration "WORDFREQUENCY")+ { ac_Map = Just mapAction+ , ac_Combine = Nothing+ , ac_Reduce = Just reduceAction+ }+ where+ mapAction + = (defaultMapConfiguration mapWordFrequency)+ { mc_Partition = partitionWordFrequency }+ reduceAction+ = (defaultReduceConfiguration reduceWordFrequency)+ { rc_Partition = partitionWordFrequency }++wordFrequencyActionMap :: ActionMap+wordFrequencyActionMap+ = KMap.insert (readActionConfiguration wordFrequencyAction) $+ KMap.empty+++-- ----------------------------------------------------------------------------+-- DemoJob+-- ----------------------------------------------------------------------------++ +wordFrequencyDemoJob :: JobInfo+wordFrequencyDemoJob = + createJobInfoFromConfiguration+ wordFrequencyAction -- action config+ () -- options+ textList -- input (Tuples)+ [] -- input (Files)+ 1 -- number of splitters+ 2 -- number of mappers+ 1 -- number of reducers+ 1 -- number of results+ TOTRawTuple -- type of the result (file of raw)+ + +textList :: [(String,String)]+textList+ = [("text1", "aaa bb c dd dd"),("text2", "aaa bb"),("text2", "aaa dd dd")]+++-- ----------------------------------------------------------------------------+-- DemoFiles+-- ----------------------------------------------------------------------------+++createWordFrequencyDemoFiles :: FS.FileSystem -> IO ()+createWordFrequencyDemoFiles fs+ = do+ let c = encode "harddisk file"+ FS.createFile "file1.txt" c fs
+ Examples/MapReduce/WordFrequency/WorkerMain.hs view
@@ -0,0 +1,62 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI++import Examples.MapReduce.WordFrequency.WordFrequency+++version :: String+version = "Holumbus-Distribution Standalone-Worker 0.1"++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version + deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do+ fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig+ let actions = wordFrequencyActionMap+ let config = MR.defaultMRWorkerConfig+ mr <- MR.mkMapReduceWorker fs actions config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs+ +-- ----------------------------------------------------------------------------
+ Examples/Utils/PickleTest/Makefile view
@@ -0,0 +1,31 @@+# Making FileSystem-Standalone Programs++HOME = ../../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT) -i../../../.+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = PickleTest++OUTPUT = output++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT) *.xml+
+ Examples/Utils/PickleTest/PickleTest.hs view
@@ -0,0 +1,35 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.FileHandling+import Holumbus.MapReduce.Types++import Examples.MapReduce.WordFrequency.WordFrequency++version :: String+version = "Pickle-Test 0.1"++main :: IO ()+main+ = do+ putStrLn version+ putStrLn "writing: job.xml"+ saveToXmlFile "job.xml" wordFrequencyDemoJob+ putStrLn "reading: job.xml"+ ji <- (loadFromXmlFile "job.xml")::IO JobInfo+ putStrLn $ show ji+ putStrLn $ "comparing: " ++ show (ji == wordFrequencyDemoJob)
+ Holumbus-MapReduce.cabal view
@@ -0,0 +1,89 @@+name: Holumbus-MapReduce+version: 0.0.1+license: OtherLicense+license-file: LICENSE+author: Uwe Schmidt, Stefan Schmidt+copyright: Copyright (c) 2008 Uwe Schmidt, Stefan Schmidt+maintainer: Stefan Schmidt <sts@holumbus.org>+stability: experimental+category: Distributed Computing+synopsis: a distributed MapReduce framework+homepage: http://holumbus.fh-wedel.de+description: This package contains a library and tools for the+ creation of distributed MapReduce applications.+build-type: Simple+cabal-version: >=1.6++extra-source-files:+ README+ Makefile+ Programs/Makefile+ Programs/MapReduceMaster/Makefile+ Programs/MapReduceMaster/Master.hs+ Examples/MapReduce/Crawler/Config.hs+ Examples/MapReduce/Crawler/Crawl.hs+ Examples/MapReduce/Crawler/Index.hs+ Examples/MapReduce/Crawler/CrawlerClient.hs+ Examples/MapReduce/Crawler/CrawlerWorker.hs+ Examples/MapReduce/Crawler/Makefile+ Examples/MapReduce/DummyFiles/ClientMain.hs+ Examples/MapReduce/DummyFiles/WorkerMain.hs+ Examples/MapReduce/Standalone/Makefile+ Examples/MapReduce/Standalone/StandaloneMain.hs+ Examples/MapReduce/Makefile+ Examples/MapReduce/WordFrequency/WordFrequency.hs+ Examples/MapReduce/WordFrequency/ClientMain.hs+ Examples/MapReduce/WordFrequency/Makefile+ Examples/MapReduce/WordFrequency/WorkerMain.hs+ Examples/MapReduce/Grep/ClientMain.hs+ Examples/MapReduce/Grep/Grep.hs+ Examples/MapReduce/Grep/Makefile+ Examples/MapReduce/Grep/WorkerMain.hs+ Examples/MapReduce/Sort/ClientMain.hs+ Examples/MapReduce/Sort/Makefile+ Examples/MapReduce/Sort/Sort.hs+ Examples/MapReduce/Sort/WorkerMain.hs+ Examples/Makefile+ Examples/Utils/PickleTest/Makefile+ Examples/Utils/PickleTest/PickleTest.hs++library+ build-depends: base >= 3+ , binary >= 0.4+ , bytestring >= 0.9+ , containers >= 0.1+ , directory >= 1.0+ , haskell98 >= 1+ , hslogger >= 1.0+ , hxt >= 8.2+ , network >= 2.1+ , time >= 1.1+ , unix >= 2.3+ , Holumbus-Distribution >= 0.0.1+ , Holumbus-Storage >= 0.0.1++ hs-source-dirs: + source+ exposed-modules: + Holumbus.Distribution.DMapReduce+ , Holumbus.Distribution.Messages+ , Holumbus.Distribution.Worker+ , Holumbus.Distribution.Worker.WorkerData+ , Holumbus.Distribution.Worker.WorkerPort+ , Holumbus.Distribution.Master+ , Holumbus.Distribution.Master.MasterData+ , Holumbus.Distribution.Master.MasterPort+ , Holumbus.Standalone.SMapReduce+ , Holumbus.MapReduce.MapReduce+ , Holumbus.MapReduce.JobController+ , Holumbus.MapReduce.TaskProcessor+ , Holumbus.MapReduce.Types+ , Holumbus.MapReduce.UserInterface+ ghc-options: -Wall -threaded+ extensions: Arrows DeriveDataTypeable ExistentialQuantification++executable Master+ main-is: Master.hs+ hs-source-dirs: Programs/MapReduceMaster source+ ghc-options: -Wall -threaded+ extensions: Arrows DeriveDataTypeable ExistentialQuantification
+ LICENSE view
@@ -0,0 +1,21 @@+The MIT License++Copyright (c) 2008 Stefan Schmidt++Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation the+rights to use, copy, modify, merge, publish, distribute, sublicense, and/or +sell copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE.
+ Makefile view
@@ -0,0 +1,58 @@+# Making all+#+# Building the library and generating documentation is done by cabal,+# only the tests and the examples will be built using make.++# TEST_BASE = test+EXAMPLES_BASE = Examples+PROGRAMS_BASE = Programs+++all : configure build doc++configure :+ @runhaskell Setup.hs configure++doc : configure+ @runhaskell Setup.hs haddock --hyperlink-source --hscolour-css=hscolour.css++build : configure+ @runhaskell Setup.hs build++install :+ @runhaskell Setup.hs install --global++prof :+ @runhaskell Setup.hs configure -p+ @runhaskell Setup.hs build++dist :+ @runhaskell Setup.hs sdist++# alltests :+# $(MAKE) -C $(TEST_BASE) all++examples :+ $(MAKE) -C $(EXAMPLES_BASE) all++programs :+ $(MAKE) -C $(PROGRAMS_BASE) all++wc :+ @wc -l `find ./source/Holumbus -wholename './_darcs/*' -prune -o -name "*.hs" -print` + # $(MAKE) -C $(TEST_BASE) wc+ $(MAKE) -C $(EXAMPLES_BASE) wc+ $(MAKE) -C $(PROGRAMS_BASE) wc++zip: clean+ @zip -r holumbus_mapreduce$$(date +%Y%m%d_%H%M).zip *++clean :+ @runhaskell Setup.hs clean+ @rm -rf holumbus_mapreduce*.zip+ # $(MAKE) -C $(TEST_BASE) clean+ $(MAKE) -C $(EXAMPLES_BASE) clean+ $(MAKE) -C $(PROGRAMS_BASE) clean++.PHONY : all configure doc build install prof dist examples programs wc zip clean+
+ Programs/Makefile view
@@ -0,0 +1,17 @@+# Making all examples++MASTER = MapReduceMaster+ALLPROGRAMS = $(MASTER)+++all :+ $(foreach i,$(ALLPROGRAMS),$(MAKE) -C $i all ;)++master :+ $(foreach i,$(MASTER),$(MAKE) -C $i all ;)++wc :+ @wc -l `find . -wholename './_darcs/*' -prune -o -name "*.hs" -print`++clean :+ $(foreach i,$(ALLPROGRAMS),$(MAKE) -C $i $@ ;)
+ Programs/MapReduceMaster/Makefile view
@@ -0,0 +1,30 @@++HOME = ../..++SOURCE = .++GHC_FLAGS = -Wall -threaded -O2 -hidir $(OUTPUT) -odir $(OUTPUT)+GHC = ghc $(GHC_FLAGS)++RM_FLAGS = -rf+RM = rm $(RM_FLAGS)++PROG = Master++OUTPUT = output++all : $(PROG)++#prof : $(PROG).hs+# [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+# $(RM) $(OUTPUT)/*+# $(GHC) -prof -auto-all -ignore-package Holumbus -i../../source/ --make -o $(PROG)_p $(PROG).hs++% : %.hs+ [ -d $(OUTPUT) ] || mkdir -p $(OUTPUT)+ $(RM) $(OUTPUT)/*+ $(GHC) --make -o $@ $<++clean :+ $(RM) $(PROG) $(OUTPUT)+
+ Programs/MapReduceMaster/Master.hs view
@@ -0,0 +1,60 @@+-- ----------------------------------------------------------------------------+{- |+ Module : + Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Main(main) where++import Holumbus.Common.Logging+import Holumbus.Common.Utils ( handleAll )++import Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Distribution.DMapReduce as MR+import qualified Holumbus.MapReduce.UserInterface as UI+++version :: String+version = "Holumbus-Distribution Standalone-Master 0.1"+++main :: IO ()+main+ = do+ putStrLn version+ handleAll (\e -> putStrLn $ "EXCEPTION: " ++ show e) $+ do+ initializeLogging []+ p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+ setPortRegistry p+ (mr,fs) <- initializeData+ UI.runUI mr version + deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData + = do + fs <- FS.mkFileSystemController FS.defaultFSControllerConfig+ let config = MR.defaultMRMasterConfig+ mr <- MR.mkMapReduceMaster fs config+ return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+ = do+ MR.closeMapReduce mr+ FS.closeFileSystem fs++-- ----------------------------------------------------------------------------
+ README view
@@ -0,0 +1,119 @@+This is the Holumbus-MapReduce Framework++Version 0.0.1++Stefan Schmidt sts@holumbus.org++http://holumbus.fh-wedel.de+++About+-----++Holumbus is a set of Haskell libraries. This package contains the +Holumbus-MapReduce library for building and running distributed MapReduce+systems.++This library depends on the Holumbus-Distributed and Holumbus-Storage libraries. +If you want to run some of the examples, e.g. the distributed Web-Crawler +and Indexer, the the Holumbus-Searchengine library must also be installed.+++Contents+--------++Examples Some example applications and utilities.+Programs The applications you need to run a distributed MapReduce system.+source Source code of the Holumbus-MapReduce library.+++Requirements+------------++So far, this library is only tested under Linux, please let us know, if there are +any problems under Windows or other OSes.+The Holumbus-MapReduce library requires at least GHC 6.10 and the +following packages (available via Hackage).++ containers+ hslogger+ directory+ network+ time+ bytestring+ binary+ hxt+ Holumbus-Distribution+++Installation+------------++A Cabal file is provided, therefore Holumbus-MapReduce can be installed using+the standard Cabal way:++$ runhaskell Setup.hs configure+$ runhaskell Setup.hs build+$ runhaskell Setup.hs install # with root privileges++If you prefer to do it the old way there with make:++$ make build+$ make install # with root privileges+++Steps to make the system running+--------------------------------++1. Compile and install the Holumbus-Distribution library.++2. Compile and install the Holumbus-MapReduce framework.+ If this is done with cabal, step 3. can be skipped, the+ master will be build and installed with cabal++$ make build+$ make install # this with root privileges++3. Compile the Master-Program.+ It is located in Programs/++$ make programs++4. Start the PortRegistry from the Holumbus-Distribution library.++5. Start the Master.+ If you want to start the master on another machine than the PortRegistry,+ then you have to copy the file "registry.xml" from the tmp directory to the+ the tmp directory on the machine the master should run on.++$ cd Programs/Master+$ ./Master++ Alternatively start the Master program, installed with cabal++Now, the system itself is running, but as you might wonder, you only have +started the registry for communication and the master for the MapReduce system.+To run your own system, you have to create a MapReduce Client and a MapReduce+Worker. Here, we'll show you to run the examples, they'll show you how to+create you own system.++6. Compile the Examples. + For some examples (e.g. the Crawler), this requires that you have installed + the Holumbus-Searchengine framework. You can get it from the Holumbus + homepage (http://holumbus.fh-wedel.de), please read its documentation for further+ instructions and details. You'll only need the Searchegine library installed+ for the MapReduce examples, not the library itself.++$ make examples # some examples require the Searchengine lib++7. Start the Workers.+ At this time every worker has to run in its own directory, so create for+ each worker, you want to run, its own subdirectory and copy the executeable+ to it. If you want to run the workers on different machines, you have to+ copy the file "registry.xml" from your tmp directory to the tmp directory+ of the machine the worker will work on.++8. Start the Client.+ Now you can start the client and wait the things to come. If you want to run+ the client on another machine than the PortRegistry, you have to copy the+ registry.xml file from the tmp directory, too.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ source/Holumbus/Distribution/DMapReduce.hs view
@@ -0,0 +1,233 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.DMapReduce+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.Distribution.DMapReduce+(+-- * Datatypes+ DMapReduce+, MapReduce(..)++-- * Configuration+, DMRMasterConf(..)+, defaultMRMasterConfig+, DMRWorkerConf(..)+, defaultMRWorkerConfig+, DMRClientConf(..)+, defaultMRClientConfig+++-- * Creation and Destruction+, mkMapReduceMaster+, mkMapReduceWorker+, mkMapReduceClient+)+where++import Control.Concurrent+import Network+import System.Log.Logger++import Holumbus.Common.Debug+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.MapReduce+import qualified Holumbus.Distribution.Master as M+import qualified Holumbus.Distribution.Master.MasterData as MD+import qualified Holumbus.Distribution.Master.MasterPort as MP+import qualified Holumbus.Distribution.Worker as W+import qualified Holumbus.Distribution.Worker.WorkerData as WD+import qualified Holumbus.Distribution.Worker.WorkerPort as WP+import qualified Holumbus.FileSystem.FileSystem as FS+import Holumbus.Network.Site+import Holumbus.Network.Port+++localLogger :: String+localLogger = "Holumbus.Distribution.DMapReduce"++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+++data DMapReduceData = + forall m w. (M.MasterClass m, W.WorkerClass w, Debug m, Debug w) =>+ DMapReduceData SiteId MapReduceType m (Maybe w)+++data DMapReduce = DMapReduce (MVar DMapReduceData)++instance Show DMapReduce where+ show _ = "DMapReduce"++++-- ---------------------------------------------------------------------------+-- Configurations+-- ---------------------------------------------------------------------------+++data DMRMasterConf = DMRMasterConf {+ msc_StartControlling :: Bool+ , msc_StreamName :: StreamName+ , msc_PortNumber :: Maybe PortNumber+ } ++defaultMRMasterConfig :: DMRMasterConf+defaultMRMasterConfig = DMRMasterConf True "MRMaster" Nothing+++data DMRWorkerConf = DMRWorkerConf {+ woc_StreamName :: StreamName+ , woc_SocketId :: Maybe SocketId+ } ++defaultMRWorkerConfig :: DMRWorkerConf+defaultMRWorkerConfig = DMRWorkerConf "MRMaster" Nothing+++data DMRClientConf = DMRClientConf {+ clc_StreamName :: StreamName+ , clc_SocketId :: Maybe SocketId+ } ++defaultMRClientConfig :: DMRClientConf+defaultMRClientConfig = DMRClientConf "MRMaster" Nothing+++-- ---------------------------------------------------------------------------+-- Creation and Destruction+-- ---------------------------------------------------------------------------+++mkMapReduceMaster + :: FS.FileSystem -> DMRMasterConf+ -> IO DMapReduce+mkMapReduceMaster fs conf+ = do+ sid <- getSiteId+ infoM localLogger $ "initialising master on site " ++ show sid + md <- MD.newMaster fs (msc_StartControlling conf) (msc_StreamName conf) (msc_PortNumber conf)+ newDMapReduce MRTMaster md (Nothing::Maybe WP.WorkerPort)+++mkMapReduceWorker+ :: FS.FileSystem -> ActionMap -> DMRWorkerConf+ -> IO DMapReduce+mkMapReduceWorker fs am conf+ = do+ sid <- getSiteId+ infoM localLogger $ "initialising worker on site " ++ show sid + mp <- MP.newMasterPort (woc_StreamName conf) (woc_SocketId conf)+ wd <- WD.newWorker fs am (woc_StreamName conf) (woc_SocketId conf)+ newDMapReduce MRTWorker mp (Just wd)+++mkMapReduceClient + :: DMRClientConf+ -> IO DMapReduce+mkMapReduceClient conf+ = do+ sid <- getSiteId+ infoM localLogger $ "initialising map-reduce-client on site " ++ show sid + mp <- MP.newMasterPort (clc_StreamName conf) (clc_SocketId conf)+ newDMapReduce MRTClient mp (Nothing::Maybe WP.WorkerPort)+ ++newDMapReduce+ :: (M.MasterClass m, W.WorkerClass w, Debug m, Debug w)+ => MapReduceType -> m -> Maybe w -> IO DMapReduce+newDMapReduce t m w+ = do+ sid <- getSiteId+ d <- newMVar (DMapReduceData sid t m w)+ return $ DMapReduce d+ +{-+getMasterRequestPort :: DMapReduce -> IO MSG.MasterRequestPort+getMasterRequestPort (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> return $ M.getMasterRequestPort m+-} ++++-- ---------------------------------------------------------------------------+-- public functions+-- ---------------------------------------------------------------------------++instance Debug DMapReduce where++ printDebug (DMapReduce mr)+ = withMVar mr $+ \(DMapReduceData s t m w) ->+ do+ putStrLn "--------------------------------------------------------"+ putStrLn "Distribtion - internal data\n"+ putStrLn "--------------------------------------------------------"+ putStrLn "SiteId:"+ putStrLn $ show s+ putStrLn "Type:"+ putStrLn $ show t+ putStrLn "--------------------------------------------------------"+ putStrLn "Master:"+ printDebug m+ putStrLn "--------------------------------------------------------"+ putStrLn "Worker:"+ maybe (putStrLn "NOTHING") (\w' -> printDebug w') w+ putStrLn "--------------------------------------------------------"+++++instance MapReduce DMapReduce where+++ closeMapReduce (DMapReduce mr)+ = withMVar mr $+ \(DMapReduceData _ _ m w) ->+ do+ case w of+ (Just w') -> W.closeWorker w'+ (Nothing) -> return ()+ M.closeMaster m+++ getMySiteId (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData s _ _ _) -> return s++ + getMapReduceType (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ t _ _) -> return t++ + startControlling (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> startControlling m++ + stopControlling (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> stopControlling m++ + isControlling (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> isControlling m++ + doSingleStep (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> doSingleStep m+++ doMapReduceJob ji (DMapReduce mr)+ = withMVar mr $ \(DMapReduceData _ _ m _) -> doMapReduceJob ji m+
+ source/Holumbus/Distribution/Master.hs view
@@ -0,0 +1,39 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Master+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Master+(+ MasterClass(..)+)+where++import qualified Holumbus.MapReduce.Types as T+import qualified Holumbus.MapReduce.MapReduce as MR+++-- ----------------------------------------------------------------------------+-- Typeclass+-- ----------------------------------------------------------------------------+++class (MR.MapReduce m) => MasterClass m where++ closeMaster :: m -> IO () + + receiveTaskCompleted :: T.TaskData -> m -> IO m++ receiveTaskError :: T.TaskData -> m -> IO m++-- ----------------------------------------------------------------------------
+ source/Holumbus/Distribution/Master/MasterData.hs view
@@ -0,0 +1,391 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Master.MasterData+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++{-# OPTIONS_GHC -fno-warn-unused-binds #-} -- for unused record field selectors++module Holumbus.Distribution.Master.MasterData+ (+ -- * Datatypes+ MasterData+ + -- * creation and destruction+ , newMaster+ )+where++import Control.Concurrent++import Data.List+import Data.Maybe+import qualified Data.Set as Set++import System.Log.Logger++import Holumbus.Common.Debug+import Holumbus.Common.Utils++import Holumbus.Network.Site+-- import Holumbus.Network.Port+-- import Holumbus.Network.Messages+import Holumbus.Network.Communication++import qualified Holumbus.MapReduce.MapReduce as MR+import Holumbus.MapReduce.JobController+import Holumbus.MapReduce.Types++import qualified Holumbus.Data.MultiMap as MMap+import qualified Holumbus.FileSystem.FileSystem as FS++import qualified Holumbus.Distribution.Messages as M+import qualified Holumbus.Distribution.Master as MC+import qualified Holumbus.Distribution.Worker as WC+import qualified Holumbus.Distribution.Worker.WorkerPort as WP+import Holumbus.Distribution.Master++localLogger :: String+localLogger = "Holumbus.Distribution.Master.MasterData"++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------++type TaskToWorkerMap = MMap.MultiMap TaskId M.WorkerId+type WorkerToTaskMap = MMap.MultiMap M.WorkerId TaskId++type ActionToWorkerMap = MMap.MultiMap ActionName M.WorkerId+++data WorkerControllerData = WorkerControllerData {+ wcd_TaskToWorkerMap :: ! TaskToWorkerMap+ , wcd_WorkerToTaskMap :: ! WorkerToTaskMap+ , wcd_ActionToWorkerMap :: ! ActionToWorkerMap+ }++type WorkerController = MVar WorkerControllerData++data MasterData = MasterData {+ md_Server :: Server+ , md_WorkerController :: WorkerController+ , md_JobController :: JobController+ , md_FileSystem :: FS.FileSystem+ }++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++newWorkerController :: IO WorkerController+newWorkerController+ = do+ let wcd = WorkerControllerData+ MMap.empty+ MMap.empty+ MMap.empty+ newMVar wcd+++newMaster+ :: FS.FileSystem+ -> Bool -> StreamName -> Maybe PortNumber+ -> IO MasterData+newMaster fs start sn pn+ = do+ -- initialise the server+ m <- newEmptyMVar+ server <- newServer sn pn (dispatch m) (Just $ registerWorker m) (Just $ unregisterWorker m)+ + -- start the WorkerController+ wc <- newWorkerController++ -- start the JobController+ jc <- newJobController+ -- configure the JobController+ setTaskSendHook (sendStartTask server wc) jc+ if (start) + then do startJobController jc+ else do return ()++ -- get the internal data+ let md = MasterData server wc jc fs+ + -- get the internal data+ putMVar m md+ + return md++ +dispatch + :: MVar MasterData + -> M.MasterRequestMessage + -> IO (Maybe M.MasterResponseMessage)+dispatch m msg+ = do+ md <- readMVar m+ case msg of+ (M.MReqTaskCompleted td) ->+ do+ MC.receiveTaskCompleted td md+ return $ Just $ M.MRspSuccess+ (M.MReqTaskError td) ->+ do+ MC.receiveTaskError td md+ return $ Just $ M.MRspSuccess+ (M.MReqStartControlling) ->+ do+ MR.startControlling md+ return $ Just $ M.MRspSuccess + (M.MReqStopControlling) ->+ do+ MR.stopControlling md+ return $ Just $ M.MRspSuccess + (M.MReqIsControlling) ->+ do+ MR.isControlling md+ return $ Just $ M.MRspSuccess + (M.MReqSingleStep) ->+ do+ MR.doSingleStep md+ return $ Just $ M.MRspSuccess+ (M.MReqPerformJob ji) ->+ do+ r <- MR.doMapReduceJob ji md+ return $ Just $ M.MRspResult r+ _ -> return Nothing++++registerWorker :: MVar MasterData -> IdType -> ClientPort -> IO ()+registerWorker m i cp+ = do+ let wp = WP.newWorkerPort cp+ as <- WC.getActionNames wp+ md <- readMVar m+ modifyMVar (md_WorkerController md) $+ \wcd ->+ do+ let wcd' = addWorkerToMaster i as wcd+ return (wcd',())+ + +++unregisterWorker :: MVar MasterData -> IdType -> ClientPort -> IO ()+unregisterWorker m i _+ = do+ md <- readMVar m+ modifyMVar (md_WorkerController md) $+ \wcd ->+ do+ -- TODO delete Tasks from Worker+ let wcd' = deleteWorkerFromMaster i wcd + return (wcd',())+ + return ()+++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------++addWorkerToMaster+ :: M.WorkerId -> [ActionName]+ -> WorkerControllerData -> WorkerControllerData+addWorkerToMaster wid as wcd+ = wcd { wcd_ActionToWorkerMap = awm' }+ where+ -- update the ActionToWorkerMap+ awm' = MMap.insertKeys as (Set.singleton wid) $ wcd_ActionToWorkerMap wcd+ ++deleteWorkerFromMaster :: M.WorkerId -> WorkerControllerData -> WorkerControllerData+deleteWorkerFromMaster wid wcd+ = wcd { wcd_ActionToWorkerMap = awm' }+ where+ -- update the ActionToWorkerMap + awm' = MMap.deleteAllElems wid $ wcd_ActionToWorkerMap wcd+++getTasksPerWorkerWithAction :: ActionName -> WorkerControllerData -> [(Int, M.WorkerId)]+getTasksPerWorkerWithAction a wcd = nullList ++ sortedList+ where+ wtm = wcd_WorkerToTaskMap wcd+ -- all Workers with the needed Action+ allWids = MMap.lookup a $ wcd_ActionToWorkerMap wcd+ -- all WorkerIds with Tasks and the Action+ taskWids = Set.intersection allWids $ MMap.keys wtm+ -- all WorkerIds without Tasks+ noTaskWids = Set.difference allWids taskWids+ -- list with all Workers with no tasks+ nullList = map (\wid -> (0,wid)) (Set.toList noTaskWids)+ -- list with all Workers with their tasks+ ls2 = map (\(wid,s) -> (Set.size s, wid)) (MMap.toList wtm)+ -- merging and sorting the list+ sortedList = sortBy (\(n1,_) (n2,_) -> compare n1 n2) ls2+ ++addTaskToWorker :: TaskId -> M.WorkerId -> WorkerControllerData -> WorkerControllerData+addTaskToWorker tid wid wcd = wcd { wcd_TaskToWorkerMap = twm', wcd_WorkerToTaskMap = wtm'}+ where+ twm' = MMap.insert tid wid (wcd_TaskToWorkerMap wcd)+ wtm' = MMap.insert wid tid (wcd_WorkerToTaskMap wcd)+ ++-- TODO use this+deleteTaskFromWorker :: TaskId -> M.WorkerId -> WorkerControllerData -> WorkerControllerData+deleteTaskFromWorker tid wid wcd = wcd { wcd_TaskToWorkerMap = twm', wcd_WorkerToTaskMap = wtm'}+ where+ twm' = MMap.deleteElem tid wid (wcd_TaskToWorkerMap wcd)+ wtm' = MMap.deleteElem wid tid (wcd_WorkerToTaskMap wcd)++-- TODO use this+deleteTaskFromWorkers :: TaskId -> WorkerControllerData -> WorkerControllerData+deleteTaskFromWorkers tid wcd = wcd''+ where+ wids = Set.toList $ MMap.lookup tid (wcd_TaskToWorkerMap wcd)+ wcd'' = foldl (\wcd' wid -> deleteTaskFromWorker tid wid wcd') wcd wids+++sendStartTask :: Server -> WorkerController -> TaskData -> IO (TaskSendResult)+sendStartTask s wc td+ = handleAll ( \ e -> do + errorM localLogger $ "sendStartTask: " ++ show e+ return TSRError+ ) $+ do+ debugM localLogger $ "sendStartTask: waiting for wc"+ modifyMVar wc $+ \wcd ->+ do+ debugM localLogger $ "sendStartTask: waiting for wc -> done"+ -- get all workers with the required action, sorted by the number of current tasks + let wls = getTasksPerWorkerWithAction (td_Action td) wcd+ if (null wls)+ then do+ warningM localLogger $ "sendStartTask: no worker with action \"" ++ (td_Action td) ++ "\"found"+ return (wcd, TSRNotSend)+ else do + infoM localLogger $ "starting Task: " ++ show (td_TaskId td)+ debugM localLogger $ "sendStartTask: wls:" ++ show wls+ let (_,wid) = head wls+ info <- getClientInfo wid s+ case info of+ (Just ci) ->+ do+ WC.startTask td $ WP.newWorkerPort (ci_Port ci)+ let wcd' = addTaskToWorker (td_TaskId td) wid wcd+ return (wcd', TSRSend) + (Nothing) ->+ return (wcd, TSRNotSend)++-- ----------------------------------------------------------------------------+-- typeclass instanciation+-- ----------------------------------------------------------------------------++instance MR.MapReduce MasterData where++ closeMapReduce md+ = closeMaster md++ getMySiteId _ + = getSiteId++ + getMapReduceType _+ = return MR.MRTMaster++ + startControlling md + = do+ debugM localLogger "startControlling"+ startJobController (md_JobController md)+++ stopControlling md + = do+ debugM localLogger "stopControlling"+ stopJobController (md_JobController md)+++ isControlling md+ = do+ debugM localLogger "isControlling"+ isJobControllerRunning (md_JobController md)++ + doSingleStep md+ = do+ debugM localLogger "doSingleStep"+ singleStepJobControlling (md_JobController md)++ + doMapReduceJob ji md+ = do+ debugM localLogger "doMapReduceJob"+ performJob ji (md_JobController md)+++++instance MasterClass MasterData where+++ closeMaster md + = do+ closeServer (md_Server md)+ return ()+++ receiveTaskCompleted td md+ = do+ debugM localLogger $ "completed Task: " ++ show (td_TaskId td)+ setTaskCompleted (md_JobController md) td+ -- TODO inform other workers+ -- TODO delete from workerController+ return md+++ receiveTaskError td md+ = do+ debugM localLogger $ "error Task: " ++ show (td_TaskId td)+ setTaskError (md_JobController md) td+ -- TODO inform other workers+ -- TODO delete from workerController+ return md+ + +++instance Debug MasterData where+ printDebug md+ = do+ putStrLn "Master-Object (full)"+ putStrLn "--------------------------------------------------------" + putStrLn "Server"+ printDebug (md_Server md)+ withMVar (md_WorkerController md) $+ \wcd -> + do+ putStrLn $ prettyRecordLine gap "TaskToWorkerMap: " (wcd_TaskToWorkerMap wcd)+ putStrLn $ prettyRecordLine gap "WorkerToTaskMap" (wcd_WorkerToTaskMap wcd)+ putStrLn $ prettyRecordLine gap "ActionToWorkerMap:" (wcd_ActionToWorkerMap wcd) + putStrLn $ "JobController:"+ jc <- printJobController (md_JobController md)+ putStrLn jc+ where+ gap = 20+
+ source/Holumbus/Distribution/Master/MasterPort.hs view
@@ -0,0 +1,169 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Master.MasterPort+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Master.MasterPort+(+-- * Datatypes+ MasterPort+ +-- * Creation and Destruction+, newMasterPort+, newMasterPortFromServerPort+)+where++import Holumbus.Network.Communication+import Holumbus.Network.Site++import Holumbus.Common.Debug+import Holumbus.MapReduce.MapReduce+import Holumbus.Distribution.Messages+import Holumbus.Distribution.Master+++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+++data MasterPort = MasterPort ServerPort+ deriving (Show)++++-- ----------------------------------------------------------------------------+-- Creation and Destruction+-- ----------------------------------------------------------------------------+++-- | Creates a new ControllerPort+newMasterPort :: StreamName -> Maybe SocketId -> IO MasterPort+newMasterPort sn soid+ = do+ sp <- newServerPort sn soid+ return (MasterPort sp)++newMasterPortFromServerPort :: ServerPort -> MasterPort+newMasterPortFromServerPort sp = MasterPort sp+++-- ----------------------------------------------------------------------------+-- Typeclass instanciation+-- ----------------------------------------------------------------------------++instance Debug MasterPort where++ printDebug (MasterPort p)+ = do+ putStrLn "MasterPort:"+ putStrLn $ show p+ + + ++instance MapReduce MasterPort where+++ closeMapReduce md+ = closeMaster md+++ getMySiteId _+ = getSiteId++ + getMapReduceType _+ = return MRTMaster++ + startControlling (MasterPort p)+ = do+ sendRequestToServer p time30 (MReqStartControlling) $+ \rsp ->+ do+ case rsp of+ (MRspSuccess) -> return (Just ())+ _ -> return Nothing++ + stopControlling (MasterPort p)+ = do+ sendRequestToServer p time30 (MReqStopControlling) $+ \rsp ->+ do+ case rsp of+ (MRspSuccess) -> return (Just ())+ _ -> return Nothing+++ isControlling (MasterPort p)+ = do+ sendRequestToServer p time30 (MReqIsControlling) $+ \rsp ->+ do+ case rsp of+ (MRspIsControlling b) -> return (Just b)+ _ -> return Nothing+ ++ + doSingleStep (MasterPort p)+ = do+ sendRequestToServer p time30 (MReqSingleStep) $+ \rsp ->+ do+ case rsp of+ (MRspSuccess) -> return (Just ())+ _ -> return Nothing++ + doMapReduceJob ji (MasterPort p)+ = do+ -- bad hack... we should build an extra client for the MapReduce-System+ -- which pings for the server and so on+ sendRequestToServer p timeIndefinitely (MReqPerformJob ji) $+ \rsp ->+ do+ case rsp of+ (MRspResult r) -> return (Just r)+ _ -> return Nothing+++++instance MasterClass MasterPort where+++ closeMaster _ = return ()+++ receiveTaskCompleted td c@(MasterPort p)+ = do+ sendRequestToServer p time30 (MReqTaskCompleted td) $+ \rsp ->+ do+ case rsp of+ (MRspSuccess) -> return (Just c)+ _ -> return Nothing+ ++ receiveTaskError td c@(MasterPort p)+ = do+ sendRequestToServer p time30 (MReqTaskError td) $+ \rsp ->+ do+ case rsp of+ (MRspSuccess) -> return (Just c)+ _ -> return Nothing
+ source/Holumbus/Distribution/Messages.hs view
@@ -0,0 +1,208 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Messages+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Messages+(+-- * Datatypes+ WorkerId++-- * Messages to and from the Master+, MasterRequestMessage(..)+, MasterResponseMessage(..)++-- * Messages to and from the Worker+, WorkerRequestMessage(..)+, WorkerResponseMessage(..)++-- * request an response handling+, performPortAction -- reexport from Holumbus.Network.Messages+)+where+++import Data.Binary++import Holumbus.Network.Messages+import Holumbus.MapReduce.Types+++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------++type WorkerId = Int++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++-- Messages to and from the Master+data MasterRequestMessage + = MReqTaskCompleted TaskData+ | MReqTaskError TaskData + | MReqStartControlling+ | MReqStopControlling+ | MReqIsControlling+ | MReqSingleStep+ | MReqPerformJob JobInfo+ | MReqUnknown+ deriving (Show)+++instance Binary MasterRequestMessage where+ put (MReqTaskCompleted td) = putWord8 1 >> put td+ put (MReqTaskError td) = putWord8 2 >> put td + put (MReqStartControlling) = putWord8 3+ put (MReqStopControlling) = putWord8 4+ put (MReqIsControlling) = putWord8 5+ put (MReqSingleStep) = putWord8 6+ put (MReqPerformJob ji) = putWord8 7 >> put ji+ put (MReqUnknown) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> get >>= \td -> return (MReqTaskCompleted td)+ 2 -> get >>= \td -> return (MReqTaskError td) + 3 -> return (MReqStartControlling)+ 4 -> return (MReqStopControlling)+ 5 -> return (MReqIsControlling)+ 6 -> return (MReqSingleStep)+ 7 -> get >>= \ji -> return (MReqPerformJob ji)+ _ -> return (MReqUnknown)++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++data MasterResponseMessage + = MRspSuccess+ | MRspIsControlling Bool+ | MRspResult JobResult+ | MRspError String+ | MRspUnknown+ deriving (Show)+++instance RspMsg MasterResponseMessage where+ isError (MRspError _) = True+ isError _ = False+ + getErrorMsg (MRspError e) = e+ getErrorMsg _ = ""+ + isUnknown (MRspUnknown) = True+ isUnknown _ = False+ + mkErrorMsg e = MRspError e+++instance Binary MasterResponseMessage where+ put (MRspSuccess) = putWord8 1+ put (MRspIsControlling b) = putWord8 2 >> put b+ put (MRspResult jr) = putWord8 3 >> put jr+ put (MRspError e) = putWord8 4 >> put e+ put (MRspUnknown) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> return (MRspSuccess)+ 2 -> get >>= \b -> return (MRspIsControlling b)+ 3 -> get >>= \jr -> return (MRspResult jr)+ 4 -> get >>= \e -> return (MRspError e)+ _ -> return (MRspUnknown)+++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++-- Messages to and from the Worker+data WorkerRequestMessage+ = WReqStartTask TaskData+ | WReqStopTask TaskId+ | WReqStopAllTasks+ | WReqGetActionNames+ | WReqUnknown+ deriving (Show)+++instance Binary WorkerRequestMessage where+ put (WReqStartTask td) = putWord8 1 >> put td+ put (WReqStopTask tid) = putWord8 2 >> put tid+ put (WReqStopAllTasks) = putWord8 3+ put (WReqGetActionNames) = putWord8 4+ put (WReqUnknown) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> get >>= \td -> return (WReqStartTask td)+ 2 -> get >>= \tid -> return (WReqStopTask tid)+ 3 -> return (WReqStopAllTasks)+ 4 -> return (WReqGetActionNames)+ _ -> return (WReqUnknown)++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++data WorkerResponseMessage+ = WRspSuccess+ | WRspGetActionNames [ActionName]+ | WRspError String+ | WRspUnknown+ deriving (Show)+++instance RspMsg WorkerResponseMessage where+ isError (WRspError _) = True+ isError _ = False+ + getErrorMsg (WRspError e) = e+ getErrorMsg _ = ""+ + isUnknown (WRspUnknown) = True+ isUnknown _ = False++ mkErrorMsg e = WRspError e+++instance Binary WorkerResponseMessage where+ put (WRspSuccess) = putWord8 1+ put (WRspGetActionNames as) = putWord8 2 >> put as+ put (WRspError e) = putWord8 3 >> put e+ put (WRspUnknown) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> return (WRspSuccess)+ 2 -> get >>= \as -> return (WRspGetActionNames as)+ 3 -> get >>= \e -> return (WRspError e)+ _ -> return (WRspUnknown)
+ source/Holumbus/Distribution/Worker.hs view
@@ -0,0 +1,39 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Worker+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Worker+(+ WorkerClass(..)+)+where++import qualified Holumbus.MapReduce.Types as T++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++class WorkerClass w where++ closeWorker :: w -> IO ()++ startTask :: T.TaskData -> w -> IO w++ stopTask :: T.TaskId -> w -> IO w++ stopAllTasks :: w -> IO w+ + getActionNames :: w -> IO [T.ActionName]
+ source/Holumbus/Distribution/Worker/WorkerData.hs view
@@ -0,0 +1,192 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Worker.WorkerData+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Worker.WorkerData+ (+ -- * Datatypes+ WorkerData+ + -- * creation and destruction+ , newWorker+ )+where+++import Control.Concurrent++import Data.Maybe++import System.Log.Logger++import Holumbus.Common.Debug+import Holumbus.Common.Utils ( handleAll )++import qualified Holumbus.FileSystem.FileSystem as FS+import Holumbus.Network.Communication+import Holumbus.MapReduce.Types+import qualified Holumbus.MapReduce.TaskProcessor as TP++import qualified Holumbus.Distribution.Messages as M+import qualified Holumbus.Distribution.Master.MasterPort as MP+import qualified Holumbus.Distribution.Master as MC+import Holumbus.Distribution.Worker++localLogger :: String+localLogger = "Holumbus.Distribution.Worker.WorkerData"+++data WorkerData = WorkerData {+ wd_Client :: Client+ , wd_TaskProcessor :: TP.TaskProcessor + }+ + +++newWorker :: FS.FileSystem -> ActionMap -> StreamName -> Maybe SocketId -> IO WorkerData+newWorker fs am sn soid+ = do+ -- initialise the client+ w <- newEmptyMVar+ client <- newClient sn soid (dispatch w)++ mp <- MP.newMasterPort sn soid+ mpMVar <- newMVar mp+ + -- configure the TaskProcessor+ tp <- TP.newTaskProcessor+ TP.setFileSystemToTaskProcessor fs tp+ TP.setActionMap am tp+ TP.setTaskCompletedHook (sendTaskCompleted mpMVar) tp+ TP.setTaskErrorHook (sendTaskError mpMVar) tp+ TP.startTaskProcessor tp+ + let wd = (WorkerData client tp)++ putMVar w wd++ return wd+++dispatch + :: MVar WorkerData+ -> M.WorkerRequestMessage + -> IO (Maybe M.WorkerResponseMessage)+dispatch w msg+ = do+ wd <- readMVar w+ case msg of+ (M.WReqStartTask td) ->+ do+ startTask td wd+ return $ Just $ M.WRspSuccess+ (M.WReqStopTask tid) ->+ do+ stopTask tid wd+ return $ Just $ M.WRspSuccess+ (M.WReqStopAllTasks) ->+ do+ stopAllTasks wd+ return $ Just $ M.WRspSuccess+ (M.WReqGetActionNames) ->+ do+ as <- getActionNames wd+ return $ Just $ M.WRspGetActionNames as+ _ -> return Nothing++++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++sendTaskCompleted :: MVar MP.MasterPort -> TaskData -> IO Bool+sendTaskCompleted mvmp td+ = handleAll (\_ -> return False) $+ modifyMVar mvmp $+ \mp ->+ do+ debugM localLogger $ "completed Task" ++ show (td_TaskId td)+ MC.receiveTaskCompleted td mp+ return (mp, True)+++sendTaskError :: MVar MP.MasterPort -> TaskData -> IO Bool+sendTaskError mvmp td+ = handleAll (\_ -> return False) $+ modifyMVar mvmp $+ \mp ->+ do+ debugM localLogger $ "error Task" ++ show (td_TaskId td)+ MC.receiveTaskError td mp+ return (mp, True)+++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------++++instance WorkerClass WorkerData where++ closeWorker wd+ = do+ closeClient (wd_Client wd)+ return ()+ + + startTask td wd+ = do+ let tp = (wd_TaskProcessor wd)+ debugM localLogger $ "executing Task" ++ show (td_TaskId td)+ TP.startTask td tp+ return wd+ + + stopTask tid wd+ = do+ let tp = (wd_TaskProcessor wd)+ debugM localLogger $ "stopping Task" ++ show tid + TP.stopTask tid tp+ return wd+ ++ stopAllTasks wd+ = do+ let tp = (wd_TaskProcessor wd)+ debugM localLogger "stopping all Tasks"+ TP.stopAllTasks tp+ return wd+++ getActionNames wd+ = do + debugM localLogger "getting action names"+ as <- TP.getActionNames (wd_TaskProcessor wd)+ debugM localLogger $ "actions: " ++ show as+ return as+ ++ +instance Debug WorkerData where+ printDebug wd+ = do+ putStrLn "Worker-Object (full)"+ printDebug (wd_Client wd)+ tp <- TP.printTaskProcessor (wd_TaskProcessor wd)+ putStrLn "TaskProcessor:"+ putStrLn tp
+ source/Holumbus/Distribution/Worker/WorkerPort.hs view
@@ -0,0 +1,102 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Distribution.Worker.WorkerPort+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Worker.WorkerPort+(+-- * Datatypes+ WorkerPort+ +-- * Creation and Destruction+, newWorkerPort+)+where++import Holumbus.Common.Debug+import Holumbus.Network.Communication+import Holumbus.Distribution.Messages+import Holumbus.Distribution.Worker+++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+++data WorkerPort = WorkerPort ClientPort+ deriving (Show)+++-- | Creates a new NodePort.+newWorkerPort :: ClientPort -> WorkerPort+newWorkerPort p = WorkerPort p++++-- ----------------------------------------------------------------------------+-- Typeclass instanciation+-- ----------------------------------------------------------------------------+++instance WorkerClass WorkerPort where+ + closeWorker _ = return ()+++ startTask td w@(WorkerPort p)+ = do+ sendRequestToClient p time30 (WReqStartTask td) $+ \rsp ->+ do+ case rsp of+ (WRspSuccess) -> return (Just w)+ _ -> return Nothing+++ stopTask tid w@(WorkerPort p)+ = do+ sendRequestToClient p time30 (WReqStopTask tid) $+ \rsp ->+ do+ case rsp of+ (WRspSuccess) -> return (Just w)+ _ -> return Nothing+++ stopAllTasks w@(WorkerPort p)+ = do+ sendRequestToClient p time30 (WReqStopAllTasks) $+ \rsp ->+ do+ case rsp of+ (WRspSuccess) -> return (Just w)+ _ -> return Nothing+++ getActionNames (WorkerPort p)+ = do+ sendRequestToClient p time30 (WReqGetActionNames) $+ \rsp ->+ do+ case rsp of+ (WRspGetActionNames as) -> return (Just as)+ _ -> return Nothing++++instance Debug WorkerPort where+ printDebug (WorkerPort p)+ = do+ putStrLn "WorkerPort:"+ putStrLn $ show p
+ source/Holumbus/MapReduce/JobController.hs view
@@ -0,0 +1,761 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.MapReduce.JobController+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.MapReduce.JobController+ ( TaskSendResult(..)+ , TaskSendFunction++ , JobController++ , printJobController++ -- * Creation and Destruction+ , newJobController+ , closeJobController+ , setFileSystemToJobController+ , setTaskSendHook++ -- * Job Controller + , startJobController+ , stopJobController+ , isJobControllerRunning+ , singleStepJobControlling++ -- * performing MapReduce-Jobs+ -- , startJob+ -- , stopJob+ , performJob++ -- * handling Task-Responses+ , setTaskCompleted+ , setTaskError+ )+where++import Prelude hiding ( catch )++import Control.Exception ( Exception+ , throwTo+ , catch+ )++import Control.Concurrent+import Control.Monad++import qualified Data.ByteString.Lazy as B+import Data.Maybe+import Data.Time+import Data.Typeable+import qualified Data.Map as Map+import qualified Data.Set as Set+import System.Log.Logger++import Holumbus.MapReduce.Types+import qualified Holumbus.FileSystem.FileSystem as FS+import qualified Holumbus.Data.AccuMap as AMap+import qualified Holumbus.Data.MultiMap as MMap+++localLogger :: String+localLogger = "Holumbus.MapReduce.JobController"++cycleLogger :: String+cycleLogger = "Holumbus.MapReduce.JobController.cycle"++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++newJobData + :: JobControllerData -> JobInfo+ -> IO (JobControllerData, JobData, MVar JobResult)+newJobData jcd info+ = do+ t <- getCurrentTime+ mVar <- newEmptyMVar+ let state = JSIdle+ parts = maybe 1 id $ getCurrentTaskPartValue state info+ jrc = JobResultContainer mVar + jid = jcd_NextJobId jcd+ jcd' = jcd { jcd_NextJobId = (jid+1) }+ pairs = initialSplit parts (ji_Input info) + -- let pairs = zip (iterate (+1) 1) (map (\i -> [i]) (ji_Input info))+ accuMap = AMap.fromList pairs + outputMap = Map.insert JSIdle accuMap Map.empty+ return (jcd', JobData jid state outputMap info t t jrc, mVar)+ where+ initialSplit n ls = ps+ where+ ns = [(x `mod` n) + 1 | x <- [1..]]+ is = map (\a -> [a]) ls + ps = zip ns is+++newTaskData + :: JobControllerData + -> JobId -> TaskType -> TaskState -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> ActionName -> TaskOutputType + -> IO (JobControllerData, TaskData)+newTaskData jcd jid tt ts opt n i a ot+ = do+ let tid = jcd_NextTaskId jcd+ let jcd' = jcd { jcd_NextTaskId = (tid+1) }+ return (jcd', TaskData jid tid tt ts opt n i [] ot a) + +data TaskSendResult = TSRSend | TSRNotSend | TSRError+ deriving (Show, Eq, Ord, Enum) + +-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------++intersections :: Ord a => [Set.Set a] -> Set.Set a +intersections [] = Set.empty+intersections (s:ss) = foldl Set.intersection s ss +++mapAccumLM+ :: (Monad m) + => (acc -> x -> m (acc, y)) -- Function of of input list+ -- and accumulator, returning new+ -- accumulator and of result list+ -> acc -- Initial accumulator + -> [x] -- Input list+ -> m (acc, [y]) -- Final accumulator and result list+mapAccumLM _ s [] = return (s, [])+mapAccumLM f s (x:xs) + = do+ (s', y ) <- f s x+ (s'',ys) <- mapAccumLM f s' xs+ return (s'',y:ys)+++++type TaskSendFunction = TaskData -> IO (TaskSendResult)++dummyTaskSendFunction :: TaskData -> IO (TaskSendResult)+dummyTaskSendFunction _ = return TSRSend+++type JobMap = Map.Map JobId JobData+type TaskMap = Map.Map TaskId TaskData+ +type StateJobIdMap = MMap.MultiMap JobState JobId++type JobIdTaskIdMap = MMap.MultiMap JobId TaskId+type TypeTaskIdMap = MMap.MultiMap TaskType TaskId+type StateTaskIdMap = MMap.MultiMap TaskState TaskId+ + +data JobControlFunctions = JobControlFunctions {+ jcf_TaskSend :: TaskSendFunction+ }+instance Show JobControlFunctions where+ show _ = "JobControlFunctions"++ +data JobControllerData = JobControllerData {+ -- internal+ jcd_ServerThreadId :: Maybe ThreadId+ , jcd_ServerDelay :: Int+ , jcd_FileSystem :: Maybe FS.FileSystem+ -- control+ , jcd_NextJobId :: JobId+ , jcd_NextTaskId :: TaskId+ , jcd_Functions :: JobControlFunctions+ -- job control+ , jcd_JobMap :: ! JobMap+ , jcd_TaskMap :: ! TaskMap+ , jcd_StateJobIdMap :: ! StateJobIdMap+ , jcd_JobIdTaskIdMap :: ! JobIdTaskIdMap+ , jcd_TypeTaskIdMap :: ! TypeTaskIdMap+ , jcd_StateTaskIdMap :: ! StateTaskIdMap+ } deriving (Show)+++type JobController = MVar JobControllerData++printJobController :: JobController -> IO String+printJobController jc+ = withMVar jc $ \jcd -> return $ show jcd+++data JobControllerException+ = KillServerException+ deriving (Show, Typeable)++instance Exception JobControllerException where++-- ----------------------------------------------------------------------------+-- Contruction / Destruction +-- ----------------------------------------------------------------------------++defaultJobControllerData :: JobControllerData+defaultJobControllerData = jcd+ where+ jcd = JobControllerData+ Nothing+ 10000 -- 10 milliseconds delay+ Nothing+ 1+ 1+ jcf+ Map.empty+ Map.empty+ MMap.empty+ MMap.empty+ MMap.empty+ MMap.empty+ jcf = JobControlFunctions+ dummyTaskSendFunction++newJobController :: IO JobController+newJobController+ = do+ let jcd = defaultJobControllerData+ jc <- newMVar jcd+ return jc ++closeJobController :: JobController -> IO ()+closeJobController jc+ = do+ stopJobController jc+++setFileSystemToJobController :: FS.FileSystem -> JobController -> IO ()+setFileSystemToJobController fs jc+ = modifyMVar jc $+ \jcd -> return $ (jcd { jcd_FileSystem = Just fs }, ())+++setTaskSendHook :: TaskSendFunction -> JobController -> IO ()+setTaskSendHook f jc+ = do+ modifyMVar jc $+ \jcd ->+ do+ let funs = jcd_Functions jcd+ let funs' = funs { jcf_TaskSend = f }+ return (jcd { jcd_Functions = funs' }, ())+++-- ----------------------------------------------------------------------------+-- server functions+-- ----------------------------------------------------------------------------++startJobController :: JobController -> IO ()+startJobController jc+ = do+ modifyMVar jc $ + \jcd -> + do+ thd <- case (jcd_ServerThreadId jcd) of+ (Just i) -> return i+ (Nothing) ->+ do+ i <- forkIO $ doProcessing jc True+ return i+ return (jcd {jcd_ServerThreadId = (Just thd)}, ())++++stopJobController :: JobController -> IO ()+stopJobController jc+ = do+ modifyMVar jc $ + \jcd -> + do+ case (jcd_ServerThreadId jcd) of+ (Nothing) -> return ()+ (Just i) -> + do+ throwTo i KillServerException+ yield+ return ()+ return (jcd {jcd_ServerThreadId = Nothing}, ())+++isJobControllerRunning :: JobController -> IO Bool+isJobControllerRunning jc+ = withMVar jc $+ \jcd -> return $ isJust (jcd_ServerThreadId jcd)+++singleStepJobControlling :: JobController -> IO ()+singleStepJobControlling jc+ = do+ singleStepAllowed <- withMVar jc $ + \jcd -> return $ isNothing (jcd_ServerThreadId jcd)+ if singleStepAllowed then doProcessing jc False else return ()++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------++-- | get the JobIds, they can be filtered by JobState, if the filter is empty+-- all Jobs are returned.+getJobIds :: [JobState] -> JobControllerData -> [JobId]+getJobIds states jcd = Set.toList $ MMap.filterElements states (jcd_StateJobIdMap jcd)+++-- | get the TaskIds, they can be filtered by JobId, TaskType and TaskState, if+-- a filter is empty, this attribute is not taken into account+getTaskIds :: [JobId] -> [TaskType] -> [TaskState] -> JobControllerData -> [TaskId]+getTaskIds jobs types states jcd = Set.toList $ intersections sets+ where+ jobSelected = MMap.filterElements jobs (jcd_JobIdTaskIdMap jcd)+ typeSelected = MMap.filterElements types (jcd_TypeTaskIdMap jcd)+ stateSelected = MMap.filterElements states (jcd_StateTaskIdMap jcd)+ sets = [jobSelected, typeSelected, stateSelected]+++allTasksFinished :: JobControllerData -> JobId -> Bool+allTasksFinished jcd jid = null notFinishedTasks+ where+ notFinishedTasks = getTaskIds [jid] [] [TSIdle, TSSending, TSInProgress, TSCompleted] jcd++++-- | adds a Job to the JobControllerData for execution+addJob :: JobData -> JobControllerData -> JobControllerData+addJob jd jcd = jcd { jcd_JobMap = jm', jcd_StateJobIdMap = sjm' }+ where+ jm' = Map.insert (jd_JobId jd) jd (jcd_JobMap jcd)+ sjm' = MMap.insert (jd_State jd) (jd_JobId jd) (jcd_StateJobIdMap jcd)+++changeJobState :: JobId -> JobState -> JobControllerData -> JobControllerData+changeJobState jid js jcd = changeJobState' (Map.lookup jid (jcd_JobMap jcd))+ where+ changeJobState' (Nothing) = jcd+ changeJobState' (Just jd) = jcd { jcd_JobMap = jm', jcd_StateJobIdMap = sjm' }+ where+ jd' = jd { jd_State = js } -- change JobData+ js' = jd_State jd -- get old state+ jm' = Map.insert jid jd' (jcd_JobMap jcd) -- change JobData+ sjm' = MMap.insert js jid $ MMap.deleteElem js' jid (jcd_StateJobIdMap jcd) -- change StateJobIdMap+++updateJob :: JobData -> JobControllerData -> JobControllerData+updateJob jd jcd = addJob jd $ maybe (jcd) (\_ -> changeJobState jid js jcd) maybeOldJob+ where+ maybeOldJob = (Map.lookup jid (jcd_JobMap jcd))+ jid = jd_JobId jd+ js = jd_State jd+++addOutputToJob :: [(Int,[FunctionData])] -> JobData -> JobData+addOutputToJob outList jd = jd { jd_OutputMap = outputmap' } + where+ outAccu = AMap.fromList outList+ state = jd_State jd+ outputmap = jd_OutputMap jd+ outputmap' = Map.insertWith (\m1 m2 -> AMap.union m1 m2) state outAccu outputmap ++addTask :: JobControllerData -> TaskData -> JobControllerData+addTask jcd td= jcd { jcd_TaskMap = tm', jcd_JobIdTaskIdMap = jtm', jcd_TypeTaskIdMap = ttm', jcd_StateTaskIdMap = stm' }+ where+ tid = td_TaskId td+ tm' = Map.insert tid td (jcd_TaskMap jcd)+ jtm' = MMap.insert (td_JobId td) tid (jcd_JobIdTaskIdMap jcd)+ ttm' = MMap.insert (td_Type td) tid (jcd_TypeTaskIdMap jcd)+ stm' = MMap.insert (td_State td) tid (jcd_StateTaskIdMap jcd)+++changeTaskState :: TaskId -> TaskState -> JobControllerData -> JobControllerData+changeTaskState tid ts jcd = changeTaskState' (Map.lookup tid (jcd_TaskMap jcd))+ where+ changeTaskState' (Nothing) = jcd+ changeTaskState' (Just td) = + if isChangeTaskStateAllowed ts (td_State td)+ then jcd { jcd_TaskMap = tm', jcd_StateTaskIdMap = stm' }+ else jcd+ where+ td' = td { td_State = ts } -- change TaskData+ ts' = td_State td -- get old state+ tm' = Map.insert tid td' (jcd_TaskMap jcd) -- change TaskData+ stm' = MMap.insert ts tid $ MMap.deleteElem ts' tid (jcd_StateTaskIdMap jcd) -- change StateTaskIdMap+++-- | newState oldState+isChangeTaskStateAllowed :: TaskState -> TaskState -> Bool+-- only Error-Tasks and Sending-Tasks can be reset to idle+isChangeTaskStateAllowed TSIdle t = Set.member t $ Set.fromList [TSSending, TSError]+-- sending only from idle+isChangeTaskStateAllowed TSSending t = Set.member t $ Set.fromList [TSIdle]+-- inProgress only from an idle Tasks or sending Task+isChangeTaskStateAllowed TSInProgress t = Set.member t $ Set.fromList [TSSending, TSIdle]+-- completed only sending, idle and inprogress+isChangeTaskStateAllowed TSCompleted t = Set.member t $ Set.fromList [TSSending, TSIdle, TSInProgress]+-- finished all but not error+isChangeTaskStateAllowed TSFinished t = Set.member t $ Set.fromList [TSSending, TSIdle, TSInProgress, TSCompleted]+-- we can always set a task to error+isChangeTaskStateAllowed TSError _ = True+-- otherwise false (there are no other cases)+-- isChangeTaskStateAllowed _ _ = False++updateTaskOutput :: TaskId -> [(Int, [FunctionData])] -> JobControllerData -> JobControllerData+updateTaskOutput _ [] jcd = jcd+updateTaskOutput tid o jcd = updateTaskOuput' (Map.lookup tid (jcd_TaskMap jcd))+ where+ updateTaskOuput' (Nothing) = jcd+ updateTaskOuput' (Just td) = jcd { jcd_TaskMap = tm' }+ where+ td' = td { td_Output = o } -- change TaskData+ tm' = Map.insert tid td' (jcd_TaskMap jcd) -- change TaskData+++-- ----------------------------------------------------------------------------+-- Info an Debug+-- ----------------------------------------------------------------------------+++-- ----------------------------------------------------------------------------+-- Task Controlling+-- ----------------------------------------------------------------------------+++startJob + :: JobInfo -> JobController -> IO (JobId, MVar JobResult)+startJob ji jc+ = do+ modifyMVar jc $+ \jcd ->+ do+ -- test the job info... we don't want false jobs+ -- let (b, m) = testJobInfo ji (jcd_MapActionMap jcd) (jcd_ReduceActionMap jcd)+ -- if b + -- then do+ debugM localLogger $ "startJob: " ++ show ji+ + (jcd',jd, jr) <- newJobData jcd ji+ return (addJob jd jcd', (jd_JobId jd, jr))+ -- else do+ -- return (jcd, Left m)++{-+stopJob :: JobId -> JobController -> IO ()+stopJob _ _+ = do+-}++performJob+ :: JobInfo -> JobController -> IO JobResult+performJob ji jc+ = do+ -- start the Job+ (_, mjr) <- startJob ji jc+ -- wait until the result is ready+ withMVar mjr $ \jr -> return jr++-- ----------------------------------------------------------------------------+-- Task Processing+-- ----------------------------------------------------------------------------+++doProcessing :: JobController -> Bool -> IO ()+doProcessing jc loop+ = do+ catch (doProcessing' jc loop)+ handler+ where+ handler :: JobControllerException -> IO ()+ handler err = errorM cycleLogger $ "doProcessing: " ++ (show err)+ doProcessing' jc' loop'+ = do+ handleTasks jc'+ handleJobs jc'+ if loop' + then do+ delay <- withMVar jc' (\jcd -> return $ jcd_ServerDelay jcd)+ threadDelay delay+ doProcessing' jc' loop'+ else+ return ()++++-- ----------------------------------------------------------------------------+-- Process Tasks+-- ----------------------------------------------------------------------------+++-- | the send task doesn't block, because a timeout causes the whole job+-- controller to pause, we don't want this+sendTask :: JobController -> TaskData -> IO ()+sendTask jc td+ = do+ -- own thread for this... so it is non-blocking+ forkIO $ + do+ yield+ -- get the sendFunction (just read it)+ sendFunction <- withMVar jc $+ \jcd -> return $ jcf_TaskSend $ jcd_Functions jcd+ -- execute the send function (this might timeout or block)+ sendResult <- sendFunction td+ -- set the task in the correct state+ case sendResult of+ (TSRSend) ->+ do+ setTaskInProgress jc td+ (TSRNotSend) ->+ do+ -- this function can be called twice before it is finished+ -- and therefore, we introduced the "sending" state for tasks+ -- and if the sending is not ok, we have to set the task to idle again+ setTaskIdle jc td+ (TSRError) ->+ do+ setTaskError jc td+ return ()+++-- toNextTaskState :: JobControllerData -> TaskData -> JobControllerData+-- toNextTaskState jcd td = changeTaskState (td_TaskId td) (getNextTaskState (td_State td)) jcd ++toSendingTaskState :: JobControllerData -> TaskData -> JobControllerData+toSendingTaskState jcd td = changeTaskState (td_TaskId td) TSSending jcd++finishTask :: JobControllerData -> TaskData -> JobControllerData+finishTask jcd td = changeTaskState (td_TaskId td) (getNextTaskState (td_State td)) jcd' +-- toNextTaskState jcd' td+ where+ jd = fromJust $ Map.lookup (td_JobId td) (jcd_JobMap jcd)+ jd' = addOutputToJob (td_Output td) jd + jcd' = updateJob jd' jcd +++handleTasks :: JobController -> IO ()+handleTasks jc+ = modifyMVar jc $+ \jcd ->+ do+ infoM cycleLogger "processing Tasks:"+ -- get all runnning jobs (not idle...)+ let runningJobs = getJobIds [JSSplit, JSMap, JSCombine, JSReduce] jcd+ infoM cycleLogger $ "running Jobs:" ++ show runningJobs+ -- get all idle tasks+ let idleTasks = getTaskIds runningJobs [] [TSIdle] jcd+ infoM cycleLogger $ "idle Tasks:" ++ show idleTasks+ let idleTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd)) idleTasks+ + -- set these tasks to sending, so they won't be called again while the sending process+ let jcd1 = foldl toSendingTaskState jcd idleTaskDatas + -- get all sending taskdatas + let sendingTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd1)) idleTasks+ -- send all idle Tasks (non-blocking)+ mapM (sendTask jc) sendingTaskDatas+ + -- get all completed Tasks+ let completedTasks = getTaskIds runningJobs [] [TSCompleted] jcd1+ let completedTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd1)) completedTasks+ -- inProgressTasks = getTaskIds runningJobs [] [TSInProgress] jcd+ let jcd2 = foldl finishTask jcd1 completedTaskDatas+ return (jcd2,())+++-- ----------------------------------------------------------------------------+-- Process Jobs+-- ----------------------------------------------------------------------------++toNextJobState :: JobControllerData -> JobData -> JobControllerData+toNextJobState jcd jd = changeJobState (jd_JobId jd) (getNextJobState (jd_State jd)) jcd++toIdleTaskState :: JobControllerData -> TaskData -> JobControllerData+toIdleTaskState jcd td = changeTaskState (td_TaskId td) TSIdle jcd++toInProgressTaskState :: JobControllerData -> TaskData -> JobControllerData+toInProgressTaskState jcd td = changeTaskState (td_TaskId td) TSInProgress jcd++toCompletedTaskState :: JobControllerData -> TaskData -> JobControllerData+toCompletedTaskState jcd td = updateTaskOutput tid o $ changeTaskState tid TSCompleted jcd+ where+ tid = td_TaskId td+ o = td_Output td++toErrorTaskState :: JobControllerData -> TaskData -> JobControllerData+toErrorTaskState jcd td = changeTaskState (td_TaskId td) TSError jcd+++hasPhase :: JobData -> Bool+hasPhase jd = isJust $ getCurrentTaskAction jd+++getCurrentTaskAction :: JobData -> Maybe ActionName+getCurrentTaskAction jd = getTaskAction' (jd_Info jd) (jd_State jd)+ where+ getTaskAction' ji JSSplit = maybe Nothing (Just . ja_Name) (ji_SplitAction ji)+ getTaskAction' ji JSMap = maybe Nothing (Just . ja_Name) (ji_MapAction ji)+ getTaskAction' ji JSCombine = maybe Nothing (Just . ja_Name) (ji_CombineAction ji)+ getTaskAction' ji JSReduce = maybe Nothing (Just . ja_Name) (ji_ReduceAction ji) + getTaskAction' _ _ = Nothing+++getCurrentTaskOutputType :: JobData -> TaskOutputType+getCurrentTaskOutputType jd = getTaskOutputType' (jd_Info jd) (jd_State jd)+ where+ getTaskOutputType' ji JSSplit = maybe TOTFile (ja_Output) (ji_SplitAction ji)+ getTaskOutputType' ji JSMap = maybe TOTFile (ja_Output) (ji_MapAction ji)+ getTaskOutputType' ji JSCombine = maybe TOTFile (ja_Output) (ji_CombineAction ji)+ getTaskOutputType' ji JSReduce = maybe TOTRawTuple (ja_Output) (ji_ReduceAction ji)+ getTaskOutputType' _ _ = TOTFile+++getCurrentTaskPartValue :: JobState -> JobInfo -> Maybe Int+getCurrentTaskPartValue state ji = gTPV state+ where+ gTPV JSIdle = maybe (gTPV JSSplit) (Just . ja_Count) (ji_SplitAction ji)+ gTPV JSSplit = maybe (gTPV JSMap) (Just . ja_Count) (ji_MapAction ji)+ gTPV JSMap = maybe (gTPV JSCombine) (\_ -> Nothing) (ji_CombineAction ji)+ gTPV JSCombine = maybe (gTPV JSReduce) (Just . ja_Count) (ji_ReduceAction ji)+ gTPV JSReduce = ji_NumOfResults ji+ gTPV _ = Nothing+ ++createTasks :: JobControllerData -> JobData -> IO JobControllerData+createTasks jcd jd+ = do+ let state = jd_State jd+ info = jd_Info jd+ outputMap = (jd_OutputMap jd)+ -- our input is the output of the previous state+ let inputAccu = maybe (AMap.empty) (id) $ Map.lookup (getPrevJobState state) outputMap+ if (hasPhase jd)+ then do+ let inputList = AMap.toList inputAccu + let jid = jd_JobId jd+ let a = fromJust $ getCurrentTaskAction jd+ let ot = getCurrentTaskOutputType jd+ let opts = ji_Option info+ let n = getCurrentTaskPartValue state info+ let tt = fromJust $ fromJobStatetoTaskType state+ -- create new tasks+ (jcd', taskDatas) <- mapAccumLM (\d i -> newTaskData d jid tt TSIdle opts n i a ot) jcd inputList+ -- add task to controller+ let jcd'' = foldl addTask jcd' taskDatas + return jcd''+ else do+ -- there is no action in this state, so we copy the input to the output+ let outputMap' = Map.insert state inputAccu outputMap + let jd' = jd { jd_OutputMap = outputMap' }+ return $ updateJob jd' jcd+++createResults :: JobControllerData -> JobData -> IO JobControllerData+createResults jcd jd+ = do+ let outputAccu = maybe (AMap.empty) (id) $ Map.lookup JSCompleted (jd_OutputMap jd)+ let outputList = AMap.toList outputAccu+ let (JobResultContainer mVarResult) = jd_Result jd+ let res = JobResult $ concat $ map (snd) outputList+ putMVar mVarResult res+ return jcd +++handleJobs :: JobController -> IO ()+handleJobs jc+ = modifyMVar jc $+ \jcd ->+ do+ -- get all working jobs (idle or running)+ infoM cycleLogger "processing Jobs"+ let workingJobs = getJobIds [JSIdle, JSSplit, JSMap, JSCombine, JSReduce] jcd+ infoM cycleLogger $ "working jobs:" ++ show workingJobs+ -- process only jobs, whose current phase is done+ let jobsWithoutTasks = filter (allTasksFinished jcd) workingJobs+ infoM cycleLogger $ "jobsWithoutTasks:" ++ show jobsWithoutTasks+ -- get the old jobdatas+ let oldJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd)) jobsWithoutTasks+ debugM cycleLogger $ "oldJobDatas:\n" ++ show oldJobDatas+ -- change the job states to the next phase+ let jcd1 = foldl toNextJobState jcd oldJobDatas+ -- get the new jobdatas+ let newJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd1)) jobsWithoutTasks+ debugM cycleLogger $ "newJobDatas:\n" ++ show newJobDatas+ -- create new tasks for each Job+ jcd2 <- foldM createTasks jcd1 newJobDatas+ + -- get all completed Jobs+ let completedJobs = getJobIds [JSCompleted] jcd2+ infoM cycleLogger $ "completed Jobs:" ++ show completedJobs+ -- get the completed JobDatas+ let completedJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd2)) completedJobs+ -- change the job states to the next phase+ let jcd3 = foldl toNextJobState jcd2 completedJobDatas+ -- get the new jobdatas+ let finishedJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd3)) completedJobs+ -- create new tasks for each Job+ jcd4 <- foldM createResults jcd3 finishedJobDatas+ + return (jcd4, ())+++++-- ---------------------------------------------------------------------------- +-- handling Task-Responses+-- ----------------------------------------------------------------------------+++setTaskIdle :: JobController -> TaskData -> IO ()+setTaskIdle jc td+ = do+ debugM localLogger $ "setTaskIdle: waiting... TaskId: " ++ show (td_TaskId td)+ modifyMVar jc $+ \jcd ->+ do+ debugM localLogger $ "setTaskIdle: setting... TaskId: " ++ show (td_TaskId td)+ let jcd' = toIdleTaskState jcd td+ return (jcd', ())+++setTaskInProgress :: JobController -> TaskData -> IO ()+setTaskInProgress jc td+ = do+ debugM localLogger $ "setTaskInProgress: waiting... TaskId: " ++ show (td_TaskId td)+ modifyMVar jc $+ \jcd ->+ do+ debugM localLogger $ "setTaskInProgress: setting... TaskId: " ++ show (td_TaskId td)+ let jcd' = toInProgressTaskState jcd td+ return (jcd', ())+++setTaskCompleted :: JobController -> TaskData -> IO ()+setTaskCompleted jc td+ = do+ debugM localLogger $ "setTaskCompleted: waiting... TaskId: " ++ show (td_TaskId td)+ modifyMVar jc $+ \jcd ->+ do+ debugM localLogger $ "setTaskCompleted: setting... TaskId: " ++ show (td_TaskId td)+ let jcd' = toCompletedTaskState jcd td+ return (jcd', ())++ +setTaskError :: JobController -> TaskData -> IO ()+setTaskError jc td+ = do+ debugM localLogger $ "setTaskError: waiting... TaskId: " ++ show (td_TaskId td)+ modifyMVar jc $+ \jcd ->+ do+ debugM localLogger $ "setTaskError: setting... TaskId: " ++ show (td_TaskId td)+ let jcd' = toErrorTaskState jcd td+ return (jcd', ())+
+ source/Holumbus/MapReduce/MapReduce.hs view
@@ -0,0 +1,75 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.MapReduce.MapReduce+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.MapReduce.MapReduce+(+ MapReduceType(..)+, MapReduce(..)+)+where++import Holumbus.Common.Debug+import Holumbus.FileSystem.FileSystem+import Holumbus.MapReduce.Types+import Holumbus.Network.Site+++data MapReduceType = MRTMaster | MRTWorker | MRTClient | MRTStandalone+ deriving (Show, Eq, Ord, Enum)+ ++class (Debug mr) => MapReduce mr where+ + closeMapReduce :: mr -> IO ()+ + -- | prints the siteId of the MapReduce instance + getMySiteId :: mr -> IO SiteId+ + -- | get the Type of the MapReduce instance+ getMapReduceType :: mr -> IO MapReduceType+ + -- | get the Controlling-Type (normal or singlestep) of the MapReduce instance+ startControlling :: mr -> IO ()++ -- | get the Controlling-Type (normal or singlestep) of the MapReduce instance+ stopControlling :: mr -> IO ()++ -- | test, if Controller is running+ isControlling :: mr -> IO Bool+ + -- | performs a single step of the controller (if mode is singlestep)+ doSingleStep :: mr -> IO ()+ + -- | starts a MapReduce-Job (blocking while finished)+ doMapReduceJob :: JobInfo -> mr -> IO JobResult+ + doMapReduce+ :: ActionConfiguration a k1 v1 k2 v2 v3 v4+ -> a -- ^ options+ -> [(k1,v1)] -- ^ input (Tuples)+ -> [FileId] -- ^ input (Files)+ -> Int -- ^ number of splitters+ -> Int -- ^ number of mappers+ -> Int -- ^ number of reducers+ -> Int -- ^ number of results+ -> TaskOutputType -- ^ type of the result (file of raw)+ -> mr -> IO ([(k2,v4)],[FileId])+ doMapReduce c o ls1 ls2 sc mc rc nor rt mr+ = do+ let ji = createJobInfoFromConfiguration c o ls1 ls2 sc mc rc nor rt+ jr <- doMapReduceJob ji mr+ let res = createListsFromJobResult c jr+ return res+
+ source/Holumbus/MapReduce/TaskProcessor.hs view
@@ -0,0 +1,545 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.MapReduce.TaskProcessor+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+module Holumbus.MapReduce.TaskProcessor+ (+ -- * Datatypes+ TaskResultFunction+ , TaskProcessor++ , printTaskProcessor++ -- * Creation and Destruction+ , newTaskProcessor+ , closeTaskProcessor+ , setFileSystemToTaskProcessor+ , setActionMap+ , setTaskCompletedHook + , setTaskErrorHook++ -- * TaskProcessor + , startTaskProcessor+ , stopTaskProcessor++ -- * Info an Debug+ , listTaskIds + , getActions+ , getActionNames++ -- * Task Creation and Destruction+ , startTask+ , stopTask+ , stopAllTasks + )+where++import Prelude hiding ( catch )++import Control.Exception ( Exception+ , throw+ , throwTo+ , catch+ )+import Control.Concurrent++-- import Data.Binary+-- import qualified Data.ByteString.Lazy as B++import qualified Data.Map as Map+import qualified Data.Set as Set+import Data.Maybe+import Data.Typeable++import System.Log.Logger++import Holumbus.Common.Utils ( handleAll )+import qualified Holumbus.Data.KeyMap as KMap+import Holumbus.MapReduce.Types+import qualified Holumbus.FileSystem.FileSystem as FS++++localLogger :: String+localLogger = "Holumbus.MapReduce.TaskProcessor"+++taskLogger :: String+taskLogger = "Holumbus.MapReduce.TaskProcessor.task"++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------++-- | a function for responding a+type TaskResultFunction = TaskData -> IO Bool++dummyTaskResultFunction :: TaskData -> IO Bool+dummyTaskResultFunction _ = return True ++data TaskProcessorFunctions+ = TaskProcessorFunctions { tpf_TaskCompleted :: TaskResultFunction+ , tpf_TaskError :: TaskResultFunction+ }++instance Show TaskProcessorFunctions where+ show _ = "{TaskProcessorFunctions}"++data TaskProcessorException + = KillServerException+ deriving (Show, Typeable)++instance Exception TaskProcessorException where++data TaskException+ = KillTaskException+ | UnkownTaskException ActionName+ deriving (Show, Typeable)++instance Exception TaskException where++-- | data, needed by the MapReduce-System to run the tasks+data TaskProcessorData = TaskProcessorData {+ -- internal+ tpd_ServerThreadId :: Maybe ThreadId+ , tpd_ServerDelay :: Int+ , tpd_FileSystem :: FS.FileSystem+ -- configuration+ , tpd_MaxTasks :: Int+ , tpd_Functions :: TaskProcessorFunctions+ , tpd_ActionMap :: ActionMap+ -- task processing+ , tpd_TaskQueue :: [TaskData]+ , tpd_CompletedTasks :: Set.Set TaskData+ , tpd_ErrorTasks :: Set.Set TaskData+ , tpd_TaskIdThreadMap :: Map.Map TaskId ThreadId+ } deriving (Show)+++type TaskProcessor = MVar TaskProcessorData+++printTaskProcessor :: TaskProcessor -> IO String+printTaskProcessor tp+ = withMVar tp $ \tpd -> return $ show tpd++++-- ----------------------------------------------------------------------------+-- Creation / Destruction+-- ----------------------------------------------------------------------------++defaultTaskProcessorData :: TaskProcessorData+defaultTaskProcessorData = tpd+ where+ funs = TaskProcessorFunctions + dummyTaskResultFunction+ dummyTaskResultFunction+ tpd = TaskProcessorData+ Nothing+ 1000 -- one millisecond delay+ undefined+ 1+ funs+ KMap.empty+ []+ Set.empty+ Set.empty+ Map.empty +++-- | creates a new TaskProcessor+newTaskProcessor :: IO TaskProcessor+newTaskProcessor+ = do+ let tpd = defaultTaskProcessorData+ tp <- newMVar tpd+ -- do not start this, perhaps the user want to change something first+ -- startTaskProcessor tp+ return tp+++closeTaskProcessor :: TaskProcessor -> IO ()+closeTaskProcessor tp+ = do+ stopTaskProcessor tp+++-- | add a filesystem-instance to the TaskProcessor+setFileSystemToTaskProcessor :: FS.FileSystem -> TaskProcessor -> IO ()+setFileSystemToTaskProcessor fs tp+ = modifyMVar tp $+ \tpd -> return $ (tpd { tpd_FileSystem = fs }, ())+ ++-- | adds an ActionMap to the TaskProcessor+setActionMap :: KMap.KeyMap ActionData -> TaskProcessor -> IO ()+setActionMap m tp+ = modifyMVar tp $+ \tpd -> return $ (tpd { tpd_ActionMap = m }, ())+++setTaskCompletedHook :: TaskResultFunction -> TaskProcessor -> IO () +setTaskCompletedHook f tp+ = modifyMVar tp $+ \tpd ->+ do+ let funs = tpd_Functions tpd+ let funs' = funs { tpf_TaskCompleted = f }+ return (tpd { tpd_Functions = funs' }, ())+++setTaskErrorHook :: TaskResultFunction -> TaskProcessor -> IO ()+setTaskErrorHook f tp+ = modifyMVar tp $+ \tpd ->+ do+ let funs = tpd_Functions tpd+ let funs' = funs { tpf_TaskError = f }+ return (tpd { tpd_Functions = funs' }, ())++-- ----------------------------------------------------------------------------+-- server functions+-- ----------------------------------------------------------------------------++startTaskProcessor :: TaskProcessor -> IO ()+startTaskProcessor tp+ = do+ modifyMVar tp $ + \tpd -> + do+ thd <- case (tpd_ServerThreadId tpd) of+ (Just i) -> return i+ (Nothing) ->+ do+ i <- forkIO $ doProcessing tp+ return i+ return (tpd {tpd_ServerThreadId = (Just thd)}, ())++++stopTaskProcessor :: TaskProcessor -> IO ()+stopTaskProcessor tp+ = do+ modifyMVar tp $ + \tpd -> + do+ case (tpd_ServerThreadId tpd) of+ (Nothing) -> return ()+ (Just i) -> + do+ throwTo i KillServerException+ yield+ return ()+ return (tpd {tpd_ServerThreadId = Nothing}, ())+++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------++containsTask :: TaskId -> TaskProcessorData -> Bool+containsTask tid tpd = isTaskRunning || isTaskQueued+ where+ isTaskRunning = Map.member tid (tpd_TaskIdThreadMap tpd)+ isTaskQueued = any (\td -> (td_TaskId td) == tid) (tpd_TaskQueue tpd)+++queueTask :: TaskData -> TaskProcessorData -> TaskProcessorData+queueTask td tpd = tpd { tpd_TaskQueue = q' }+ where+ q = tpd_TaskQueue tpd+ q' = q ++ [td]++ +dequeueTask :: TaskId -> TaskProcessorData -> TaskProcessorData+dequeueTask tid tpd = tpd { tpd_TaskQueue = q' }+ where+ q = tpd_TaskQueue tpd+ q' = filter (\td -> (td_TaskId td) /= tid) q +++getTaskThreadId :: TaskId -> TaskProcessorData -> Maybe ThreadId+getTaskThreadId tid tpd = Map.lookup tid (tpd_TaskIdThreadMap tpd)+++getTasksIds :: TaskProcessorData -> [TaskId]+getTasksIds tpd = Set.toList $ Set.union (Set.fromList qs) (Set.fromList ts) + where+ qs = map (\td -> td_TaskId td) (tpd_TaskQueue tpd)+ ts = Map.keys (tpd_TaskIdThreadMap tpd)+++addTask :: TaskData -> TaskProcessorData -> TaskProcessorData+addTask td tpd = if containsTask tid tpd then tpd else queueTask td tpd+ where+ tid = td_TaskId td+++deleteTask :: TaskId -> TaskProcessorData -> TaskProcessorData+deleteTask tid tpd = dequeueTask tid tpd'+ where+ tpd' = tpd { tpd_TaskIdThreadMap = ttm' }+ ttm = tpd_TaskIdThreadMap tpd+ ttm' = Map.delete tid ttm++++-- ----------------------------------------------------------------------------+-- Info an Debug+-- ----------------------------------------------------------------------------+++listTaskIds :: TaskProcessor -> IO [TaskId] +listTaskIds tp+ = withMVar tp $+ \tpd -> return $ getTasksIds tpd+++-- | Lists all Actions with Name, Descrition and so on+getActions :: TaskProcessor -> IO [ActionData]+getActions tp+ = withMVar tp $+ \tpd -> return $ KMap.elems (tpd_ActionMap tpd)+++-- | Lists all Names of the Actions+getActionNames :: TaskProcessor -> IO [ActionName]+getActionNames tp+ = do+ actions <- getActions tp+ return $ map (\a -> ad_Name a) actions+++-- ----------------------------------------------------------------------------+-- Task Controlling+-- ----------------------------------------------------------------------------+++-- | adds a Task to the TaskProcessor, the execution might be later+startTask :: TaskData -> TaskProcessor -> IO ()+startTask td tp+ = do+ debugM localLogger $ "waiting to add Task " ++ show (td_TaskId td)+ modifyMVar tp $+ \tpd->+ do+ debugM localLogger $ "adding Task " ++ show (td_TaskId td)+ return (addTask td tpd, ())+ + +stopTask :: TaskId -> TaskProcessor -> IO ()+stopTask tid tp+ = do+ debugM localLogger $ "waiting to stop Task " ++ show tid+ mthd <- modifyMVar tp $+ \tpd-> + do+ let thd = getTaskThreadId tid tpd + return (deleteTask tid tpd, thd)+ debugM localLogger $ "stopping Task " ++ show tid+ maybe (return ()) (\thd -> throwTo thd KillTaskException) mthd+++stopAllTasks :: TaskProcessor -> IO () +stopAllTasks tp+ = do+ debugM localLogger $ "waiting to stop all Tasks"+ tids <- withMVar tp $ \tpd -> return $ getTasksIds tpd+ mapM (\tid -> stopTask tid tp) tids+ return ()++++++-- ----------------------------------------------------------------------------+-- Task Processing+-- ----------------------------------------------------------------------------+++setTaskCompleted :: TaskData -> TaskProcessorData -> TaskProcessorData+setTaskCompleted td tpd = tpd { tpd_TaskIdThreadMap = ttm', tpd_CompletedTasks = ct' }+ where+ tid = td_TaskId td+ ttm' = Map.delete tid (tpd_TaskIdThreadMap tpd)+ ct' = Set.insert td (tpd_CompletedTasks tpd)+++setTaskError :: TaskData -> TaskProcessorData -> TaskProcessorData+setTaskError td tpd = tpd { tpd_TaskIdThreadMap = ttm', tpd_ErrorTasks = et' }+ where+ tid = td_TaskId td+ ttm' = Map.delete tid (tpd_TaskIdThreadMap tpd)+ et' = Set.insert td (tpd_ErrorTasks tpd)+++-- | mark the task as error and invoke the reply function+reportErrorTask :: TaskData -> TaskProcessor -> IO ()+reportErrorTask td tp+ = modifyMVar tp $+ \tpd -> + do + let tpd' = setTaskError td tpd+ return (tpd', ())+++-- | mark the task as completed and invoke the reply function+reportCompletedTask :: TaskData -> TaskProcessor -> IO ()+reportCompletedTask td tp+ = modifyMVar tp $+ \tpd -> + do + let tpd' = setTaskCompleted td tpd+ return (tpd', ())+++setTaskRunning :: TaskId -> ThreadId -> TaskProcessorData -> TaskProcessorData+setTaskRunning tid thd tpd = tpd { tpd_TaskIdThreadMap = ttm' }+ where+ ttm = tpd_TaskIdThreadMap tpd+ ttm' = Map.insert tid thd ttm+++getNextQueuedTask :: TaskProcessorData -> (Maybe TaskData, TaskProcessorData)+getNextQueuedTask tpd = (td , tpd { tpd_TaskQueue = q' })+ where+ q = tpd_TaskQueue tpd+ q' = if null q then q else tail q+ td = if null q then Nothing else Just $ head q++-- ----------------------------------------------------------------------------+--+-- ----------------------------------------------------------------------------+++doProcessing :: TaskProcessor -> IO ()+doProcessing tp+ = do+ catch (doProcessing' tp)+ handler+ where+ handler :: TaskProcessorException -> IO ()+ handler err = putStrLn (show err)+ doProcessing' tp'+ = do+ handleNewTasks tp'+ handleFinishedTasks tp'+ delay <- withMVar tp' (\tpd -> return $ tpd_ServerDelay tpd)+ threadDelay delay+ doProcessing' tp'+++handleNewTasks :: TaskProcessor -> IO ()+handleNewTasks tp+ = do+ modifyMVar tp $+ \tpd ->+ do+ tpd' <- handleNewTasks' tpd+ return (tpd',())+ where+ handleNewTasks' tpd + = do+ -- we can only start new tasks, if there are any left...+ let maxTasks = tpd_MaxTasks tpd+ let runningTasks = Map.size (tpd_TaskIdThreadMap tpd)+ let moreTasks = not $ null (tpd_TaskQueue tpd)+ if (moreTasks && (runningTasks < maxTasks)) + then do+ -- take the task from the queue+ let (mtd, tpd') = getNextQueuedTask tpd+ let td = fromJust mtd+ -- start its thread+ thd <- runTask td tp+ -- save it+ let tpd'' = setTaskRunning (td_TaskId td) thd tpd'+ -- try to start more+ handleNewTasks' tpd''+ else do+ return tpd+++runTask :: TaskData -> TaskProcessor -> IO ThreadId+runTask td tp+ = do+ -- spawn a new thread for each tasks+ forkIO $ + handleAll ( \ e -> do+ errorM taskLogger $ "Exception: " ++ show e+ reportErrorTask td tp+ ) $ + do yield+ td' <- performTask td tp+ reportCompletedTask td' tp++-- not used, because we are doi +handleFinishedTasks :: TaskProcessor -> IO ()+handleFinishedTasks tp+ = do+ modifyMVar tp $+ \tpd ->+ do+ cts' <- sendTasksResults (tpd_CompletedTasks tpd) (tpf_TaskCompleted $ tpd_Functions tpd)+ ets' <- sendTasksResults (tpd_ErrorTasks tpd) (tpf_TaskError $ tpd_Functions tpd)+ let tpd' = tpd { tpd_CompletedTasks = cts', tpd_ErrorTasks = ets' }+ return (tpd', ()) ++ +sendTasksResults :: Set.Set TaskData -> TaskResultFunction -> IO (Set.Set TaskData)+sendTasksResults set fun+ = do+ let ls = Set.toList set+ sendResults <- mapM fun ls + let (failures,_) = unzip $ filter (\(_,b) -> not b) $ zip ls sendResults+ return $ Set.fromList failures++-- ----------------------------------------------------------------------------+-- Performing a Task+-- ----------------------------------------------------------------------------++-- | doing a map task+performTask :: TaskData -> TaskProcessor-> IO TaskData+performTask td tp+ = do+ infoM taskLogger $ "Task " ++ show (td_TaskId td)+ debugM taskLogger $ "input td: " ++ show td+ + -- get all functions+ (ad, parts, bin, fs, opt) <- withMVar tp $+ \tpd ->+ do+ let action = KMap.lookup (td_Action td) (tpd_ActionMap tpd)+ let parts = (td_PartValue td)+ let input = (td_Input td)+ let option = (td_Option td)+ let filesystem = (tpd_FileSystem tpd)+ return (action, parts, input, filesystem, option)+ + case ad of+ (Nothing) ->+ -- TODO throw execption here+ return td+ (Just a) ->+ do+ action <- case (getActionForTaskType (td_Type td) a) of+ (Just a') -> return a'+ (Nothing) -> throw (UnkownTaskException $ td_Action td)+ + let env = mkActionEnvironment td fs+ bout <- action env opt parts bin+ let td' = td { td_Output = bout }+ debugM taskLogger $ "output td: " ++ show td'+ return td'
+ source/Holumbus/MapReduce/Types.hs view
@@ -0,0 +1,961 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.MapReduce.Types+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++{-# OPTIONS -fglasgow-exts #-}+{-# LANGUAGE Arrows, NoMonomorphismRestriction #-}+module Holumbus.MapReduce.Types+(+ FunctionData(..)+ +{-+, encodeTuple+, decodeTuple+, decodeTupleList+, encodeTupleList+-}++-- * TaskData+, TaskId+, TaskType(..)+, TaskState(..)+, getNextTaskState+, TaskOutputType(..)+, TaskData(..)++-- * JobData+, JobId+, JobState(..)+, getNextJobState+, getPrevJobState+, fromJobStatetoTaskType+, OutputMap+, JobAction(..)+, JobInfo(..)+, JobData(..)+, JobResultContainer(..)+, JobResult(..)+++-- * TaskAction+, ActionName+, ActionInfo++, ActionEnvironment(..)+, mkActionEnvironment++, InputReader+, OutputWriter++, OptionsDecoder++, ActionConfiguration(..)+, SplitConfiguration(..)+, MapConfiguration(..)+, ReduceConfiguration(..)+, defaultActionConfiguration+, defaultSplitConfiguration+, defaultMapConfiguration+, defaultReduceConfiguration+, readActionConfiguration+, createJobInfoFromConfiguration+, createListsFromJobResult++, ActionData(..)+, getActionForTaskType+, ActionMap++-- * MapAction+, MapAction+-- , BinaryMapAction +, MapFunction+, MapPartition++-- * Combine-Reduce-Action+, ReduceAction+-- , BinaryReduceAction+, ReduceMerge +, ReduceFunction+, ReducePartition+)+where++import Control.Concurrent+import Data.Binary+import qualified Data.ByteString.Lazy as B+import qualified Data.Map as Map+import qualified Data.HashTable as Hash+import Data.Maybe+import Data.Time++import System.Log.Logger++import Text.XML.HXT.Arrow++import qualified Holumbus.Data.AccuMap as AMap+import Holumbus.Common.FileHandling+import Holumbus.Common.Utils+import qualified Holumbus.Data.AccuMap as AMap+import qualified Holumbus.Data.KeyMap as KMap+import qualified Holumbus.FileSystem.FileSystem as FS+++localLogger :: String+localLogger = "Holumbus.MapReduce.Types"++-- ----------------------------------------------------------------------------+-- general datatypes+-- ----------------------------------------------------------------------------+++data FunctionData+ = TupleFunctionData B.ByteString+ | FileFunctionData FS.FileId+ deriving (Show, Eq, Ord)++instance Binary FunctionData where+ put (TupleFunctionData t) = putWord8 0 >> put t+ put (FileFunctionData f) = putWord8 1 >> put f+ get+ = do+ t <- getWord8+ case t of+ 1 -> get >>= \f -> return (FileFunctionData f)+ _ -> get >>= \b -> return (TupleFunctionData b)+ + +instance XmlPickler FunctionData where+ xpickle = xpFunctionData+ +xpFunctionData :: PU FunctionData+xpFunctionData+ = xpElem "data" $+ xpAlt tag ps+ where+ tag (TupleFunctionData _) = 0+ tag (FileFunctionData _) = 1+ ps = [xpWrap (\p -> TupleFunctionData (encodeTuple p), \(TupleFunctionData p) -> decodeTuple p) xpRawFunctionData+ ,xpWrap (\f -> FileFunctionData f, \(FileFunctionData f) -> f) xpFileFunctionData ]+ xpRawFunctionData = xpPair (xpElem "key" xpText0) (xpElem "value" xpText0)+ xpFileFunctionData = xpElem "filename" xpText+++ +-- ----------------------------------------------------------------------------+-- general encoding / decoding+-- ----------------------------------------------------------------------------++encodeTuple :: (Binary k, Binary v) => (k, v) -> B.ByteString+encodeTuple t = encode t+++decodeTuple :: (Binary k, Binary v) => B.ByteString -> (k, v)+decodeTuple t = decode t+++-- encodeTupleList :: (Binary k, Binary v) => [(k, v)] -> [B.ByteString]+-- encodeTupleList ls = map encodeTuple ls+++-- decodeTupleList :: (Binary k, Binary v) => [B.ByteString] -> [(k, v)]+-- decodeTupleList ls = map decodeTuple ls+++++-- ----------------------------------------------------------------------------+-- Task DataTypes+-- ----------------------------------------------------------------------------+++-- | the task id (should be unique in the system)+type TaskId = Integer+++-- | which type (map, combine, reduce)+data TaskType = TTSplit | TTMap | TTCombine | TTReduce | TTError+ deriving (Show, Eq, Ord)++instance Binary TaskType where+ put (TTSplit) = putWord8 1+ put (TTMap) = putWord8 2+ put (TTCombine) = putWord8 3+ put (TTReduce) = putWord8 4+ put (TTError) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> return (TTSplit)+ 2 -> return (TTMap)+ 3 -> return (TTCombine)+ 4 -> return (TTReduce)+ _ -> return (TTError)+++-- | the task state+data TaskState = TSIdle | TSSending | TSInProgress | TSCompleted | TSFinished | TSError+ deriving (Show, Eq, Ord, Enum)++instance Binary TaskState where+ put (TSIdle) = putWord8 1+ put (TSSending) = putWord8 2+ put (TSInProgress) = putWord8 3+ put (TSCompleted) = putWord8 4+ put (TSFinished) = putWord8 5+ put (TSError) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> return (TSIdle)+ 2 -> return (TSSending)+ 3 -> return (TSInProgress)+ 4 -> return (TSCompleted)+ 5 -> return (TSFinished)+ _ -> return (TSError)++getNextTaskState :: TaskState -> TaskState+getNextTaskState TSError = TSError+getNextTaskState TSFinished = TSFinished+getNextTaskState s = succ s+++data TaskOutputType = TOTRawTuple | TOTFile+ deriving (Show, Read, Eq, Ord, Enum)+ +instance Binary TaskOutputType where+ put (TOTRawTuple) = putWord8 0+ put (TOTFile) = putWord8 1+ get+ = do+ t <- getWord8+ case t of+ 0 -> return (TOTRawTuple)+ _ -> return (TOTFile)++instance XmlPickler TaskOutputType where+ xpickle = xpAttr "output" $ xpPrim+++-- | the TaskData, contains all information to do the task+data TaskData = TaskData {+ td_JobId :: ! JobId+ , td_TaskId :: ! TaskId+ , td_Type :: ! TaskType+ , td_State :: ! TaskState+ , td_Option :: ! B.ByteString+ , td_PartValue :: ! (Maybe Int) + , td_Input :: ! (Int, [FunctionData])+ , td_Output :: ! [(Int,[FunctionData])]+ , td_OutputType :: ! TaskOutputType+ , td_Action :: ! ActionName+ } deriving (Show, Eq, Ord)++instance Binary TaskData where+ put (TaskData jid tid tt ts opt pv i o ot a)+ = put jid >> put tid >> put tt >> put ts >> put opt >> put pv >> put i >> put o >> put ot >> put a+ get+ = do+ jid <- get+ tid <- get+ tt <- get+ ts <- get+ opt <- get+ pv <- get+ i <- get+ o <- get+ ot <- get+ a <- get+ return (TaskData jid tid tt ts opt pv i o ot a)+++++-- ----------------------------------------------------------------------------+-- Job Datatypes+-- ----------------------------------------------------------------------------+++-- | the job id (should be unique in the system)+type JobId = Integer+++-- | the job state+data JobState = JSPlanned | JSIdle | JSSplit | JSMap | JSCombine | JSReduce | JSCompleted | JSFinished | JSError+ deriving(Show, Eq, Ord, Enum)++instance Binary JobState where+ put (JSPlanned) = putWord8 1+ put (JSIdle) = putWord8 2+ put (JSSplit) = putWord8 3+ put (JSMap) = putWord8 4+ put (JSCombine) = putWord8 5+ put (JSReduce) = putWord8 6+ put (JSCompleted) = putWord8 7+ put (JSFinished) = putWord8 8+ put (JSError) = putWord8 0+ get+ = do+ t <- getWord8+ case t of+ 1 -> return (JSPlanned)+ 2 -> return (JSIdle)+ 3 -> return (JSSplit)+ 4 -> return (JSMap)+ 5 -> return (JSCombine)+ 6 -> return (JSReduce)+ 7 -> return (JSCompleted)+ 8 -> return (JSFinished)+ _ -> return (JSError)++getNextJobState :: JobState -> JobState+getNextJobState JSError = JSError+getNextJobState JSFinished = JSFinished+getNextJobState s = succ s++getPrevJobState :: JobState -> JobState+getPrevJobState JSIdle = JSIdle+getPrevJobState JSError = JSError+getPrevJobState s = pred s++fromJobStatetoTaskType :: JobState -> Maybe TaskType+fromJobStatetoTaskType JSSplit = Just TTSplit+fromJobStatetoTaskType JSMap = Just TTMap+fromJobStatetoTaskType JSCombine = Just TTCombine+fromJobStatetoTaskType JSReduce = Just TTReduce+fromJobStatetoTaskType _ = Nothing++++type OutputMap = Map.Map JobState (AMap.AccuMap Int FunctionData)++data JobAction = JobAction {+ ja_Name :: ! ActionName+ , ja_Output :: ! TaskOutputType+ , ja_Count :: ! Int+ } deriving (Show, Eq)++instance Binary JobAction where+ put (JobAction n o c) = put n >> put o >> put c+ get + = do+ n <- get+ o <- get+ c <- get+ return (JobAction n o c)++instance XmlPickler JobAction where+ xpickle = xpJobAction+ +xpJobAction :: PU JobAction+xpJobAction =+ xpWrap+ (\(n, o, c) -> JobAction n o c, + \(JobAction n o c) -> (n, o, c)) xpAction+ where+ xpAction = xpTriple (xpFunction) (xpickle) (xpCount)+ xpFunction = xpAttr "name" xpText+ xpCount = xpWrap ((maybe 1 id), Just) $ xpOption $ xpAttr "count" xpickle+ +++-- | defines a job, this is all data the user has to give to run a job+data JobInfo = JobInfo {+ ji_Description :: ! String+ , ji_Option :: ! B.ByteString+ , ji_SplitAction :: ! (Maybe JobAction)+ , ji_MapAction :: ! (Maybe JobAction)+ , ji_CombineAction :: ! (Maybe JobAction)+ , ji_ReduceAction :: ! (Maybe JobAction)+ , ji_NumOfResults :: ! (Maybe Int)+ , ji_Input :: ! [FunctionData]+ } deriving (Show, Eq)++instance Binary JobInfo where+ put (JobInfo d opt sa ma ca ra nor i)+ = put d >> put opt >> + put sa >> put ma >> put ca >> put ra >> put nor >> put i+ get+ = do+ d <- get+ opt <- get+ sa <- get+ ma <- get+ ca <- get+ ra <- get+ nor <- get+ i <- get+ return (JobInfo d opt sa ma ca ra nor i)+ +instance XmlPickler JobInfo where+ xpickle = xpJobInfo+ +xpJobInfo :: PU JobInfo+xpJobInfo = + xpElem "jobInfo" $+ xpWrap+ (\(n, (sa,ma,ca,ra), nor, i) -> JobInfo n (encode ()) sa ma ca ra nor i, + \(JobInfo n _ sa ma ca ra nor i) -> (n, (sa,ma,ca,ra), nor, i)) xpJob+ where+ xpJob = + xp4Tuple+ (xpAttr "name" xpText0)+ (xpActions)+ (xpCount)+ (xpFunctionDataList)+ xpActions = xp4Tuple (xpSplitAction) (xpMapAction) (xpCombineAction) (xpReduceAction)+ xpSplitAction = xpOption $ xpElem "split" $ xpJobAction+ xpMapAction = xpOption $ xpElem "map" $ xpJobAction+ xpCombineAction = xpOption $ xpElem "combine" $ xpJobAction+ xpReduceAction = xpOption $ xpElem "reduce" $ xpJobAction+ xpFunctionDataList+ = xpWrap (filterEmptyList,setEmptyList) $ xpOption $+ xpElem "inputList" $ xpList xpFunctionData+ xpCount = xpOption $ xpAttr "numOfResults" xpickle+ ++-- | the job data, include the user-input and some additional control-data+data JobData = JobData {+ jd_JobId :: JobId+ , jd_State :: JobState+ , jd_OutputMap :: OutputMap+ , jd_Info :: JobInfo+ , jd_startTime :: UTCTime+ , jd_endTime :: UTCTime+ , jd_Result :: JobResultContainer+ }++instance Show JobData where+ show (JobData jid state om _ t1 t2 _) =+ "JobId:\t" ++ show jid ++ "\n" +++ "State:\t" ++ show state ++ "\n" +++ "Info:\tJobInfo\n" +++ "StartTime:\t" ++ show t1 ++ "\n" +++ "EndTime:\t" ++ show t2 ++ "\n" +++ "OutputMap:\n" ++ show om ++ "\n"+++data JobResultContainer = JobResultContainer (MVar JobResult)++instance Show JobResultContainer where+ show _ = "JobResultContainer"++-- | the result of the job, given by the master+data JobResult = JobResult {+ jr_Output :: [FunctionData]+ } deriving (Show)++instance Binary JobResult where+ put (JobResult o) = put o+ get+ = do+ o <- get+ return (JobResult o)++++-- ----------------------------------------------------------------------------+-- Reader and Writer+-- ----------------------------------------------------------------------------++-- | the ActionEnvironment contains all data that might be needed+-- during an action. So far, it only keeps the current task data and+-- a reference to the global filesystem and the options. +-- This is a good place to implement counters for the map-reduce-system +-- or other stuff...+data ActionEnvironment = ActionEnvironment {+ ae_TaskData :: TaskData+ , ae_FileSystem :: FS.FileSystem+ }++mkActionEnvironment :: TaskData -> FS.FileSystem -> ActionEnvironment+mkActionEnvironment td fs = ActionEnvironment td fs+++type InputReader k1 v1 = B.ByteString -> IO [(k1,v1)]++type OutputWriter k2 v2 = [(k2,v2)] -> IO B.ByteString++readConnector+ :: (Binary k1, Binary v1)+ => InputReader k1 v1+ -> ActionEnvironment+ -> [FunctionData] + -> IO [(k1,v1)]+readConnector ic ae ls+ = do+ debugM localLogger $ "readConnector: " ++ show ls+ os <- mapM (readInput (ae_FileSystem ae)) ls+ return $ concat $ catMaybes os+ where + -- readInput :: FS.FileSystem -> FunctionData -> IO (Maybe [(k1,v1)])+ readInput _ (TupleFunctionData t) = return $ Just $ [decode t] + readInput fs (FileFunctionData f)+ = do + debugM localLogger $ "loadInputList: getting content for: " ++ f+ mbc <- FS.getFileContent f fs+ debugM localLogger $ "loadInputList: content is: " ++ show mbc+ if isNothing mbc+ then do+ debugM localLogger $ "loadInputList: no content found"+ return Nothing+ else do+ d <- ic $ fromJust mbc+ return $ Just d+++writeConnector + :: (Binary k2, Binary v2)+ => OutputWriter k2 v2+ -> ActionEnvironment+ -> [(Int,[(k2,v2)])] + -> IO [(Int,[FunctionData])]+writeConnector oc ae ls+ = do+ debugM localLogger $ "writeConnector: "+ os <- mapM (writeOutput (ae_FileSystem ae) tot) ls+ return $ catMaybes os+ where + td = ae_TaskData ae+ tot = td_OutputType $ ae_TaskData ae+ -- writeOutput :: FS.FileSystem -> TaskOutputType -> (Int,[(k2,v2)]) -> IO (Maybe (Int,[FunctionData]))+ writeOutput _ TOTRawTuple (i,ts) + = return $ Just $ (i,bs)+ where+ bs = map (\t -> TupleFunctionData $ encode t) ts+ -- TODO exception werfen + writeOutput fs _ (i,ts)+ = do+ c <- oc ts+ FS.appendFile fn c fs+ return $ Just (i,[FileFunctionData fn])+ where+ fn = "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i+++defaultInputReader :: (Binary k1, Binary v1) => InputReader k1 v1+defaultInputReader b+ = return $ parseByteStringToList b+++defaultOutputWriter :: (Binary k2, Binary v2) => OutputWriter k2 v2+defaultOutputWriter ls+ = return $ B.concat $ map encode ls+++type ActionName = String++type ActionInfo = String++data ActionConfiguration a k1 v1 k2 v2 v3 v4+ = ActionConfiguration {+ ac_Name :: ActionName+ , ac_Info :: ActionInfo+ , ac_OptEncoder :: OptionsEncoder a+ , ac_OptDecoder :: OptionsDecoder a+ , ac_InputEncoder :: InputEncoder k1 v1+ , ac_OutputDecoder :: OutputDecoder k2 v4+ , ac_Split :: Maybe (SplitConfiguration a k1 v1)+ , ac_Map :: Maybe (MapConfiguration a k1 v1 k2 v2)+ , ac_Combine :: Maybe (ReduceConfiguration a k2 v2 v3)+ , ac_Reduce :: Maybe (ReduceConfiguration a k2 v3 v4)+ } +++data SplitConfiguration a k1 v1+ = SplitConfiguration {+ sc_Function :: SplitFunction a k1 v1+ , sc_Reader :: InputReader k1 v1+ , sc_Writer :: OutputWriter k1 v1+ }+++data MapConfiguration a k1 v1 k2 v2+ = MapConfiguration {+ mc_Function :: MapFunction a k1 v1 k2 v2+ , mc_Partition :: MapPartition a k2 v2+ , mc_Reader :: InputReader k1 v1+ , mc_Writer :: OutputWriter k2 v2+ }+++data ReduceConfiguration a k2 v2 v3+ = ReduceConfiguration {+ rc_Merge :: ReduceMerge a k2 v2+ , rc_Function :: ReduceFunction a k2 v2 v3+ , rc_Partition :: ReducePartition a k2 v3+ , rc_Reader :: InputReader k2 v2+ , rc_Writer :: OutputWriter k2 v3+ }+++defaultActionConfiguration+ :: (Binary a, Binary k1, Binary v1, Binary k2, Binary v4)+ => ActionName+ -> ActionConfiguration a k1 v1 k2 v2 v3 v4+defaultActionConfiguration name+ = ActionConfiguration+ name+ ""+ defaultOptionsEncoder+ defaultOptionsDecoder+ defaultInputEncoder+ defaultOutputDecoder+ (Just defaultSplitConfiguration)+ Nothing+ Nothing+ Nothing+++defaultSplitConfiguration+ :: (Binary a, Binary k1, Binary v1)+ => SplitConfiguration a k1 v1+defaultSplitConfiguration + = SplitConfiguration+ defaultSplit+ defaultInputReader+ defaultOutputWriter +++defaultMapConfiguration+ :: (Ord k2, Binary a, Binary k1, Binary v1, Binary k2, Binary v2)+ => MapFunction a k1 v1 k2 v2+ -> MapConfiguration a k1 v1 k2 v2+defaultMapConfiguration fct+ = MapConfiguration+ fct+ defaultPartition+ defaultInputReader+ defaultOutputWriter+++defaultReduceConfiguration+ :: (Ord k2, Binary a, Binary k2, Binary v2, Binary v3)+ => ReduceFunction a k2 v2 v3+ -> ReduceConfiguration a k2 v2 v3+defaultReduceConfiguration fct+ = ReduceConfiguration+ defaultMerge+ fct+ defaultPartition+ defaultInputReader+ defaultOutputWriter+++defaultSplit+ :: SplitFunction a k1 v1+defaultSplit _ _ n ls + = return partedList+ where+ partedList = AMap.toList $ AMap.fromList ps+ ns = [(x `mod` n) + 1 | x <- [0..]]+ is = map (\a -> [a]) ls + ps = zip ns is+++defaultPartition+ :: (Binary k2, Binary v2)+ => MapPartition a k2 v2+defaultPartition _ _ 1 ls+ -- To make it faster, wenn no partition is used+ = return [(1,ls)]+defaultPartition _ _ n ls+ = do+ -- calculate partition-Values+ let markedList = map (\t@(k,_) -> (hash k,t)) ls+ -- merge them+ + -- TODO this might change (revert) the order of the Elements...+ let resultList = AMap.toList $ AMap.fromTupleList markedList+ return resultList+ where+ -- calculate a hash-value, because we only have the Binary-Instance, we+ -- can only use the Bytestring of the Value+ hash k = ((fromIntegral $ Hash.hashString (show $ encode k)) `mod` n) + 1+++defaultMerge+ :: (Ord k2, Binary k2, Binary v2)+ => ReduceMerge a k2 v2+defaultMerge _ _ ls+ = return $ AMap.toList $ AMap.fromTupleList ls++++data ActionData+ = ActionData {+ ad_Name :: ActionName+ , ad_Info :: ActionInfo+ , ad_Split :: Maybe BinarySplitAction+ , ad_Map :: Maybe BinaryMapAction+ , ad_Combine :: Maybe BinaryReduceAction+ , ad_Reduce :: Maybe BinaryReduceAction+ }++instance KMap.Key ActionData where+ getKey = ad_Name++instance Show ActionData where+ show (ActionData n i _ _ _ _) = "{ActionData name:\"" ++ n ++ "\" info:\"" ++ i ++ "\"}"+++type ActionMap = KMap.KeyMap ActionData++type OptionsEncoder a = a -> B.ByteString+type OptionsDecoder a = B.ByteString -> a++defaultOptionsEncoder :: (Binary a) => OptionsEncoder a+defaultOptionsEncoder = encode++defaultOptionsDecoder :: (Binary a) => OptionsDecoder a+defaultOptionsDecoder = decode+++type InputEncoder k1 v1 = [(k1,v1)] -> [FS.FileId] -> [FunctionData]+type OutputDecoder k2 v4 = [FunctionData] -> ([(k2,v4)],[FS.FileId])++defaultInputEncoder :: (Binary k1, Binary v1) => InputEncoder k1 v1+defaultInputEncoder ls1 ls2 = ls1' ++ ls2'+ where+ ls1' = map (\t -> TupleFunctionData (encode t)) ls1+ ls2' = map (\f -> FileFunctionData f) ls2++defaultOutputDecoder :: (Binary k2, Binary v4) => OutputDecoder k2 v4+defaultOutputDecoder ls = (ls1, ls2)+ where+ ls1 = map (\t -> decode t) $ catMaybes ls1'+ ls2 = catMaybes ls2'+ (ls1',ls2') = unzip $ map (splitter) ls+ splitter (TupleFunctionData t) = (Just t, Nothing)+ splitter (FileFunctionData f) = (Nothing, Just f)+ ++getActionForTaskType :: TaskType -> ActionData -> Maybe BinaryReduceAction+getActionForTaskType TTSplit ad = ad_Split ad+getActionForTaskType TTMap ad = ad_Map ad+getActionForTaskType TTCombine ad = ad_Combine ad+getActionForTaskType TTReduce ad = ad_Reduce ad+getActionForTaskType _ _ = Nothing++ ++readActionConfiguration+ :: ( Ord k2, Binary a+ , Binary k1, Binary v1+ , Binary k2, Binary v2+ , Binary v3, Binary v4)+ => ActionConfiguration a k1 v1 k2 v2 v3 v4 + -> ActionData+readActionConfiguration (ActionConfiguration n i _ optDec _ _ sc mc cc rc)+ = ActionData n i sf mf cf rf+ where+ sf = maybe Nothing (\(SplitConfiguration f ir ow) -> Just $ performSplitAction optDec f ir ow) sc + mf = maybe Nothing (\(MapConfiguration f p ir ow) -> Just $ performMapAction optDec f p ir ow) mc+ cf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) cc+ rf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) rc++++createJobInfoFromConfiguration+ :: ActionConfiguration a k1 v1 k2 v2 v3 v4+ -> a -- ^ options+ -> [(k1,v1)] -- ^ input (Tuples)+ -> [FS.FileId] -- ^ input (Files)+ -> Int -- ^ number of splitters+ -> Int -- ^ number of mappers+ -> Int -- ^ number of reducers+ -> Int -- ^ number of results+ -> TaskOutputType -- ^ type of the result (file of raw)+ -> JobInfo+createJobInfoFromConfiguration (ActionConfiguration n _ optEnc _ inEnc _ sConf mConf cConf rConf) opts' ls1 ls2 sCnt mCnt rCnt nor' rt+ = JobInfo n opts sa ma ca ra nor i+ where+ opts = optEnc opts'+ sa = maybe Nothing (\_ -> Just $ JobAction n TOTFile sCnt) sConf+ ma = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) mConf+ ca = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) cConf+ ra = maybe Nothing (\_ -> Just $ JobAction n rt rCnt) rConf+ nor = Just nor'+ i = inEnc ls1 ls2+ + +createListsFromJobResult+ :: ActionConfiguration a k1 v1 k2 v2 v3 v4+ -> JobResult+ -> ([(k2,v4)],[FS.FileId])+createListsFromJobResult ac jr = (ac_OutputDecoder ac) (jr_Output jr)+ +-- ----------------------------------------------------------------------------+-- SplitAction+-- ----------------------------------------------------------------------------++-- | general SplitAction+type SplitAction a k1 v1 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k1,v1)])]++-- | SplitAction on ByteStrings+type BinarySplitAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] ++type SplitFunction a k1 v1 = SplitAction a k1 v1++performSplitAction+ :: (Binary a, Binary k1, Binary v1)+ => OptionsDecoder a+ -> SplitFunction a k1 v1+ -> InputReader k1 v1+ -> OutputWriter k1 v1+ -> ActionEnvironment+ -> B.ByteString+ -> Maybe Int+ -> (Int,[FunctionData])+ -> IO [(Int, [FunctionData])]+performSplitAction optDec fct reader writer env opts n (i,ls)+ = do+ let a = optDec opts+ + infoM localLogger "performSplitAction"+ + infoM localLogger "reading inputList"+ inputList <- readConnector reader env ls+ + infoM localLogger "doing split"+ partedList <- case n of+ (Just n') -> fct env a n' inputList+ (Nothing) -> return [(i,inputList)]+ + infoM localLogger "writing outputlist"+ outputList <- writeConnector writer env partedList+ return outputList+++-- ----------------------------------------------------------------------------+-- MapAction+-- ----------------------------------------------------------------------------+++-- | general MapAction+type MapAction a k1 v1 k2 v2 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k2,v2)])]++-- | MapAction on ByteStrings+type BinaryMapAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] ++type MapFunction a k1 v1 k2 v2 = ActionEnvironment -> a -> k1 -> v1 -> IO [(k2, v2)]++type MapPartition a k2 v2 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v2)])]+ ++performMapAction+ :: (Ord k2,+ Binary a, Binary k1, Binary v1, Binary k2, Binary v2)+ => OptionsDecoder a+ -> MapFunction a k1 v1 k2 v2+ -> MapPartition a k2 v2+ -> InputReader k1 v1+ -> OutputWriter k2 v2+ -> ActionEnvironment+ -> B.ByteString+ -> Maybe Int+ -> (Int,[FunctionData])+ -> IO [(Int, [FunctionData])]+performMapAction optDec fct part reader writer env opts n (i,ls)+ = do+ -- decode the options+ let a = optDec opts+ + infoM localLogger "performMapAction"+ + infoM localLogger "reading inputList"+ inputList <- readConnector reader env ls+ + infoM localLogger "doing map"+ mappedList <- mapM (\(k1, v1) -> fct env a k1 v1) inputList+ let tupleList = concat mappedList+ + infoM localLogger "doing partition"+ partedList <- case n of+ (Just n') -> part env a n' tupleList+ (Nothing) -> return [(i,tupleList)]+ + infoM localLogger "writing outputlist"+ outputList <- writeConnector writer env partedList+ return outputList++++++-- ----------------------------------------------------------------------------+-- Combine- / ReduceTask+-- ----------------------------------------------------------------------------++-- | general MapAction+type ReduceAction a k2 v2 v3 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v3)])]++-- | MapAction on ByteStrings+type BinaryReduceAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] ++type ReduceMerge a k2 v2 = ActionEnvironment -> a -> [(k2,v2)] -> IO [(k2,[v2])]++type ReduceFunction a k2 v2 v3 = ActionEnvironment -> a -> k2 -> [v2] -> IO (Maybe v3)++type ReducePartition a k2 v3 = ActionEnvironment -> a -> Int -> [(k2,v3)] -> IO [(Int, [(k2,v3)])]+ ++performReduceAction+ :: (Ord k2,+ Binary a, Binary k2, Binary v2, Binary v3)+ => OptionsDecoder a+ -> ReduceMerge a k2 v2+ -> ReduceFunction a k2 v2 v3+ -> ReducePartition a k2 v3+ -> InputReader k2 v2+ -> OutputWriter k2 v3+ -> ActionEnvironment+ -> B.ByteString+ -> Maybe Int+ -> (Int,[FunctionData])+ -> IO [(Int, [FunctionData])]+performReduceAction optDec merge fct part reader writer env opts n (i,ls)+ = do+ -- decode the options+ let a = optDec opts+ + infoM localLogger "performReduceAction"+ + infoM localLogger "reading inputList"+ inputList <- readConnector reader env ls+ + infoM localLogger "doing merge"+ mergedList <- merge env a inputList+ + infoM localLogger "doing reduce"+ maybesList <- mapM (\(k2,v2s) -> performReduceFunction a k2 v2s) mergedList+ let tupleList = catMaybes maybesList+ + infoM localLogger "doing partition" + partedList <- case n of+ (Just n') -> part env a n' tupleList+ (Nothing) -> return [(i,tupleList)] + + infoM localLogger "writing outputlist"+ outputList <- writeConnector writer env partedList+ + return outputList+ where+ performReduceFunction a k2 v2s+ = do+ mbV3 <- fct env a k2 v2s+ case mbV3 of+ (Nothing) -> return Nothing+ (Just v3) -> return $ Just (k2,v3)
+ source/Holumbus/MapReduce/UserInterface.hs view
@@ -0,0 +1,151 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.MapReduce.UserInterface+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.MapReduce.UserInterface+ (+ -- * operations+ runUI+ )+where++import Data.Maybe++import Holumbus.Common.Utils ( handleAll )+import Holumbus.Common.FileHandling+import qualified Holumbus.Common.Debug as DEBUG+import qualified Holumbus.Console.Console as Console++import qualified Holumbus.MapReduce.MapReduce as MR+import qualified Holumbus.MapReduce.Types as T++++-- ----------------------------------------------------------------------------+-- Operations+-- ----------------------------------------------------------------------------++-- | runs the user interface... just add an fileSystem an a fancy version-number+runUI :: (MR.MapReduce mr) => mr -> String -> IO ()+runUI mr version+ = do+ -- starts the console with the specified commands+ Console.handleUserInput (createConsole version) mr++++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------+++createConsole :: (MR.MapReduce mr) => String -> Console.ConsoleData mr+createConsole version =+ Console.addConsoleCommand "site" getMySiteId "" $+ Console.addConsoleCommand "mrtype" getMapReduceType "" $+ Console.addConsoleCommand "startC" startControlling "" $+ Console.addConsoleCommand "stopC" stopControlling "" $+ Console.addConsoleCommand "isC" isControlling "" $+ Console.addConsoleCommand "step" doSingleStep "" $+ Console.addConsoleCommand "mrJob" doMapReduceJob "" $+ Console.addConsoleCommand "parse" parseJob "" $+ Console.addConsoleCommand "debug" printDebug "" $ + Console.addConsoleCommand "version" (printVersion version) "prints the version" $ + Console.initializeConsole++ +getMySiteId :: (MR.MapReduce mr) => mr -> [String] -> IO ()+getMySiteId mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ s <- MR.getMySiteId mr+ putStrLn $ show s+++getMapReduceType :: (MR.MapReduce mr) => mr -> [String] -> IO ()+getMapReduceType mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ t <- MR.getMapReduceType mr+ putStrLn $ show t++ +startControlling :: (MR.MapReduce mr) => mr -> [String] -> IO ()+startControlling mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ MR.startControlling mr+++stopControlling :: (MR.MapReduce mr) => mr -> [String] -> IO ()+stopControlling mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ MR.stopControlling mr+++isControlling :: (MR.MapReduce mr) => mr -> [String] -> IO ()+isControlling mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ b <- MR.isControlling mr+ putStrLn $ show b+++ +doSingleStep :: (MR.MapReduce mr) => mr -> [String] -> IO ()+doSingleStep mr _+ = handleAll (\e -> putStrLn $ show e) $+ do+ MR.doSingleStep mr+++ +doMapReduceJob :: (MR.MapReduce mr) => mr -> [String] -> IO ()+doMapReduceJob mr opts+ = do+ handleAll (\e -> putStrLn $ show e) $+ do+ (mbName,_) <- Console.nextOption opts+ jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+ r <- MR.doMapReduceJob jobInfo mr+ putStrLn "RESULT:" + putStrLn $ show r+ return ()+++parseJob :: (MR.MapReduce mr) => mr -> [String] -> IO ()+parseJob _ opts+ = do+ handleAll (\e -> putStrLn $ show e) $+ do+ (mbName,_) <- Console.nextOption opts+ jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+ putStrLn $ show jobInfo+ return ()+++printDebug :: (MR.MapReduce mr) => mr -> [String] -> IO ()+printDebug mr _+ = do+ handleAll (\e -> putStrLn $ show e) $+ do+ DEBUG.printDebug mr++ +printVersion :: (MR.MapReduce mr) => String -> mr -> [String] -> IO ()+printVersion version _ _+ = do+ putStrLn version+
+ source/Holumbus/Standalone/SMapReduce.hs view
@@ -0,0 +1,243 @@+-- ----------------------------------------------------------------------------+{- |+ Module : Holumbus.Standalone.SMapReduce+ Copyright : Copyright (C) 2008 Stefan Schmidt+ License : MIT++ Maintainer : Stefan Schmidt (stefanschmidt@web.de)+ Stability : experimental+ Portability: portable+ Version : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Standalone.SMapReduce+(+-- * Datatypes+ SMapReduce+, MapReduce(..)++-- * Configurations+, SMRConf+, defaultStandaloneConfig+ +-- * Creation and Initialisation+, newSMapReduce+)+where++import Control.Concurrent+import System.Log.Logger++import Holumbus.Common.Debug+import qualified Holumbus.FileSystem.FileSystem as FS+import Holumbus.MapReduce.Types+import Holumbus.MapReduce.JobController+import Holumbus.MapReduce.TaskProcessor+import Holumbus.MapReduce.MapReduce+import Holumbus.Network.Site+++localLogger :: String+localLogger = "Holumbus.Standalone.SMapReduce"++-- ----------------------------------------------------------------------------+-- Datatypes+-- ----------------------------------------------------------------------------+++data SMapReduceData = SMapReduceData {+ sad_JobController :: JobController+ , sad_TaskProcessor :: TaskProcessor+ }+++data SMapReduce = SMapReduce (MVar SMapReduceData)++instance Show SMapReduce where+ show _ = "SMapReduce"++++-- ---------------------------------------------------------------------------+-- Configurations+-- ---------------------------------------------------------------------------+++data SMRConf = SMRConf {+ stc_StartControlling :: Bool+}++defaultStandaloneConfig :: SMRConf+defaultStandaloneConfig = SMRConf True++++-- ----------------------------------------------------------------------------+-- "glue" between JobController and TaskProcessor+-- ----------------------------------------------------------------------------+++sendStartTask :: TaskProcessor -> TaskData -> IO (TaskSendResult)+sendStartTask tp td+ = do+ debugM localLogger "sendStartTask"+ debugM localLogger "TaskData:"+ debugM localLogger $ show td+ startTask td tp+ return TSRSend+++sendTaskCompleted :: JobController -> TaskData -> IO Bool+sendTaskCompleted jc td+ = do+ debugM localLogger "Task completed"+ debugM localLogger "TaskData:"+ debugM localLogger $ show td+ setTaskCompleted jc td+ return True+++sendTaskError :: JobController -> TaskData -> IO Bool+sendTaskError jc td+ = do+ debugM localLogger "Task error"+ debugM localLogger "TaskData:"+ debugM localLogger $ show td+ setTaskError jc td+ return True+++++-- ----------------------------------------------------------------------------+-- Creation and Initialisation+-- ----------------------------------------------------------------------------+++newSMapReduce + :: FS.FileSystem -> ActionMap -> SMRConf+ -> IO SMapReduce+newSMapReduce fs am conf+ = do+ let start = stc_StartControlling conf+ -- get a new JobController an TaskProcessor+ jc <- newJobController+ tp <- newTaskProcessor+ + -- configure the JobController+ setTaskSendHook (sendStartTask tp) jc+ + -- configure the TaskProcessor+ setFileSystemToTaskProcessor fs tp+ setActionMap am tp+ setTaskCompletedHook (sendTaskCompleted jc) tp+ setTaskErrorHook (sendTaskError jc) tp+ + if (start)+ then do startJobController jc+ else do return ()+ + startTaskProcessor tp+ + let sad = SMapReduceData jc tp+ sa <- (newMVar sad)+ return (SMapReduce sa) ++ +{- +printJobResult :: MVar JobResult -> IO ()+printJobResult mVarRes+ = do+ forkIO $+ withMVar mVarRes $ + \(JobResult outs) -> + do+ putStrLn "RESULT:" + putStrLn $ show (decodeResult outs)+ return ()+ where+ decodeResult :: [FunctionData] -> [(String, Integer)]+ decodeResult ls = map decodeResult' ls+ where+ decodeResult' (FileFunctionData f) = (f, -1)+ decodeResult' (TupleFunctionData b) = decodeTuple b+-}++++-- ----------------------------------------------------------------------------+-- Typeclass instanciation+-- ----------------------------------------------------------------------------++instance Debug SMapReduce where++ printDebug (SMapReduce sa)+ = do+ withMVar sa $+ \sad ->+ do+ putStrLn "--------------------------------------------------------" + putStrLn "Job-Controller"+ showJC <- printJobController (sad_JobController sad)+ putStrLn showJC+ putStrLn "--------------------------------------------------------"+ putStrLn "Task-Processor"+ showTP <- printTaskProcessor (sad_TaskProcessor sad)+ putStrLn $showTP+ putStrLn "--------------------------------------------------------"+ putStrLn "Actions"+ actions <- getActions (sad_TaskProcessor sad)+ putStrLn $ show $ actions+++++instance MapReduce SMapReduce where+++ closeMapReduce (SMapReduce sa)+ = modifyMVar sa $+ \sad ->+ do+ closeTaskProcessor (sad_TaskProcessor sad)+ closeJobController (sad_JobController sad)+ return (sad, ())+++ getMySiteId _+ = getSiteId++ + getMapReduceType _+ = return MRTStandalone++ + startControlling (SMapReduce sa)+ = withMVar sa $+ \sad -> do startJobController (sad_JobController sad)+ + + + stopControlling (SMapReduce sa)+ = withMVar sa $+ \sad -> do stopJobController (sad_JobController sad)+ ++ + isControlling (SMapReduce sa)+ = withMVar sa $+ \sad -> do isJobControllerRunning (sad_JobController sad)+ ++ + doSingleStep (SMapReduce sa)+ = withMVar sa $+ \sad -> do singleStepJobControlling (sad_JobController sad)++ + doMapReduceJob ji (SMapReduce sa)+ = withMVar sa $+ \sad -> do performJob ji (sad_JobController sad)