new threading model in gui_helper

6 files changed, 157 insertions, 98 deletions

diff --git a/portato/gui/gtk/__init__.py b/portato/gui/gtk/__init__.py
index 22e5f76..41161d6 100644
--- a/portato/gui/gtk/__init__.py
+++ b/portato/gui/gtk/__init__.py
@@ -14,7 +14,7 @@ import gtk
 from portato import plugin
 from portato.backend import system
 from windows import MainWindow, SearchWindow, EbuildWindow
-from uncaughtException import register_ex_handler
+from exception_handling import register_ex_handler
 
 def run ():
 	try:
diff --git a/portato/gui/gtk/uncaughtException.py b/portato/gui/gtk/exception_handling.py
index 04f1a6e..e9d19d1 100644
--- a/portato/gui/gtk/uncaughtException.py
+++ b/portato/gui/gtk/exception_handling.py
@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 #
-# File: portato/gui/gtk/uncaughtException.py
+# File: portato/gui/gtk/exception_handling.py
 # This file is part of the Portato-Project, a graphical portage-frontend.
 #
 # Copyright (C) 2007 René 'Necoro' Neumann
@@ -8,24 +8,40 @@
 # the GNU General Public License version 2.
 # There is NO WARRANTY, to the extent permitted by law.
 #
-# Written by Gustavo Carneiro
-# original code: http://www.daa.com.au/pipermail/pygtk/attachments/20030828/2d304204/gtkexcepthook.py
 #
-# Modified by René 'Necoro' Neumann
+# Written by René 'Necoro' Neumann
 
-import sys
-import gtk, pango
-from StringIO import StringIO
-import traceback
+import gtk, pango, gobject
+import sys, traceback
+
+from threading import Thread
 from gettext import lgettext as _
+from StringIO import StringIO
 
 from portato.helper import error
 
-class UncaughExceptionDialog(gtk.MessageDialog):
+class GtkThread (Thread):
+	def run(self):
+		try:
+			Thread.run(self)
+		except SystemExit:
+			raise # let normal thread handle it
+		except:
+			type, val, tb = sys.exc_info()
+			try:
+				try:
+					sys.excepthook(type, val, tb, thread = self.getName())
+				except TypeError:
+					raise type, val, tb # let normal thread handle it
+			finally:
+				del type, val, tb
 
-	def __init__(self, type, value, tb):
+class UncaughtExceptionDialog(gtk.MessageDialog):
+	"""Original idea by Gustavo Carneiro - original code: http://www.daa.com.au/pipermail/pygtk/attachments/20030828/2d304204/gtkexcepthook.py."""
 
-		super(UncaughExceptionDialog,self).__init__(parent=None, flags=0,  type=gtk.MESSAGE_WARNING, buttons=gtk.BUTTONS_NONE, message_format=_("A programming error has been detected during the execution of this program."))
+	def __init__(self, type, value, tb, thread = None):
+
+		super(UncaughtExceptionDialog,self).__init__(parent=None, flags=0,  type=gtk.MESSAGE_WARNING, buttons=gtk.BUTTONS_NONE, message_format=_("A programming error has been detected during the execution of this program."))
 		self.set_title(_("Bug Detected"))
 		self.format_secondary_text(_("It probably isn't fatal, but should be reported to the developers nonetheless."))
 
@@ -49,7 +65,10 @@ class UncaughExceptionDialog(gtk.MessageDialog):
 		self.vbox.add(self.tbFrame)
 		
 		textbuffer = self.textview.get_buffer()
-		textbuffer.set_text(get_trace(type, value, tb))
+		text = get_trace(type, value, tb)
+		if thread:
+			text = _("Exception in thread \"%(thread)s\":\n%(trace)s") % {"thread": thread, "trace": text}
+		textbuffer.set_text(text)
 		self.textview.set_size_request(gtk.gdk.screen_width()/2, gtk.gdk.screen_height()/3)
 
 		self.details = self.tbFrame
@@ -58,7 +77,7 @@ class UncaughExceptionDialog(gtk.MessageDialog):
 
 	def run (self):
 		while True:
-			resp = super(UncaughExceptionDialog, self).run()
+			resp = super(UncaughtExceptionDialog, self).run()
 			if resp == 1:
 				self.details.show_all()
 				self.set_response_sensitive(1, False)
