de117ea · distributed-process

git.iko.soy/distributed-process

on master help sign in

Make work-stealing master run in constant space

de117eaEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

23remotable ['slave]

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+

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

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

26master n slaves = do

36master n slaves = do

27 us <- getSelfPid

37 us <- getSelfPid

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

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

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

43 -- Wait for the result

53 -- Wait for the result

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

54+ sumIntegers (fromIntegral n)

45−

46− -- And return the sum

47− return (sum partials)

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

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

324 case st ^. incomingAt cid of

325 Just (_, ToProc weakQueue) -> do

326 mQueue <- deRefWeak weakQueue

327− forM_ mQueue $ \queue ->

327+ forM_ mQueue $ \queue -> do

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

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

330− enqueue queue $ payloadToMessage payload

330+ let msg = payloadToMessage payload

331+ enqueue queue msg

331 go st

332 go st

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

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

333 mChan <- deRefWeak chan'

334 mChan <- deRefWeak chan'