Better download connection handling: Detect server error earlier, fallback to single connection if possible

8 files changed, 150 insertions, 51 deletions

diff --git a/module/Api.py b/module/Api.py
index ba79a31ef..fc36c9fea 100644
--- a/module/Api.py
+++ b/module/Api.py
@@ -29,9 +29,13 @@ from network.RequestFactory import getURL
 from remote import activated
 
 if activated:
-    from remote.thriftbackend.thriftgen.pyload.ttypes import *
-    from remote.thriftbackend.thriftgen.pyload.Pyload import Iface
-    BaseObject = TBase
+    try:
+        from remote.thriftbackend.thriftgen.pyload.ttypes import *
+        from remote.thriftbackend.thriftgen.pyload.Pyload import Iface
+        BaseObject = TBase
+    except ImportError:
+        print "Thrift not imported"
+        from remote.socketbackend.ttypes import *
 else:
     from remote.socketbackend.ttypes import *
 
diff --git a/module/network/HTTPChunk.py b/module/network/HTTPChunk.py
index 69eedb19c..582067aa8 100644
--- a/module/network/HTTPChunk.py
+++ b/module/network/HTTPChunk.py
@@ -16,7 +16,7 @@
     
     @author: RaNaN
 """
-from os import remove, stat
+from os import remove, stat, fsync
 from os.path import exists
 from time import sleep
 from re import search
@@ -146,6 +146,9 @@ class HTTPChunk(HTTPRequest):
         self.sleep = 0.000
         self.lastSize = 0
 
+    def __repr__(self):
+        return "<HTTPChunk id=%d, size=%d, arrived=%d>" % (self.id, self.size, self.arrived)
+
     @property
     def cj(self):
         return self.p.cj
@@ -157,7 +160,7 @@ class HTTPChunk(HTTPRequest):
         self.c.setopt(pycurl.WRITEFUNCTION, self.writeBody)
         self.c.setopt(pycurl.HEADERFUNCTION, self.writeHeader)
 
-        # request one byte more, since some servers in russia seems to have a defect arihmetic unit
+        # request all bytes, since some servers in russia seems to have a defect arihmetic unit
 
         if self.resume:
             self.fp = open(self.p.info.getChunkName(self.id), "ab")
@@ -259,10 +262,25 @@ class HTTPChunk(HTTPRequest):
 
         self.headerParsed = True
 
+    def stop(self):
+        """The download will not proceed after next call of writeBody"""
+        self.range = [0,0]
+        self.size = 0
+
+    def resetRange(self):
+        """ Reset the range, so the download will load all data available  """
+        self.range = None
+
     def setRange(self, range):
         self.range = range
         self.size = range[1] - range[0]
 
+    def flushFile(self):
+        """  flush and close file """
+        self.fp.flush()
+        fsync(self.fp.fileno()) #make sure everything was written to disk
+        self.fp.close() #needs to be closed, or merging chunks will fail
+
     def close(self):
         """ closes everything, unusable after this """
         if self.fp: self.fp.close()
diff --git a/module/network/HTTPDownload.py b/module/network/HTTPDownload.py
index 1a2886332..13c674833 100644
--- a/module/network/HTTPDownload.py
+++ b/module/network/HTTPDownload.py
@@ -140,7 +140,7 @@ class HTTPDownload():
 
                 return self._download(chunks, False)
             else:
-                raise e
+                raise
         finally:
             self.close()
 
@@ -161,7 +161,7 @@ class HTTPDownload():
 
         lastFinishCheck = 0
         lastTimeCheck = 0
-        chunksDone = set()
+        chunksDone = set()  # list of curl handles that are finished
         chunksCreated = False
         done = False
         if self.info.getCount() > 1: # This is a resume, if we were chunked originally assume still can
@@ -202,32 +202,76 @@ class HTTPDownload():
             t = time()
 
             # reduce these calls
-            while lastFinishCheck + 1 < t:
+            while lastFinishCheck + 0.5 < t:
+                # list of failed curl handles
+                failed = []
+                ex = None # save only last exception, we can only raise one anyway
+
                 num_q, ok_list, err_list = self.m.info_read()
                 for c in ok_list:
-                    chunksDone.add(c)
+                    chunk = self.findChunk(c)
+                    try: # check if the header implies success, else add it to failed list
+                        chunk.verifyHeader()
+                    except BadHeader, e:
+                        self.log.debug("Chunk %d failed: %s" % (chunk.id + 1, str(e)))
+                        failed.append(chunk)
+                        ex = e
+                    else:
+                        chunksDone.add(c)
+
                 for c in err_list:
                     curl, errno, msg = c
-                    #test if chunk was finished, otherwise raise the exception
+                    chunk = self.findChunk(curl)
+                    #test if chunk was finished
                     if errno != 23 or "0 !=" not in msg:
-                        raise pycurl.error(errno, msg)
-
-                    #@TODO KeyBoardInterrupts are seen as finished chunks,
-                    #but normally not handled to this process, only in the testcase
+                        failed.append(chunk)
+                        ex = pycurl.error(errno, msg)
+                        self.log.debug("Chunk %d failed: %s" % (chunk.id + 1, str(ex)))
+                        continue
+
+                    try: # check if the header implies success, else add it to failed list
+                        chunk.verifyHeader()
+                    except BadHeader, e:
+                        self.log.debug("Chunk %d failed: %s" % (chunk.id + 1, str(e)))
+                        failed.append(chunk)
+                        ex = e
+                    else:
+                        chunksDone.add(curl)
+                if not num_q: # no more infos to get
+
+                    # check if init is not finished so we reset download connections
+                    # note that other chunks are closed and downloaded with init too
+                    if failed and init not in failed and init.c not in chunksDone:
+                        self.log.error(_("Download chunks failed, fallback to single connection | %s" % (str(ex))))
+
+                        #list of chunks to clean and remove
+                        to_clean = filter(lambda x: x is not init, self.chunks)
+                        for chunk in to_clean:
+                            self.closeChunk(chunk)
+                            self.chunks.remove(chunk)
+                            remove(self.info.getChunkName(chunk.id))
+
+                        #let first chunk load the rest and update the info file
+                        init.resetRange()
+                        self.info.clear()
+                        self.info.addChunk("%s.chunk0" % self.filename, (0, self.size))
+                        self.info.save()
+                    elif failed:
+                        raise ex
 
-                    chunksDone.add(curl)
-                if not num_q:
                     lastFinishCheck = t
 
-                    if len(chunksDone) == len(self.chunks):
-                        done = True #all chunks loaded
+                    if len(chunksDone) >= len(self.chunks):
+                        if len(chunksDone) > len(self.chunks):
+                            self.log.warning("Finished download chunks size incorrect, please report bug.")
+                        done = True  #all chunks loaded
 
                     break
 
             if done:
                 break #all chunks loaded
 
-            # calc speed once per second
+            # calc speed once per second, averaging over 3 seconds
             if lastTimeCheck + 1 < t:
                 diff = [c.arrived - (self.lastArrived[i] if len(self.lastArrived) > i else 0) for i, c in
                         enumerate(self.chunks)]
@@ -247,15 +291,7 @@ class HTTPDownload():
 
         failed = False
         for chunk in self.chunks:
-            try:
-                chunk.verifyHeader()
-            except BadHeader, e:
-                failed = e.code
-                remove(self.info.getChunkName(chunk.id))
-
-            chunk.fp.flush()
-            fsync(chunk.fp.fileno()) #make sure everything was written to disk
-            chunk.fp.close() #needs to be closed, or merging chunks will fail
+            chunk.flushFile() #make sure downloads are written to disk
 
         if failed: raise BadHeader(failed)
 
@@ -265,11 +301,16 @@ class HTTPDownload():
         if self.progressNotify:
             self.progressNotify(self.percent)
 
+    def findChunk(self, handle):
+        """ linear search to find a chunk (should be ok since chunk size is usually low) """
+        for chunk in self.chunks:
+            if chunk.c == handle: return chunk
+
     def closeChunk(self, chunk):
         try:
             self.m.remove_handle(chunk.c)
-        except pycurl.error:
-            self.log.debug("Error removing chunk")
+        except pycurl.error, e:
+            self.log.debug("Error removing chunk: %s" % str(e))
         finally:
             chunk.close()
 
diff --git a/module/network/HTTPRequest.py b/module/network/HTTPRequest.py
index bd8cdd72e..e58fd114e 100644
--- a/module/network/HTTPRequest.py
+++ b/module/network/HTTPRequest.py
@@ -30,6 +30,7 @@ from module.plugins.Plugin import Abort
 def myquote(url):
     return quote(url, safe="%/:=&?~#+!$,;'@()*[]")
 
+bad_headers = range(400, 404) + range(405, 418) + range(500, 506)
 
 class BadHeader(Exception):
     def __init__(self, code, content=""):
@@ -211,11 +212,15 @@ class HTTPRequest():
     def verifyHeader(self):
         """ raise an exceptions on bad headers """
         code = int(self.c.getinfo(pycurl.RESPONSE_CODE))
-        if code in range(400, 404) or code in range(405, 418) or code in range(500, 506):
+        if code in bad_headers:
             #404 will NOT raise an exception
             raise BadHeader(code, self.getResponse())
         return code
 
+    def checkHeader(self):
+        """ check if header indicates failure"""
+        return int(self.c.getinfo(pycurl.RESPONSE_CODE)) not in bad_headers
+
     def getResponse(self):
         """ retrieve response from string io """
         if self.rep is None: return ""
diff --git a/module/remote/socketbackend/create_ttypes.py b/module/remote/socketbackend/create_ttypes.py
index 1bf8856a2..05662cb50 100644
--- a/module/remote/socketbackend/create_ttypes.py
+++ b/module/remote/socketbackend/create_ttypes.py
@@ -68,7 +68,7 @@ class BaseObject(object):
         #create init
         args = ["self"] + ["%s=None" % x for x in klass.__slots__]
 
-        f.write("\tdef init(%s):\n" % ", ".join(args))
+        f.write("\tdef __init__(%s):\n" % ", ".join(args))
         for attr in klass.__slots__:
             f.write("\t\tself.%s = %s\n" % (attr, attr))
 
diff --git a/module/remote/socketbackend/ttypes.py b/module/remote/socketbackend/ttypes.py
index 58e638689..f8ea121fa 100644
--- a/module/remote/socketbackend/ttypes.py
+++ b/module/remote/socketbackend/ttypes.py
@@ -31,10 +31,27 @@ class ElementType:
 	File = 1
 	Package = 0
 
+class Input:
+	BOOL = 4
+	CHOICE = 6
+	CLICK = 5
+	LIST = 8
+	MULTIPLE = 7
+	NONE = 0
+	PASSWORD = 3
+	TABLE = 9
+	TEXT = 1
+	TEXTBOX = 2
+
+class Output:
+	CAPTCHA = 1
+	NOTIFICATION = 4
+	QUESTION = 2
+
 class AccountInfo(BaseObject):
 	__slots__ = ['validuntil', 'login', 'options', 'valid', 'trafficleft', 'maxtraffic', 'premium', 'type']
 
-	def init(self, validuntil=None, login=None, options=None, valid=None, trafficleft=None, maxtraffic=None, premium=None, type=None):
+	def __init__(self, validuntil=None, login=None, options=None, valid=None, trafficleft=None, maxtraffic=None, premium=None, type=None):
 		self.validuntil = validuntil
 		self.login = login
 		self.options = options
@@ -47,7 +64,7 @@ class AccountInfo(BaseObject):
 class CaptchaTask(BaseObject):
 	__slots__ = ['tid', 'data', 'type', 'resultType']
 
-	def init(self, tid=None, data=None, type=None, resultType=None):
+	def __init__(self, tid=None, data=None, type=None, resultType=None):
 		self.tid = tid
 		self.data = data
 		self.type = type
@@ -56,7 +73,7 @@ class CaptchaTask(BaseObject):
 class ConfigItem(BaseObject):
 	__slots__ = ['name', 'description', 'value', 'type']
 
-	def init(self, name=None, description=None, value=None, type=None):
+	def __init__(self, name=None, description=None, value=None, type=None):
 		self.name = name
 		self.description = description
 		self.value = value
@@ -65,7 +82,7 @@ class ConfigItem(BaseObject):
 class ConfigSection(BaseObject):
 	__slots__ = ['name', 'description', 'items', 'outline']
 
-	def init(self, name=None, description=None, items=None, outline=None):
+	def __init__(self, name=None, description=None, items=None, outline=None):
 		self.name = name
 		self.description = description
 		self.items = items
@@ -74,7 +91,7 @@ class ConfigSection(BaseObject):
 class DownloadInfo(BaseObject):
 	__slots__ = ['fid', 'name', 'speed', 'eta', 'format_eta', 'bleft', 'size', 'format_size', 'percent', 'status', 'statusmsg', 'format_wait', 'wait_until', 'packageID', 'packageName', 'plugin']
 
-	def init(self, fid=None, name=None, speed=None, eta=None, format_eta=None, bleft=None, size=None, format_size=None, percent=None, status=None, statusmsg=None, format_wait=None, wait_until=None, packageID=None, packageName=None, plugin=None):
+	def __init__(self, fid=None, name=None, speed=None, eta=None, format_eta=None, bleft=None, size=None, format_size=None, percent=None, status=None, statusmsg=None, format_wait=None, wait_until=None, packageID=None, packageName=None, plugin=None):
 		self.fid = fid
 		self.name = name
 		self.speed = speed
@@ -95,7 +112,7 @@ class DownloadInfo(BaseObject):
 class EventInfo(BaseObject):
 	__slots__ = ['eventname', 'id', 'type', 'destination']
 
-	def init(self, eventname=None, id=None, type=None, destination=None):
+	def __init__(self, eventname=None, id=None, type=None, destination=None):
 		self.eventname = eventname
 		self.id = id
 		self.type = type
@@ -104,7 +121,7 @@ class EventInfo(BaseObject):
 class FileData(BaseObject):
 	__slots__ = ['fid', 'url', 'name', 'plugin', 'size', 'format_size', 'status', 'statusmsg', 'packageID', 'error', 'order']
 
-	def init(self, fid=None, url=None, name=None, plugin=None, size=None, format_size=None, status=None, statusmsg=None, packageID=None, error=None, order=None):
+	def __init__(self, fid=None, url=None, name=None, plugin=None, size=None, format_size=None, status=None, statusmsg=None, packageID=None, error=None, order=None):
 		self.fid = fid
 		self.url = url
 		self.name = name
@@ -120,20 +137,34 @@ class FileData(BaseObject):
 class FileDoesNotExists(Exception):
 	__slots__ = ['fid']
 
-	def init(self, fid=None):
+	def __init__(self, fid=None):
 		self.fid = fid
 
+class InteractionTask(BaseObject):
+	__slots__ = ['iid', 'input', 'structure', 'preset', 'output', 'data', 'title', 'description', 'plugin']
+
+	def __init__(self, iid=None, input=None, structure=None, preset=None, output=None, data=None, title=None, description=None, plugin=None):
+		self.iid = iid
+		self.input = input
+		self.structure = structure
+		self.preset = preset
+		self.output = output
+		self.data = data
+		self.title = title
+		self.description = description
+		self.plugin = plugin
+
 class OnlineCheck(BaseObject):
 	__slots__ = ['rid', 'data']
 
-	def init(self, rid=None, data=None):
+	def __init__(self, rid=None, data=None):
 		self.rid = rid
 		self.data = data
 
 class OnlineStatus(BaseObject):
 	__slots__ = ['name', 'plugin', 'packagename', 'status', 'size']
 
-	def init(self, name=None, plugin=None, packagename=None, status=None, size=None):
+	def __init__(self, name=None, plugin=None, packagename=None, status=None, size=None):
 		self.name = name
 		self.plugin = plugin
 		self.packagename = packagename
@@ -143,7 +174,7 @@ class OnlineStatus(BaseObject):
 class PackageData(BaseObject):
 	__slots__ = ['pid', 'name', 'folder', 'site', 'password', 'dest', 'order', 'linksdone', 'sizedone', 'sizetotal', 'linkstotal', 'links', 'fids']
 
-	def init(self, pid=None, name=None, folder=None, site=None, password=None, dest=None, order=None, linksdone=None, sizedone=None, sizetotal=None, linkstotal=None, links=None, fids=None):
+	def __init__(self, pid=None, name=None, folder=None, site=None, password=None, dest=None, order=None, linksdone=None, sizedone=None, sizetotal=None, linkstotal=None, links=None, fids=None):
 		self.pid = pid
 		self.name = name
 		self.folder = folder
@@ -161,13 +192,13 @@ class PackageData(BaseObject):
 class PackageDoesNotExists(Exception):
 	__slots__ = ['pid']
 
-	def init(self, pid=None):
+	def __init__(self, pid=None):
 		self.pid = pid
 
 class ServerStatus(BaseObject):
 	__slots__ = ['pause', 'active', 'queue', 'total', 'speed', 'download', 'reconnect']
 
-	def init(self, pause=None, active=None, queue=None, total=None, speed=None, download=None, reconnect=None):
+	def __init__(self, pause=None, active=None, queue=None, total=None, speed=None, download=None, reconnect=None):
 		self.pause = pause
 		self.active = active
 		self.queue = queue
@@ -179,7 +210,7 @@ class ServerStatus(BaseObject):
 class ServiceCall(BaseObject):
 	__slots__ = ['plugin', 'func', 'arguments', 'parseArguments']
 
-	def init(self, plugin=None, func=None, arguments=None, parseArguments=None):
+	def __init__(self, plugin=None, func=None, arguments=None, parseArguments=None):
 		self.plugin = plugin
 		self.func = func
 		self.arguments = arguments
@@ -188,20 +219,20 @@ class ServiceCall(BaseObject):
 class ServiceDoesNotExists(Exception):
 	__slots__ = ['plugin', 'func']
 
-	def init(self, plugin=None, func=None):
+	def __init__(self, plugin=None, func=None):
 		self.plugin = plugin
 		self.func = func
 
 class ServiceException(Exception):
 	__slots__ = ['msg']
 
-	def init(self, msg=None):
+	def __init__(self, msg=None):
 		self.msg = msg
 
 class UserData(BaseObject):
 	__slots__ = ['name', 'email', 'role', 'permission', 'templateName']
 
-	def init(self, name=None, email=None, role=None, permission=None, templateName=None):
+	def __init__(self, name=None, email=None, role=None, permission=None, templateName=None):
 		self.name = name
 		self.email = email
 		self.role = role
diff --git a/pavement.py b/pavement.py
index c534b4877..852179e94 100644
--- a/pavement.py
+++ b/pavement.py
@@ -11,6 +11,7 @@ from subprocess import call, Popen, PIPE
 from zipfile import ZipFile
 
 PROJECT_DIR = path(__file__).dirname()
+sys.path.append(PROJECT_DIR)
 
 options = environment.options
 path('pyload').mkdir()
@@ -172,7 +173,6 @@ def thrift(options):
     (outdir / "thriftgen").rmtree()
     (outdir / "gen-py").move(outdir / "thriftgen")
 
-
     #create light ttypes
     from module.remote.socketbackend.create_ttypes import main
     main()
diff --git a/pyLoadCore.py b/pyLoadCore.py
index 97c9ec64b..cbc270036 100755
--- a/pyLoadCore.py
+++ b/pyLoadCore.py
@@ -374,9 +374,9 @@ class Core(object):
         self.lastClientConnected = 0
 
         # later imported because they would trigger api import, and remote value not set correctly
+        from module import Api
         from module.HookManager import HookManager
         from module.ThreadManager import ThreadManager
-        from module import Api
 
         if Api.activated != self.remote:
             self.log.warning("Import error: API remote status not correct.")