@@ -75,8 +94,15 @@ def get_trace(type, value, tb):
 	
 def register_ex_handler():
 	
-	def handler(*args):
-		error(_("An uncaught exception has occured:\n%s"), get_trace(*args))
-		UncaughExceptionDialog(*args).run()
+	def handler(type, val, tb, thread = None):
+		def run_dialog():
+			UncaughtExceptionDialog(type, val, tb, thread).run()
+		
+		if thread:
+			error(_("Exception in thread \"%(thread)s\":\n%(trace)s"), {"thread": thread, "trace": get_trace(type, val, tb)})
+		else:
+			error(_("Exception:\n%s"), get_trace(type, val, tb))
+		
+		gobject.idle_add(run_dialog)
 
 	sys.excepthook = handler
diff --git a/portato/gui/gtk/windows.py b/portato/gui/gtk/windows.py
index c1789bd..5d61f92 100644
--- a/portato/gui/gtk/windows.py
+++ b/portato/gui/gtk/windows.py
@@ -30,11 +30,11 @@ from portato.gui.gui_helper import Database, Config, EmergeQueue
 from dialogs import *
 from wrapper import GtkTree, GtkConsole
 from usetips import UseTips
+from exception_handling import GtkThread
 
 # other
 import types, logging
 from subprocess import Popen
-from threading import Thread
 from gettext import lgettext as _
 
 gtk.glade.bindtextdomain (APP, LOCALE_DIR)
@@ -1037,7 +1037,7 @@ class MainWindow (Window):
 
 		# set emerge queue
 		self.queueTree = GtkTree(self.queueList.get_model())
-		self.queue = EmergeQueue(console = self.console, tree = self.queueTree, db = self.db, title_update = self.title_update)
+		self.queue = EmergeQueue(console = self.console, tree = self.queueTree, db = self.db, title_update = self.title_update, threadClass = GtkThread)
 	
 		self.window.maximize()
 
@@ -1281,7 +1281,7 @@ class MainWindow (Window):
 				finally:
 					self.window.window.set_cursor(None)
 				
-			Thread(name="Update-Thread", target=__update).start()
+			GtkThread(name="Update-Thread", target=__update).start()
 		
 		return True
 
@@ -1393,7 +1393,7 @@ class MainWindow (Window):
 			gobject.idle_add(cb_idle_watch, packages)
 			gobject.idle_add(cb_idle)
 		
-		Thread(name="Show Updates Thread", target = __update).start()
+		GtkThread(name="Show Updates Thread", target = __update).start()
 		return True
 
 	def cb_right_click (self, object, event):
diff --git a/portato/gui/gui_helper.py b/portato/gui/gui_helper.py
index 062a5c3..ebe0bcf 100644
--- a/portato/gui/gui_helper.py
+++ b/portato/gui/gui_helper.py
@@ -13,7 +13,8 @@
 # some backend things
 from portato import backend
 from portato.backend import flags, system, set_system
-from portato.helper import *
+from portato.helper import debug, info, send_signal_to_group, set_log_level, unique_array
+from portato.waiting_queue import WaitingQueue
 from portato import plugin
 
 # parser
@@ -24,7 +25,7 @@ from wrapper import Console, Tree
 
 # some stuff needed
 from subprocess import Popen, PIPE, STDOUT
-from threading import Thread
+import threading
 import pty
 import time
 import os
@@ -251,7 +252,7 @@ class Database:
 class EmergeQueue:
 	"""This class manages the emerge queue."""
 	
