Changeset b6021d in indico

Timestamp:

01/26/11 18:37:38 (2 years ago)

Author:

Pedro Ferreira <jose.pedro.ferreira@…>

Branches:

master, burotel, hello-world-walkthrough, ipv6, v0.98-series, v0.98.2, v0.98.3, v0.98b1, v0.98b2, v0.99, 051b2622c51afb171a1dedb46a0df4fbb0cbd02e, d9941f8582b36b24821a11ea5ba16fda6a457fb1

Children:

756437

Parents:

1df37b

git-author:

Pedro Ferreira <jose.pedro.ferreira@…> (01/18/11 17:48:28)

git-committer:

Pedro Ferreira <jose.pedro.ferreira@…> (01/26/11 18:37:38)

Message:

[IMP] Back to single threaded :(

Location:

indico/ext/livesync

Files:

• agent.py (modified) (2 diffs)
• invenio/agent.py (modified) (5 diffs)

indico/ext/livesync/agent.py

@@ -342 +342 @@
 
 
-class UploaderSlave(Thread):
-    """
-    A generic threaded "work slave" for agents
-    """
-
-    def __init__(self, name, queue, logger, agent):
-        self._keepRunning = True
+class RecordUploader(object):
+
+    DEFAULT_BATCH_SIZE, DEFAULT_NUM_SLAVES = 1000, 2
+
+    def __init__(self, logger, agent, batchSize=DEFAULT_BATCH_SIZE):
         self._logger = logger
-        self._terminate = False
-        self.result = True
-        self._queue = queue
-        self._name = name
-        self._uploaded = 0
         self._agent = agent
-
-        super(UploaderSlave, self).__init__(name=name)
-
-    def run(self):
-
-        DBMgr.getInstance().startRequest()
-
-        self._logger.debug('Worker [%s] started' % self._name)
-        try:
-            while True:
-                taskFetched = False
-                try:
-                    batch = self._queue.get(True, 2)
-                    taskFetched = True
-                    self.result &= self._uploadBatch(batch)
-                    self._uploaded += len(batch)
-                except Empty:
-                    pass
-                finally:
-                    if taskFetched:
-                        self._queue.task_done()
-
-                if self._terminate:
-                    break
-        except:
-            self._logger.exception('Worker [%s]:' % self._name)
-            return 1
-        finally:
-            DBMgr.getInstance().endRequest()
-            self._logger.debug('Worker [%s] finished' % self._name)
-        self._logger.debug('Worker [%s] returning %s' % (self._name, self.result))
-
-    def terminate(self):
-        self._terminate = True
+        self._batchSize = batchSize
 
     def _uploadBatch(self, batch):
         """
-        Sends a batch of records
-        Overloaded by agent.
-        """
-        raise Exception("Unimplemented method")
-
-    def _getMetadata(self, records):
-        """
-        Generates the metadata for a batch of records.
-        Should be overloaded.
-        """
-        raise Exception("Unimplemented method")
-
+        To be overloaded by uploaders
+        """
+        raise Exception("Unimplemented method!")
+
+    def iterateOver(self, iterator):
+        """
+        Consumes an iterator
+        """
+
+        currentBatch = []
+
+        # take operations and choose which records to send
+        for record in iterator:
+
+            if len(currentBatch) > (self._batchSize - 1):
+                self._uploadBatch(currentBatch)
+                currentBatch = []
+
+            currentBatch.append(record)
+
+        if currentBatch:
+            self._uploadBatch(currentBatch)
+
+        return True
+
+###################
+# This code is now unused.
+#
+#
 
 class ThreadedRecordUploader(object):
@@ -529 +506 @@
         stream.write("%d uploaded (total)\n\n" % totalUp)
         stream.flush()
+
+
+class UploaderSlave(Thread):
+    """
+    A generic threaded "work slave" for agents
+    """
+
+    def __init__(self, name, queue, logger, agent):
+        self._keepRunning = True
+        self._logger = logger
+        self._terminate = False
+        self.result = True
+        self._queue = queue
+        self._name = name
+        self._uploaded = 0
+        self._agent = agent
+        self._startTime = time.time()
+        self._currentBatch = []
+        self._dead = False
+
+        super(UploaderSlave, self).__init__(name=name)
+
+    def run(self):
+
+        DBMgr.getInstance().startRequest()
+
+        self._logger.debug('Worker [%s] started' % self._name)
+        try:
+            while True:
+                taskFetched = False
+                try:
+                    self._currentBatch = self._queue.get(True, 2)
+                    taskFetched = True
+                    self._startTime = time.time()
+                    self.result &= self._uploadBatch(self._currentBatch)
+                    self._uploaded += len(self._currentBatch)
+                except Empty:
+                    pass
+                finally:
+                    if taskFetched:
+                        self._queue.task_done()
+
+                if self._terminate:
+                    break
+        except:
+            self._logger.exception('Worker [%s]:' % self._name)
+            return 1
+        finally:
+            DBMgr.getInstance().endRequest()
+            self._logger.debug('Worker [%s] finished' % self._name)
+        self._logger.debug('Worker [%s] returning %s' % (self._name, self.result))
+
+    def terminate(self):
+        self._terminate = True
+
+    def _uploadBatch(self, batch):
+        """
+        Sends a batch of records
+        Overloaded by agent.
+        """
+        raise Exception("Unimplemented method")
+
+    def _getMetadata(self, records):
+        """
+        Generates the metadata for a batch of records.
+        Should be overloaded.
+        """
+        raise Exception("Unimplemented method")
+

indico/ext/livesync/invenio/agent.py

@@ -24 +24 @@
 # plugin imports
 from indico.ext.livesync.agent import PushSyncAgent, AgentProviderComponent, \
-     UploaderSlave, ThreadedRecordUploader
+     RecordUploader
 from indico.ext.livesync.invenio.invenio_connector import InvenioConnector
 
@@ -38 +38 @@
 
 
-class InvenioUploaderSlave(UploaderSlave):
-    """
-    A worker that uploads data using HTTP
-    """
-
-    def __init__(self, name, queue, logger, agent, server):
-        super(InvenioUploaderSlave, self).__init__(name, queue, logger, agent)
-        self._server = server
-
-    def _uploadBatch(self, batch):
-        """
-        Uploads a batch to the Invenio server
-        """
-
-        self._logger.debug('getting a batch')
-
-        tstart = time.time()
-        # get a batch
-        data = self._agent._getMetadata(batch)
-
-        tgen = time.time() - tstart
-
-        result = self._server.upload_marcxml(data, "-ir").read()
-
-        tupload = time.time() - (tstart + tgen)
-
-        self._logger.debug('rec %s result: %s' % (batch, result))
-
-        if result.startswith('[INFO]'):
-            fpath = result.strip().split(' ')[-1]
-            self._logger.info('Batch of %d records stored in server (%s) '
-                              '[%f s %f s]- %d batches left' % \
-                              (len(batch), fpath, tgen, tupload, self._queue.qsize()))
-        else:
-            self._logger.error('Records: %s output: %s' % (batch, result))
-            raise Exception('upload failed')
-
-        return True
-
+# Attention: if this class is not declared, the LiveSync management interface
+# will never know this plugin exists!
 
 class InvenioRecordProcessor(object):
@@ -168 +131 @@
     """
 
-    _workerClass = InvenioUploaderSlave
     _creationState = STATUS_CREATED
     _extraOptions = {'url': 'Server URL'}
@@ -206 +168 @@
 
         # the uploader will manage everything for us...
-        uploader = ThreadedRecordUploader(InvenioUploaderSlave,
-                                          self._v_logger,
-                                          extraSlaveArgs=(server,),
-                                          monitor=monitor)
-        uploader.spawn()
+
+        uploader = InvenioRecordUploader(logger, self, server)
 
         if self._v_logger:
@@ -216 +175 @@
 
         # iterate over the returned records and upload them
-        uploader.iterateOver(records)
-
-        # wait for uploader to finish
-        result = uploader.join()
-
-        return result
-
-
-# Attention: if this class is not declared, the LiveSync management interface
-# will never know this plugin exists!
+        return uploader.iterateOver(records)
+
+
+class InvenioRecordUploader(RecordUploader):
+    """
+    A worker that uploads data using HTTP
+    """
+
+    def __init__(self, logger, agent, server):
+        super(InvenioRecordUploader, self).__init__(logger, agent)
+        self._server = server
+
+    def _uploadBatch(self, batch):
+        """
+        Uploads a batch to the Invenio server
+        """
+
+        self._logger.debug('getting a batch')
+
+        tstart = time.time()
+        # get a batch
+
+        self._logger.info('Generating metadata')
+        data = self._agent._getMetadata(batch)
+        self._logger.info('Metadata ready ')
+
+        tgen = time.time() - tstart
+
+        result = self._server.upload_marcxml(data, "-ir").read()
+
+        tupload = time.time() - (tstart + tgen)
+
+        self._logger.debug('rec %s result: %s' % (batch, result))
+
+        if result.startswith('[INFO]'):
+            fpath = result.strip().split(' ')[-1]
+            self._logger.info('Batch of %d records stored in server (%s) '
+                              '[%f s %f s]' % \
+                              (len(batch), fpath, tgen, tupload))
+        else:
+            self._logger.error('Records: %s output: %s' % (batch, result))
+            raise Exception('upload failed')
+
+        return True
+
 
 class InvenioAgentProviderComponent(AgentProviderComponent):