finally merge the assistant into master

Progress bars still need to be done, otherwise it's fully working.
Although much work remains to hit all the use cases.

1 files changed, 133 insertions, 34 deletions

diff --git a/Assistant.hs b/Assistant.hs
index e924d9477..413e5e90e 100644
--- a/Assistant.hs
+++ b/Assistant.hs
@@ -10,7 +10,7 @@
  - 	The initial thread run, double forks to background, starts other
  - 	threads, and then stops, waiting for them to terminate,
  - 	or for a ctrl-c.
- - Thread 2: watcher
+ - Thread 2: Watcher
  - 	Notices new files, and calls handlers for events, queuing changes.
  - Thread 3: inotify internal
  - 	Used by haskell inotify library to ensure inotify event buffer is
@@ -19,67 +19,166 @@
  - 	Scans the tree and registers inotify watches for each directory.
  -	A MVar lock is used to prevent other inotify handlers from running
  -	until this is complete.
- - Thread 5: committer
+ - Thread 5: Committer
  - 	Waits for changes to occur, and runs the git queue to update its
- - 	index, then commits.
- - Thread 6: status logger
+ - 	index, then commits. Also queues Transfer events to send added
+ - 	files to other remotes.
+ - Thread 6: Pusher
+ - 	Waits for commits to be made, and pushes updated branches to remotes,
+ - 	in parallel. (Forks a process for each git push.)
+ - Thread 7: PushRetryer
+ - 	Runs every 30 minutes when there are failed pushes, and retries
+ - 	them.
+ - Thread 8: Merger
+ - 	Waits for pushes to be received from remotes, and merges the
+ - 	updated branches into the current branch.
+ - 	(This uses inotify on .git/refs/heads, so there are additional
+ - 	inotify threads associated with it, too.)
+ - Thread 9: TransferWatcher
+ - 	Watches for transfer information files being created and removed,
+ - 	and maintains the DaemonStatus currentTransfers map.
+ - 	(This uses inotify on .git/annex/transfer/, so there are
+ - 	additional inotify threads associated with it, too.)
+ - Thread 10: Transferrer
+ - 	Waits for Transfers to be queued and does them.
+ - Thread 11: StatusLogger
  - 	Wakes up periodically and records the daemon's status to disk.
- - Thread 7: sanity checker
+ - Thread 12: SanityChecker
  - 	Wakes up periodically (rarely) and does sanity checks.
+ - Thread 13: MountWatcher
+ - 	Either uses dbus to watch for drive mount events, or, when
+ - 	there's no dbus, polls to find newly mounted filesystems.
+ - 	Once a filesystem that contains a remote is mounted, updates
+ - 	state about that remote, pulls from it, and queues a push to it,
+ - 	as well as an update, and queues it onto the
+ - 	ConnectedRemoteChan
+ - Thread 13: NetWatcher
+ - 	Deals with network connection interruptions, which would cause
+ - 	transfers to fail, and can be recovered from by waiting for a
+ - 	network connection, and syncing with all network remotes.
+ - 	Uses dbus to watch for network connections, or when dbus
+ - 	cannot be used, assumes there's been one every 30 minutes.
+ - Thread 15: TransferScanner
+ - 	Does potentially expensive checks to find data that needs to be
+ - 	transferred from or to remotes, and queues Transfers.
+ - 	Uses the ScanRemotes map.
+ - Thread 16: WebApp
+ - 	Spawns more threads as necessary to handle clients.
+ - 	Displays the DaemonStatus.
  -
  - ThreadState: (MVar)
  - 	The Annex state is stored here, which allows resuscitating the
- - 	Annex monad in IO actions run by the inotify and committer
+ - 	Annex monad in IO actions run by the watcher and committer
  - 	threads. Thus, a single state is shared amoung the threads, and
  - 	only one at a time can access it.
- - DaemonStatusHandle: (MVar)
- - 	The daemon's current status. This MVar should only be manipulated
- - 	from inside the Annex monad, which ensures it's accessed only
- - 	after the ThreadState MVar.
+ - DaemonStatusHandle: (STM TMVar)
+ - 	The daemon's current status.
  - ChangeChan: (STM TChan)
  - 	Changes are indicated by writing to this channel. The committer
  - 	reads from it.
+ - CommitChan: (STM TChan)
+ - 	Commits are indicated by writing to this channel. The pusher reads
+ - 	from it.
+ - FailedPushMap (STM TMVar)
+ - 	Failed pushes are indicated by writing to this TMVar. The push
+ - 	retrier blocks until they're available.
+ - TransferQueue (STM TChan)
+ - 	Transfers to make are indicated by writing to this channel.
+ - TransferSlots (QSemN)
+ - 	Count of the number of currently available transfer slots.
+ - 	Updated by the transfer watcher, this allows other threads
+ - 	to block until a slot is available.
+ - 	This MVar should only be manipulated from inside the Annex monad,
+ - 	which ensures it's accessed only after the ThreadState MVar.
+ - ScanRemotes (STM TMVar)
+ - 	Remotes that have been disconnected, and should be scanned
+ - 	are indicated by writing to this TMVar.
  -}
 
