From 3eed70c3d2dfec84f739fcfaf591513743b2e2de Mon Sep 17 00:00:00 2001
From: Joey Hess <joey@kitenet.net>
Date: Wed, 22 May 2013 00:27:12 -0400
Subject: avoid building up a lot of threads all waiting for their chance to
 run a push

Only 2 threads are needed, one running, and one waiting to push something
new. Any more is redundant and wasteful.
---
 Assistant/NetMessager.hs | 73 ++++++++++++++++++++++++++++++++++++------------
 1 file changed, 55 insertions(+), 18 deletions(-)

(limited to 'Assistant/NetMessager.hs')

diff --git a/Assistant/NetMessager.hs b/Assistant/NetMessager.hs
index 7ee4d464b..fe96df56f 100644
--- a/Assistant/NetMessager.hs
+++ b/Assistant/NetMessager.hs
@@ -70,35 +70,72 @@ checkImportantNetMessages (storedclient, sentclient) = go <<~ netMessager
 		return (fromMaybe S.empty stored, fromMaybe S.empty sent)
 
 {- Runs an action that runs either the send or receive side of a push.
+ - Only one such action per side can run at a time. Other pushes, for
+ - the same, or other clients, need to wait their turn.
+ -
+ - Only one push is allowed to wait per client.
+ - There is no point in building up more.
  -
  - While the push is running, netMessagesPush will get messages put into it
  - relating to this push, while any messages relating to other pushes
  - on the same side go to netMessagesDeferred. Once the push finishes,
  - those deferred messages will be fed to handledeferred for processing.
- -
- - If this is called when a push of the same side is running, it will
- - block until that push completes, and then run.
  -}
-runPush :: PushSide -> ClientID -> Assistant a -> Assistant a
+runPush :: PushSide -> ClientID -> Assistant Bool -> Assistant Bool
 runPush side clientid a = do
+	debugmsg "preparing to run"
 	nm <- getAssistant netMessager
-	let v = getSide side $ netMessagerPushRunning nm
-	debugmsg <- asIO1 $ \s -> netMessagerDebug clientid [s, show side]
 	let setup = do
-		debugmsg "preparing to run"
-		atomically $ ifM (isNothing <$> tryReadTMVar v)
-			( putTMVar v clientid
+		(canrun, release) <- atomically $ checkcanrun nm
+		if canrun
+			then atomically $ waittorun nm release
+			else return (False, noop)
+	let cleanup (_, release) = atomically release
+	go <- asIO1 $ \(run, _) ->
+		if run
+			then do
+				debugmsg "started running"
+				r <- a
+				debugmsg "finished running"
+				{- Empty the inbox, because stuff may have
+				 - been left in it if the push failed. -}
+				emptyInbox clientid side
+				return r
+			else do
+				debugmsg "skipping running"
+				return False
+	r <- liftIO $ E.bracket setup cleanup go
+	return r
+  where
+	debugmsg s = netMessagerDebug clientid [s, show side]
+	-- check that this is one of the two threads allowed
+	-- to run at the same time, pushing to the same client
+	-- on the same side
+  	checkcanrun nm = do
+		let v = getSide side $ netMessagerPushThreadCount nm
+		m <- readTVar v
+		case M.lookup clientid m of
+			Just count
+				| count > 2 -> return (False, noop)
+			_ -> do
+				writeTVar v $
+					M.insertWith' (+) clientid 1 m
+				let release = modifyTVar' v $
+					M.insertWith' (-) clientid 1
+				return (True, release)
+	-- block until this is the only thread performing
+	-- a push on this side, to any client
+	waittorun nm release = do
+		let v = getSide side $ netMessagerPushRunning nm
+		ifM (isNothing <$> tryReadTMVar v)
+			( do
+				putTMVar v clientid
+				let release' = do
+					void $ takeTMVar v
+					release
+				return (True, release')
 			, retry
 			)
-		debugmsg "started running"
-	let cleanup = do
-		debugmsg "finished running"
-		atomically $ takeTMVar v
-	r <- E.bracket_ setup cleanup <~> a
-	{- Empty the inbox, because stuff may have been left in it
-	 - if the push failed. -}
-	emptyInbox clientid side
-	return r
 
 {- Stores messages for a push into the appropriate inbox.
  -
-- 
cgit v1.2.3