Merge lp:~cmiller/desktopcouch/mother-forking-threads into lp:desktopcouch

[Chad Miller]

Split desktopcouch-service threads into to discrete processes, since they do not depend on each other, and because multithreading breaks with recent changes to libraries. (LP: #524566, #530605)

For places we use DBus function calls, pull the DBus glib mainloop setting close to the place we assign the mainloop to the twisted reactor.

[Tim Cole (tcole)]

Cool, +1 on processes versus threads.

review: Approve

[Tim Cole (tcole)]

(It may also be worth thinking about having the parent restart the child if it dies.)

[Rick McBride (rmcbride)]

I like it. I did see mention that Joshua Hoover was going to review this (after I started my review) so I'll defer changing the merge request status pending his review.

review: Approve

[Joshua Hoover (joshuahoover)]

I tested this yesterday the best I could based on the bug comments. As long as Tim and Rick are cool with it then we should get it released ASAP.

=== modified file 'bin/desktopcouch-pair'
--- bin/desktopcouch-pair	2009-11-30 12:42:47 +0000
+++ bin/desktopcouch-pair	2010-04-05 22:05:31 +0000
@@ -56,6 +56,8 @@
 
 from twisted.internet import gtk2reactor
 gtk2reactor.install()
+from dbus.mainloop.glib import DBusGMainLoop
+DBusGMainLoop(set_as_default=True)
 from twisted.internet.reactor import run as run_program
 from twisted.internet.reactor import stop as stop_program
 

=== modified file 'bin/desktopcouch-service'
--- bin/desktopcouch-service	2010-02-24 19:46:40 +0000
+++ bin/desktopcouch-service	2010-04-05 22:05:31 +0000
@@ -34,15 +34,19 @@
 
 """
 
+import os
+import time
 import logging
 import logging.handlers
-from errno import ENOENT
+import signal
 
 from twisted.internet import glib2reactor
 glib2reactor.install()
+from dbus.mainloop.glib import DBusGMainLoop
+DBusGMainLoop(set_as_default=True)
+
 from twisted.internet import reactor as mainloop
-import dbus.service, gobject, os, errno, time
-from dbus import DBusException
+import dbus.service
 
 import desktopcouch
 from desktopcouch import local_files
@@ -75,54 +79,18 @@
         self.death()
 
 
-def main(ctx=local_files.DEFAULT_CONTEXT):
-    pid = desktopcouch.find_pid(start_if_not_running=False, ctx=ctx)
-
-    should_shut_down = False
-    if pid is None:
-        logging.warn("Starting up personal couchdb.")
-        pid = desktopcouch.find_pid(start_if_not_running=True, ctx=ctx)
-        should_shut_down = True
-    else:
-        logging.warn("Personal couchdb is already running at PID#%d.", pid)
-
-    try:
-        port = desktopcouch._direct_access_find_port(pid=pid, ctx=ctx)
-
-        # Advertise the port
-        replication_runtime = replication.set_up(lambda: port)
-        portAdvertiser = PortAdvertiser(mainloop.stop)  # TODO: send port.
-        try:
-            logging.debug("starting main loop")
-            mainloop.run()
-            logging.debug("ending main loop")
-
-        finally:
-            if replication_runtime:
-                replication.tear_down(*replication_runtime)
-
-    except:
-        logging.exception("uncaught exception makes us shut down.")
-    finally:
-        logging.info("exiting.")
-        if should_shut_down:
-            logging.warn("shutting down personal couchdb.")
-            stop_local_couchdb.stop_couchdb(ctx=ctx)
-
-
-if __name__ == "__main__":
+def set_up_logging(name):
+    """Set logging preferences for this process."""
     import xdg.BaseDirectory
-    import gobject
-    gobject.set_application_name("desktopcouch service")
-
     log_directory = os.path.join(xdg.BaseDirectory.xdg_cache_home,
             "desktop-couch/log")
     try:
         os.makedirs(log_directory)
     except:
         pass
+
     rotating_log = logging.handlers.TimedRotatingFileHandler(
-            os.path.join(log_directory, "desktop-couch-replication.log"),
+            os.path.join(log_directory, "desktop-couch-%s.log" % (name,)),
             "midnight", 1, 14)
     rotating_log.setLevel(logging.DEBUG)
     formatter = logging.Formatter('%(asctime)s %(levelname)-8s %(message)s')
@@ -130,8 +98,78 @@
     logging.getLogger('').addHandler(rotating_log)
     console_log = logging.StreamHandler()
     console_log.setLevel(logging.WARNING)
-    console_log.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
+    console_log.setFormatter(logging.Formatter(
+            "%s %%(asctime)s - %%(message)s" % (name,)))
     logging.getLogger('').addHandler(console_log)
     logging.getLogger('').setLevel(logging.DEBUG)
 
-    main()
+
+def replicator_main(couchdb_port, ctx=local_files.DEFAULT_CONTEXT):
+    replication_runtime = replication.set_up(lambda: couchdb_port)
+    try:
+        logging.debug("starting replicator main loop")
+        mainloop.run()
+    finally:
+        logging.debug("ending replicator main loop")
+        if replication_runtime:
+            replication.tear_down(*replication_runtime)
+        
+
+def dbus_server_main(ctx=local_files.DEFAULT_CONTEXT):
+    portAdvertiser = PortAdvertiser(mainloop.stop)
+    logging.debug("starting dbus main loop")
+    try:
+        mainloop.run()
+    finally:
+        logging.debug("ending dbus main loop")
+
+
+def main(ctx=local_files.DEFAULT_CONTEXT):
+    should_shut_down_couchdb = False
+    couchdb_pid = desktopcouch.find_pid(start_if_not_running=False, ctx=ctx)
+    if couchdb_pid is None:
+        logging.warn("Starting up personal couchdb.")
+        couchdb_pid = desktopcouch.find_pid(start_if_not_running=True, ctx=ctx)
+        should_shut_down_couchdb = True
+    else:
+        logging.warn("Personal couchdb is already running at PID#%d.",
+                couchdb_pid)
+
+    couchdb_port = desktopcouch._direct_access_find_port(pid=couchdb_pid,
+            ctx=ctx)
+    child_pid = os.fork()  # Split!
+    if child_pid != 0:
+        # Let's be the replicator!
+        set_up_logging("replication")
+        #os.nice(10)
+        replicator_main(couchdb_port)
+        return
+    else:
+        # Let's be the dbus server!  This is the main process.  When we exit, 
+        # we kill everything.
+        try:
+            set_up_logging("dbus")
+            dbus_server_main()
+        except:
+            logging.exception("uncaught exception makes us shut down.")
+        finally:
+            logging.info("exiting.")
+            if should_shut_down_couchdb:
+                logging.warn("shutting down personal couchdb.")
+                stop_local_couchdb.stop_couchdb(ctx=ctx)
+                
+            try:
+                os.kill(child_pid, signal.SIGTERM)
+                time.sleep(1)
+                os.kill(child_pid, signal.SIGKILL)
+            except OSError:
+                pass  
+
+            return
+
+
+if __name__ == "__main__":
+    import gobject
+    gobject.set_application_name("desktopcouch service")
+
+    main() 

=== modified file 'desktopcouch/pair/couchdb_pairing/dbus_io.py'
--- desktopcouch/pair/couchdb_pairing/dbus_io.py	2010-02-23 16:15:01 +0000
+++ desktopcouch/pair/couchdb_pairing/dbus_io.py	2010-04-05 22:05:31 +0000
@@ -20,9 +20,7 @@
 import logging
 
 import dbus
-from dbus.mainloop.glib import DBusGMainLoop
 import avahi
-DBusGMainLoop(set_as_default=True)
 
 from desktopcouch.pair.couchdb_pairing import couchdb_io
 

=== modified file 'desktopcouch/replication.py'
--- desktopcouch/replication.py	2010-02-23 16:36:20 +0000
+++ desktopcouch/replication.py	2010-04-05 22:05:31 +0000
@@ -34,7 +34,6 @@
 
 
 known_bad_service_names = set()
-already_replicating = False
 is_running = True
 
 
@@ -72,9 +71,6 @@
     log.debug("started replicating")
     local_uri = couchdb_io.mkuri("localhost", local_port)
     try:
-        global already_replicating  # Fuzzy, as not really critical,
-        already_replicating = True  #   just trying to be polite.
-
         try:
             # All machines running desktopcouch must advertise themselves with
             # zeroconf.  We collect those elsewhere and filter out the ones
@@ -203,16 +199,8 @@
             log.exception("replication of services aborted")
             pass
     finally:
-        already_replicating = False
-    log.debug("finished replicating")
-
-
-def replicate_local_databases_to_paired_hosts(local_port):
-    if already_replicating:
-        log.warn("haven't finished replicating before next time to start.")
-        return False
-
-    reactor.callInThread(do_all_replication, local_port)
+        log.debug("finished replicating")
+
 
 def set_up(port_getter):
     port = port_getter()
@@ -231,7 +219,7 @@
 
     dbus_io.maintain_discovered_servers()
 
-    t = task.LoopingCall(replicate_local_databases_to_paired_hosts, int(port))
+    t = task.LoopingCall(do_all_replication, int(port))
     t.start(600)
 
     # TODO:  port may change, so every so often, check it and