+{-# LANGUAGE CPP #-}
+
 module Assistant where
 
-import Common.Annex
+import Assistant.Common
 import Assistant.ThreadedMonad
 import Assistant.DaemonStatus
 import Assistant.Changes
-import Assistant.Watcher
-import Assistant.Committer
-import Assistant.SanityChecker
+import Assistant.Commits
+import Assistant.Pushes
+import Assistant.ScanRemotes
+import Assistant.TransferQueue
+import Assistant.TransferSlots
+import Assistant.Threads.Watcher
+import Assistant.Threads.Committer
+import Assistant.Threads.Pusher
+import Assistant.Threads.Merger
+import Assistant.Threads.TransferWatcher
+import Assistant.Threads.Transferrer
+import Assistant.Threads.SanityChecker
+import Assistant.Threads.MountWatcher
+import Assistant.Threads.NetWatcher
+import Assistant.Threads.TransferScanner
+#ifdef WITH_WEBAPP
+import Assistant.Threads.WebApp
+#else
+#warning Building without the webapp. You probably need to install Yesod..
+#endif
 import qualified Utility.Daemon
 import Utility.LogFile
+import Utility.ThreadScheduler
 
 import Control.Concurrent
 
-startDaemon :: Bool -> Annex ()
-startDaemon foreground
+stopDaemon :: Annex ()
+stopDaemon = liftIO . Utility.Daemon.stopDaemon =<< fromRepo gitAnnexPidFile
+
+startDaemon :: Bool -> Bool -> Maybe (Url -> FilePath -> IO ()) -> Annex ()
+startDaemon assistant foreground webappwaiter
 	| foreground = do
-		showStart "watch" "."
+		showStart (if assistant then "assistant" else "watch") "."
+		liftIO . Utility.Daemon.lockPidFile =<< fromRepo gitAnnexPidFile
 		go id
 	| otherwise = do
 		logfd <- liftIO . openLog =<< fromRepo gitAnnexLogFile
 		pidfile <- fromRepo gitAnnexPidFile
 		go $ Utility.Daemon.daemonize logfd (Just pidfile) False
 	where
-		go a = withThreadState $ \st -> do
-			checkCanWatch
-			dstatus <- startDaemonStatus
-			liftIO $ a $ do
-				changechan <- newChangeChan
-				-- The commit thread is started early,
-				-- so that the user can immediately
-				-- begin adding files and having them
-				-- committed, even while the startup scan
-				-- is taking place.
-				_ <- forkIO $ commitThread st changechan
-				_ <- forkIO $ daemonStatusThread st dstatus
-				_ <- forkIO $ sanityCheckerThread st dstatus changechan
-				-- Does not return.
-				watchThread st dstatus changechan
+		go d = startAssistant assistant d webappwaiter
 
-stopDaemon :: Annex ()
-stopDaemon = liftIO . Utility.Daemon.stopDaemon =<< fromRepo gitAnnexPidFile
+startAssistant :: Bool -> (IO () -> IO ()) -> Maybe (Url -> FilePath -> IO ()) -> Annex ()
+startAssistant assistant daemonize webappwaiter = do
+	withThreadState $ \st -> do
+		checkCanWatch
+		dstatus <- startDaemonStatus
+		liftIO $ daemonize $ run dstatus st
+	where
+		run dstatus st = do
+			changechan <- newChangeChan
+			commitchan <- newCommitChan
+			pushmap <- newFailedPushMap
+			transferqueue <- newTransferQueue
+			transferslots <- newTransferSlots
+			scanremotes <- newScanRemoteMap
+			mapM_ startthread
+				[ watch $ commitThread st changechan commitchan transferqueue dstatus
+#ifdef WITH_WEBAPP
+				, assist $ webAppThread (Just st) dstatus scanremotes transferqueue transferslots Nothing webappwaiter
+#endif
+				, assist $ pushThread st dstatus commitchan pushmap
+				, assist $ pushRetryThread st dstatus pushmap
+				, assist $ mergeThread st
+				, assist $ transferWatcherThread st dstatus
+				, assist $ transfererThread st dstatus transferqueue transferslots
+				, assist $ daemonStatusThread st dstatus
+				, assist $ sanityCheckerThread st dstatus transferqueue changechan
+				, assist $ mountWatcherThread st dstatus scanremotes
+				, assist $ netWatcherThread st dstatus scanremotes
+				, assist $ transferScannerThread st dstatus scanremotes transferqueue
+				, watch $ watchThread st dstatus transferqueue changechan
+				]
+			waitForTermination
+		watch a = (True, a)
+		assist a = (False, a)
+		startthread (watcher, a)
+			| watcher || assistant = void $ forkIO a
+			| otherwise = noop