de117ea · distributed-process

git.iko.soy/distributed-process

Make work-stealing master run in constant space

commitde117ea40f8db3857261875ed8fcec6b368a76b6 authorEdsko de Vries <edsko@edsko.net> dateMon 1 Oct 2012, 14:18 (14 years ago) parent6f40a8f

2 files changed+14−6

compare with master

unifiedsplit

distributed-process-demos/src/WorkStealing/WorkStealing.hs+11−4 distributed-process/src/Control/Distributed/Process/Node.hs+3−2

distributed-process-demos/src/WorkStealing/WorkStealing.hs+11−4

@@ -22,6 +22,16 @@ slave (master, workQueue) = do

2222

2323remotable ['slave]

2424

25+-- | Wait for n integers and sum them all up

26+sumIntegers :: Int -> Process Integer

27+sumIntegers = go 0

28+ where

29+ go :: Integer -> Int -> Process Integer

30+ go !acc 0 = return acc

31+ go !acc n = do

32+ m <- expect

33+ go (acc + m) (n - 1)

34+

2535master :: Integer -> [NodeId] -> Process Integer

2636master n slaves = do

2737 us <- getSelfPid

@@ -41,7 +51,4 @@ master n slaves = do

4151 forM_ slaves $ \nid -> spawn nid ($(mkClosure 'slave) (us, workQueue))

4252

4353 -- Wait for the result

44− partials <- replicateM (fromIntegral n) (expect :: Process Integer)

45−

46− -- And return the sum

47− return (sum partials)

54+ sumIntegers (fromIntegral n)

distributed-process/src/Control/Distributed/Process/Node.hs+3−2

@@ -324,10 +324,11 @@ handleIncomingMessages node = go initConnectionState

324324 case st ^. incomingAt cid of

325325 Just (_, ToProc weakQueue) -> do

326326 mQueue <- deRefWeak weakQueue

327− forM_ mQueue $ \queue ->

327+ forM_ mQueue $ \queue -> do

328328 -- TODO: if we find that the queue is Nothing, should we remove

329329 -- it from the NC state? (and same for channels, below)

330− enqueue queue $ payloadToMessage payload

330+ let msg = payloadToMessage payload

331+ enqueue queue msg

331332 go st

332333 Just (_, ToChan (TypedChannel chan')) -> do

333334 mChan <- deRefWeak chan'