support multiple threads for polldb

Author: toluaina

pgsync/__init__.py

@@ -4,4 +4,4 @@
 
 __author__ = "Tolu Aina"
 __email__ = "[email protected]"
-__version__ = "2.2.0"
+__version__ = "2.2.1"

pgsync/elastichelper.py

@@ -150,6 +150,10 @@ def _bulk(
         )
         raise_on_error: bool = raise_on_error or ELASTICSEARCH_RAISE_ON_ERROR
 
+        # when using multiple threads for poll_db we need to account for other
+        # threads performing deletions
+        ignore_status: Tuple[int] = (400, 404)
+
         if ELASTICSEARCH_STREAMING_BULK:
             for _ in helpers.streaming_bulk(
                 self.__es,
@@ -178,6 +182,7 @@ def _bulk(
                 refresh=refresh,
                 raise_on_exception=raise_on_exception,
                 raise_on_error=raise_on_error,
+                ignore_status=ignore_status,
             ):
                 self.doc_count += 1
 

pgsync/redisqueue.py

@@ -28,7 +28,6 @@ def __init__(self, name: str, namespace: str = "queue", **kwargs):
         except ConnectionError as e:
             logger.exception(f"Redis server is not running: {e}")
             raise
-        self._pipeline = self.__db.pipeline()
 
     @property
     def qsize(self) -> int:
@@ -38,9 +37,10 @@ def qsize(self) -> int:
     def bulk_pop(self, chunk_size: Optional[int] = None) -> List[dict]:
         """Remove and return multiple items from the queue."""
         chunk_size: int = chunk_size or REDIS_READ_CHUNK_SIZE
-        self._pipeline.lrange(self.key, 0, chunk_size - 1)
-        self._pipeline.ltrim(self.key, chunk_size, -1)
-        items: List[List[bytes], bool] = self._pipeline.execute()
+        pipeline = self.__db.pipeline()
+        pipeline.lrange(self.key, 0, chunk_size - 1)
+        pipeline.ltrim(self.key, chunk_size, -1)
+        items: List[List[bytes], bool] = pipeline.execute()
         logger.debug(f"bulk_pop nsize: {len(items[0])}")
         return list(map(lambda value: json.loads(value), items[0]))
 

pgsync/settings.py

@@ -29,6 +29,8 @@
 # page block size
 BLOCK_SIZE = env.int("BLOCK_SIZE", default=2048 * 10)
 QUERY_LITERAL_BINDS = env.bool("QUERY_LITERAL_BINDS", default=None)
+# number of threads to spawn for poll db
+NTHREADS_POLLDB = env.int("NTHREADS_POLLDB", default=1)
 
 # Elasticsearch:
 ELASTICSEARCH_SCHEME = env.str("ELASTICSEARCH_SCHEME", default="http")

pgsync/sync.py

@@ -44,6 +44,7 @@
 from .settings import (
     CHECKPOINT_PATH,
     LOG_INTERVAL,
+    NTHREADS_POLLDB,
     POLL_TIMEOUT,
     REDIS_POLL_INTERVAL,
     REDIS_WRITE_CHUNK_SIZE,
@@ -1107,7 +1108,7 @@ def status(self):
             sys.stdout.flush()
             time.sleep(LOG_INTERVAL)
 
-    def receive(self) -> None:
+    def receive(self, nthreads_polldb=None) -> None:
         """
         Receive events from db.
 
@@ -1119,7 +1120,9 @@ def receive(self) -> None:
         """
         # start a background worker producer thread to poll the db and populate
         # the Redis cache
-        self.poll_db()
+        nthreads_polldb = nthreads_polldb or NTHREADS_POLLDB
+        for _ in range(nthreads_polldb):
+            self.poll_db()
 
         # sync up to current transaction_id
         self.pull()
@@ -1187,6 +1190,13 @@ def receive(self) -> None:
     default=False,
     help="Analyse database",
 )
+@click.option(
+    "--nthreads_polldb",
+    "-n",
+    help="Number of threads to spawn for poll db",
+    type=int,
+    default=NTHREADS_POLLDB,
+)
 def main(
     config,
     daemon,
@@ -1199,6 +1209,7 @@ def main(
     verbose,
     version,
     analyze,
+    nthreads_polldb,
 ):
     """Main application syncer."""
     if version:
@@ -1237,7 +1248,7 @@ def main(
             sync: Sync = Sync(document, verbose=verbose, **kwargs)
             sync.pull()
             if daemon:
-                sync.receive()
+                sync.receive(nthreads_polldb)
 
 
 if __name__ == "__main__":

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.1.11
+current_version = 2.2.0
 commit = True
 tag = True
 

tests/test_redisqueue.py

@@ -23,7 +23,6 @@ def test_redis_conn(self, mock_logger, mocker):
         mock_get_redis_url.assert_called_once()
         mock_ping.assert_called_once()
         mock_logger.exception.assert_not_called()
-        assert queue._pipeline is not None
 
     @patch("pgsync.redisqueue.logger")
     def test_redis_conn_fail(self, mock_logger, mocker):