-	def __init__ (self, tree = None, console = None, db = None, title_update = None):
+	def __init__ (self, tree = None, console = None, db = None, title_update = None, threadClass = threading.Thread):
 		"""Constructor.
 		
 		@param tree: Tree to append all the items to.
@@ -270,6 +271,7 @@ class EmergeQueue:
 		
 		# the emerge process
 		self.process = None
+		self.threadQueue = WaitingQueue(threadClass = threadClass)
 
 		# dictionaries with data about the packages in the queue
 		self.iters = {} # iterator in the tree
@@ -448,38 +450,11 @@ class EmergeQueue:
 		else:
 			if cpv not in self.oneshotmerge:
 				self.oneshotmerge.append(cpv)
-	
-	def _update_packages(self, packages):
-		"""This updates the packages-list. It simply makes the db to rebuild the specific category.
-		
-		@param packages: The packages which we emerged.
-		@type packages: list of cpvs"""
-		
-		old_title = self.console.get_window_title()
-		while self.process and self.process.poll() is None:
-			if self.title_update : 
-				title = self.console.get_window_title()
-				if title != old_title:
-					self.title_update(title)
-				time.sleep(0.5)
-
-		if self.title_update: self.title_update(None)
-
-		if self.process is None: # someone resetted this
-			return
-
-		@plugin.hook("after_emerge", packages = packages, retcode = self.process.returncode)
-		def update_packages():
-			for p in packages:
-				if p in ["world", "system"]: continue
-				cat = system.split_cpv(p)[0] # get category
-				self.db.reload(cat)
-				debug("Category %s refreshed", cat)
-
-		update_packages()
-		self.process = None
 
-	def _emerge (self, options, packages, it, command = None):
+	def doEmerge (self, *args, **kwargs):
+		self.threadQueue.put(self.__emerge, *args, **kwargs)
+	
+	def __emerge (self, options, packages, it, command = None):
 		"""Calls emerge and updates the terminal.
 		
 		@param options: options to send to emerge
@@ -491,16 +466,6 @@ class EmergeQueue:
 		@param command: the command to execute - default is "/usr/bin/python /usr/bin/emerge"
 		@type command: string[]"""
 
-		if self.process is not None:
-			def wait():
-				while self.process is not None:
-					time.sleep(0.5)
-
-				self._emerge(options, packages, it, command)
-
-			Thread(name="Waiting-Thread", target=wait).start()
-			return
-
 		@plugin.hook("emerge", packages = packages, command = command, console = self.console, title_update = self.title_update)
 		def sub_emerge(command):
 			if command is None:
@@ -513,14 +478,34 @@ class EmergeQueue:
 			# start emerge
 			self.process = Popen(command+options+packages, stdout = slave, stderr = STDOUT, shell = False, env = system.get_environment())
 			
-			# start thread waiting for the stop of emerge
-			if packages:
-				Thread(name="Emerge-Thread", target=self._update_packages, args=(packages+self.deps.keys(),)).start()
-			
-			# remove
+			# remove packages from queue
 			for i in it:
 				self.remove_with_children(i)
+			
+			# update title
+			old_title = self.console.get_window_title()
+			while self.process and self.process.poll() is None:
+				if self.title_update : 
+					title = self.console.get_window_title()
+					if title != old_title:
+						self.title_update(title)
+					time.sleep(0.5)
+
+			if self.title_update: self.title_update(None)
+
+			if self.process is None: # someone resetted this
+				return
+
+			@plugin.hook("after_emerge", packages = packages, retcode = self.process.returncode)
+			def update_packages():
+				for cat in unique_array([system.split_cpv(p)[0] for p in packages if p not in ["world", "system"]]):
+					self.db.reload(cat)
+					debug("Category %s refreshed", cat)
 
+			update_packages()
+			self.process = None
+			self.threadQueue.next()
+			
 		sub_emerge(command)
 
 	def emerge (self, force = False, options = None):
@@ -550,7 +535,7 @@ class EmergeQueue:
 			if not force: s += system.get_pretend_option()
 			if options is not None: s += options
 			
-			self._emerge(s, list, its)
+			self.doEmerge(s, list, its, caller = self.emerge)
 		
 		# normal queue
 		if self.mergequeue:
@@ -561,7 +546,7 @@ class EmergeQueue:
 			if not force: s = system.get_pretend_option()
 			if options is not None: s += options
 		
-			self._emerge(s, list, its)
+			self.doEmerge(s, list, its, caller = self.emerge)
 
 	def unmerge (self, force = False, options = None):
 		"""Unmerges everything in the umerge-queue.
@@ -580,7 +565,7 @@ class EmergeQueue:
 		if not force: s += system.get_pretend_option()
 		if options is not None: s += options
 		
-		self._emerge(s,list, [self.unmergeIt])
+		self.doEmerge(s,list, [self.unmergeIt], caller = self.unmerge)
 
 	def update_world(self, force = False, newuse = False, deep = False, options = None):
 		"""Does an update world. newuse and deep are the arguments handed to emerge.
@@ -601,7 +586,7 @@ class EmergeQueue:
 		if not force: opts += system.get_pretend_option()
 		if options is not None: opts += options
 
-		self._emerge(opts, ["world"], [self.emergeIt])
+		self.doEmerge(opts, ["world"], [self.emergeIt], caller = self.update_world)
 
 	def sync (self, command = None):
 		"""Calls "emerge --sync".
