support job batching (#12)

* support job batching

* worker: pull more frequently

3 files changed, 109 insertions, 44 deletions

diff --git a/joblib.py b/joblib.py
index 7ecde6a..9e8cf84 100644
--- a/joblib.py
+++ b/joblib.py
@@ -43,11 +43,21 @@ class JobQueue(object):
     async def request(self, jobtype, payload, priority=0):
         """Create a new job and return its id."""
         async with self._pool.acquire() as con:
-            return await con.fetchval("""
-                INSERT INTO jobs(type, payload, priority)
-                VALUES($1, $2, $3)
-                RETURNING id
-            """, jobtype, json.dumps(payload), priority)
+            if isinstance(payload, list):
+                payloads = payload
+            else:
+                payloads = [payload]
+            ids = []
+            for pl in payloads:
+                ids.append(await con.fetchval("""
+                    INSERT INTO jobs(type, payload, priority)
+                    VALUES($1, $2, $3)
+                    RETURNING id
+                """, jobtype, json.dumps(pl), priority))
+            if isinstance(payload, list):
+                return ids
+            else:
+                return ids[0]
 
     async def acquire(self, jobtype):
         """Mark a job as running, return id, payload and priority.
@@ -76,7 +86,7 @@ class JobQueue(object):
                 except asyncpg.exceptions.SerializationError:
                     # job is being picked up by another worker, try again
                     pass
-    
+
     async def status(self, jobid):
         """Return the status of a job."""
         async with self._pool.acquire() as con:
@@ -87,34 +97,51 @@ class JobQueue(object):
             """, jobid)
 
     async def finish(self, jobid):
-        """Mark a job as completed."""
+        """Mark jobs as completed."""
         async with self._pool.acquire() as con:
-            while True:
-                try:
-                    async with con.transaction(isolation="serializable"):
-                        await con.execute("""
-                            UPDATE jobs
-                            SET status='finished'
-                            WHERE id=$1
-                        """, jobid)
-                        return
-                except asyncpg.exceptions.SerializationError:
-                    pass
+            if not isinstance(jobid, list):
+                jobids = [jobid]
+            else:
+                jobids = jobid
+            for jid in jobids:
+                await con.execute("""
+                    UPDATE jobs
+                    SET status='finished'
+                    WHERE id=$1
+                """, jid)
 
     async def fail(self, jobid, reason):
         """Mark a job as failed."""
         async with self._pool.acquire() as con:
-            while True:
-                try:
-                    async with con.transaction(isolation="serializable"):
-                        await con.execute("""
-                            UPDATE jobs
-                            SET status='failed', payload=payload||$2::jsonb
-                            WHERE id=$1
-                        """, jobid, json.dumps({"error": reason}))
-                        return
-                except asyncpg.exceptions.SerializationError:
-                    pass
+            if not isinstance(jobid, list):
+                jobids = [jobid]
+            else:
+                jobids = jobid
+            if not isinstance(reason, list):
+                reasons = [reason]
+            else:
+                reasons = reason
+            assert len(jobids) == len(reasons)
+            for jid, rsn in zip(jobids, reasons):
+                await con.execute("""
+                    UPDATE jobs
+                    SET status='failed', payload=payload||$2::jsonb
+                    WHERE id=$1
+                """, jid, json.dumps({"error": rsn}))
+
+    async def reset(self, jobid):
+        """Mark a job as open."""
+        async with self._pool.acquire() as con:
+            if not isinstance(jobid, list):
+                jobids = [jobid]
+            else:
+                jobids = jobid
+            for jid in jobids:
+                await con.execute("""
+                    UPDATE jobs
+                    SET status='open'
+                    WHERE id=$1
+                """, jid)
 
     async def cleanup(self):
         """Reopen all unfinished jobs."""
diff --git a/test_joblib.py b/test_joblib.py
index 865cfab..1a760f8 100644
--- a/test_joblib.py
+++ b/test_joblib.py
@@ -49,6 +49,16 @@ class TestJoblib:
         assert None == (await queue.acquire(jobtype="testing"))[1]
 
     @pytest.mark.asyncio
+    async def test_request_and_acquire_batch(self, queue, payload):
+        # request should succeed
+        await queue.request(jobtype="testing", payload=[payload]*5)
+        # acquire should return same payload
+        for _ in range(5):
+            assert payload == (await queue.acquire(jobtype="testing"))[1]
+        # there should not be another job
+        assert None == (await queue.acquire(jobtype="testing"))[1]
+
+    @pytest.mark.asyncio
     async def test_cleanup(self, queue, payload):
         await queue.request(jobtype="testing", payload=payload)
         # mark job as processing
diff --git a/worker.py b/worker.py
index 2737671..163ebfd 100644
--- a/worker.py
+++ b/worker.py
@@ -24,33 +24,61 @@ class Worker(object):
         # override
         pass
 
+    async def _windup(self):
+        # override
+        pass
+
     async def _execute_job(self, jobid, payload, priority):
         # override
         pass
 
+    async def _teardown(self, failed):
+        # override
+        pass
+
     async def _work(self):
-        """Fetch a job and run it."""
+        """Fetch a job and run it.
+        Return id."""
         jobid, payload, priority = await self._queue.acquire(
             jobtype=self._jobtype)
         if jobid is None:
-            raise LookupError("no jobs available")
+            raise LookupError
         try:
             await self._execute_job(jobid, payload, priority)
-            await self._queue.finish(jobid)
-        except JobFailed as error:
-            logging.warning("%s: failed with %s", jobid,
-                            error.args[0])
-            await self._queue.fail(jobid, error.args[0])
+        except JobFailed as err:
+            raise JobFailed(err.args[0], jobid)
+        return jobid
 
-    async def run(self):
+    async def run(self, batchlimit=1):
         """Start jobs forever."""
         while True:
+            await self._windup()
+            jobids = []
+            error = None
+            low_load = False
             try:
-                await self._work()
-            except LookupError:
-                await asyncio.sleep(1)
+                for _ in range(batchlimit):
+                    try:
+                        jobids.append(await self._work())
+                    except LookupError:
+                        low_load = True
+                        break
+                    except JobFailed as err:
+                        error = err.args[0]
+                        jobids.append(err.args[1])
+                        break
+            finally:
+                if error is not None:
+                    await self._queue.reset(jobids[:-1])
+                    logging.debug(jobids)
+                    await self._queue.fail(jobids[-1], error)
+                    logging.warning("batch failed, reset")
+                else:
+                    await self._queue.finish(jobids)
+                await self._teardown(failed=error is not None)
+            if low_load:
+                await asyncio.sleep(0.1)
 
-    async def start(self, number=1):
-        """Start jobs in background."""
-        for _ in range(number):
-            asyncio.ensure_future(self.run())
+    async def start(self, batchlimit=1):
+        """Start in background."""
+        asyncio.ensure_future(self.run(batchlimit))