/head: changeset 60

Start of support for multi-threading

head/offlineimap/imapserver.py

@@ -17,9 +17,34 @@
 #    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
 from offlineimap import imaplib, imaputil
+from threading import *
+
+class UsefulIMAPMixIn:
+    def getstate(self):
+        return self.state
+    def getselectedfolder(self):
+        if self.getstate() == 'SELECTED':
+            return self.selectedfolder
+        return None
+
+    def select(self, mailbox='INBOX', readonly=None):
+        if self.getselectedfolder() == mailbox and not readonly:
+            # No change; return.
+            return
+        result = self.__class__.__bases__[1].select(self, mailbox, readonly)
+        if result[0] != 'OK':
+            raise ValueError, "Error from select: %s" % str(result)
+        if self.getstate() == 'SELECTED' and not readonly:
+            self.selectedfolder = mailbox
+        else:
+            self.selectedfolder = None
+
+class UsefulIMAP4(UsefulIMAPMixIn, imaplib.IMAP4): pass
+class UsefulIMAP4_SSL(UsefulIMAPMixIn, imaplib.IMAP4_SSL): pass
 
 class IMAPServer:
-    def __init__(self, username, password, hostname, port = None, ssl = 1):
+    def __init__(self, username, password, hostname, port = None, ssl = 1,
+                 maxconnections = 1):
         self.username = username
         self.password = password
         self.hostname = hostname
@@ -32,30 +57,57 @@ class IMAPServer:
                 self.port = 993
             else:
                 self.port = 143
-        self.imapobj = None
+        self.maxconnections = maxconnections
+        self.availableconnections = []
+        self.assignedconnections = []
+        self.semaphore = BoundedSemaphore(self.maxconnections)
+        self.connectionlock = Lock()
+        
 
     def getdelim(self):
         """Returns this server's folder delimiter.  Can only be called
-        after one or more calls to makeconnection."""
+        after one or more calls to acquireconnection."""
         return self.delim
 
     def getroot(self):
         """Returns this server's folder root.  Can only be called after one
-        or more calls to makeconnection."""
+        or more calls to acquireconnection."""
         return self.root
 
-    def makeconnection(self):
-        """Opens a connection to the server and returns an appropriate
+
+    def releaseconnection(self, connection):
+        self.connectionlock.acquire()
+        self.assignedconnections.remove(connection)
+        self.availableconnections.append(connection)
+        self.connectionlock.release()
+        self.semaphore.release()
+            
+
+    def acquireconnection(self):
+        """Fetches a connection from the pool, making sure to create a new one
+        if needed, to obey the maximum connection limits, etc.
+        Opens a connection to the server and returns an appropriate
         object."""
 
-        if self.imapobj != None:
-            return self.imapobj
+        self.semaphore.acquire()
+        self.connectionlock.acquire()
         imapobj = None
-        if self.usessl:
-            imapobj = imaplib.IMAP4_SSL(self.hostname, self.port)
-        else:
-            imapobj = imaplib.IMAP4(self.hostname, self.port)
 
+        if len(self.availableconnections): # One is available.
+            imapobj = self.availableconnections[0]
+            self.assignedconnections.append(imapobj)
+            del(self.availableconnections[0])
+            self.connectionlock.release()
+            return imapobj
+        
+        self.connectionlock.release()   # Release until need to modify data
+
+        # Generate a new connection.
+        if self.usessl:
+            imapobj = UsefulIMAP4_SSL(self.hostname, self.port)
+        else:
+            imapobj = UsefulIMAP4(self.hostname, self.port)
+            
         imapobj.login(self.username, self.password)
 
         if self.delim == None:
@@ -64,9 +116,34 @@ class IMAPServer:
             self.delim = imaputil.dequote(self.delim)
             self.root = imaputil.dequote(self.root)
 
-        self.imapobj = imapobj
+        self.connectionlock.acquire()
+        self.assignedconnections.append(imapobj)
+        self.connectionlock.release()
         return imapobj
     
+    def connectionwait(self):
+        """Waits until there is a connection available.  Note that between
+        the time that a connection becomes available and the time it is
+        requested, another thread may have grabbed it.  This function is
+        mainly present as a way to avoid spawning thousands of threads
+        to copy messages, then have them all wait for 3 available connections.
+        It's OK if we have maxconnections + 1 or 2 threads, which is what
+        this will help us do."""
+        self.semaphore.acquire()
+        self.semaphore.release()
+
     def close(self):
-        self.imapobj.logout()
-        self.imapobj = None
+        # Make sure I own all the semaphores.  Let the threads finish
+        # their stuff.  This is a blocking method.
+        self.connectionlock.acquire()
+        for i in range(self.maxconnections):
+            self.semaphore.acquire()
+        for imapobj in self.assignedconnections + self.availableconnections:
+            imapobj.logout()
+        self.assignedconnections = []
+        self.availableconnections = []
+        for i in range(self.maxconnections):
+            self.semaphore.release()
+        self.connectionlock.release()
+        
+

head/offlineimap/repository/IMAP.py

@@ -19,6 +19,7 @@
 from Base import BaseRepository
 from offlineimap import folder, imaputil
 import re
+from threading import *
 
 class IMAPRepository(BaseRepository):
     def __init__(self, config, accountname, imapserver):
@@ -30,6 +31,7 @@ class IMAPRepository(BaseRepository):
         self.imapobj = imapserver.makeconnection()
         self.folders = None
         self.nametrans = lambda foldername: foldername
+        self.maxconnections = config.getint(accountname, 'maxconnections')
         if config.has_option(accountname, 'nametrans'):
             self.nametrans = eval(config.get(accountname, 'nametrans'))