@@ -611,32 +596,19 @@ class EmergeQueue:
 
 		if command is None:
 			command = system.get_sync_command()
-		
-		def threaded_sync (cmd):
-			ret = self.process.wait()
-			self.process = None
-			if ret == 0:
-				__sync(cmd, False)
-		
-		def __sync(cmd, startThread = True):
-			try:
-				idx = cmd.index("&&")
-			except ValueError: # no && in there -> normal behavior
-				self._emerge([],[],[], command = cmd)
-			else:
-				self._emerge([],[],[], command = cmd[:idx])
-
-				if startThread:
-					Thread(name = "SyncThread", target = threaded_sync, args = (cmd[idx+1:],)).start()
-				else:
-					threaded_sync(cmd[idx+1:])
-
-		__sync(command)
-
+	
+		try:
+			while True:
+				idx = command.index("&&")
+				self.doEmerge([],[],[], command[:idx], caller = self.sync)
+				command = command[idx+1:]
+		except ValueError: # no && in command
+			self.doEmerge([],[],[], command, caller = self.sync)
 
 	def kill_emerge (self):
 		"""Kills the emerge process."""
 		if self.process is not None:
+			self.threadQueue.clear() # remove all pending emerge threads
 			try:
 				send_signal_to_group(signal.SIGTERM)
 				debug("Process should be killed")
diff --git a/portato/helper.py b/portato/helper.py
index abb91f9..754f566 100644
--- a/portato/helper.py
+++ b/portato/helper.py
@@ -14,7 +14,7 @@
 Some nice functions used in the program.
 """
 
-import types, os, signal, logging
+import os, signal, logging
 
 debug 		= logging.getLogger("portatoLogger").debug
 info 		= logging.getLogger("portatoLogger").info
@@ -66,7 +66,7 @@ def flatten (listOfLists):
 	@returns: flattend list
 	@rtype: list"""
 
-	if type(listOfLists) != types.ListType:
+	if not isinstance(listOfLists, list):
 		return [listOfLists]
 
 	ret = []
diff --git a/portato/waiting_queue.py b/portato/waiting_queue.py
new file mode 100644
index 0000000..409843c
--- /dev/null
+++ b/portato/waiting_queue.py
@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+#
+# File: portato/waiting_queue.py
+# This file is part of the Portato-Project, a graphical portage-frontend.
+#
+# Copyright (C) 2007 René 'Necoro' Neumann
+# This is free software.  You may redistribute copies of it under the terms of
+# the GNU General Public License version 2.
+# There is NO WARRANTY, to the extent permitted by law.
+#
+# Written by René 'Necoro' Neumann <necoro@necoro.net>
+
+from threading import Thread, Event
+from Queue import Queue
+
+class WaitingQueue (Queue):
+
+	def __init__ (self, setTrue = True, threadClass = Thread):
+		if not issubclass(threadClass, Thread):
+			raise ValueError, "Only subclasses of threading.Thread are allowed."
+		
+		Queue.__init__(self)
+		self.event = Event()
+		self.counter = 0
+		self.threadClass = threadClass
+
+		if setTrue:
+			self.event.set() # true at the beginning
+
+		waitingThread = self.threadClass(name = "Waiting-Queue-Thread", target = self.runThread)
+		waitingThread.setDaemon(True)
+		waitingThread.start()
+
+	def put (self, method, *args, **kwargs):
+		self.counter += 1;
+		
+		if "caller" in kwargs:
+			name = "Waiting Thread #%d (called by:%s)" % (self.counter, kwargs["caller"])
+			del kwargs["caller"]
+		else:
+			name = "Waiting Thread #%d" % self.counter
+
+		t = self.threadClass(name = name, target = method, args = args, kwargs = kwargs)
+		t.setDaemon(True)
+		Queue.put(self, t, False)
+
+	def runThread (self):
+		while True:
+			self.event.wait()
+			t = self.get(True)
+			self.event.clear()
+			t.run()
+
+	def next (self):
+		self.event.set()
+
+	def clear (self):
+		self.mutex.acquire()
+		self.queue.clear()
+		self.mutex.release()
+		self.event.set()