Default queue_size to the concurrency n when not configured

With a fixed default of 10, running with a higher --n left most of the
thread pool idle (effective parallelism is min(n, queue_size)). An unset
queue_size now resolves to n at processing time, so the pool is never
starved by the default. An explicit config or constructor value still
takes precedence.

Author: lfoppiano

Readme.md

@@ -403,7 +403,7 @@ settings.
 | Parameter       | Description                                                                                                      | Default                 |
 |-----------------|------------------------------------------------------------------------------------------------------------------|-------------------------|
 | `grobid_server` | GROBID server URL                                                                                                | `http://localhost:8070` |
-| `queue_size`    | Number of files queued per processing chunk. **Tune carefully: a large queue size will result in the data being written less frequently**   | 1000                    |
+| `queue_size`    | Number of files queued per processing chunk. If not set, it follows the concurrency `n` so the thread pool is never starved. **Tune carefully: a large queue size will result in the data being written less frequently, and on the archive/S3 paths a whole chunk is held in memory at once**   | same as `n`             |
 | `sleep_time`    | Wait time when server is busy (seconds)                                                                          | 5                       |
 | `timeout`       | Client-side timeout (seconds)                                                                                    | 180                     |
 | `coordinates`   | XML elements for coordinate extraction                                                                           | See above               |

grobid_client/grobid_client.py

@@ -91,7 +91,9 @@ class GrobidClient(ApiClient):
     # Default configuration values
     DEFAULT_CONFIG: dict = {
         'grobid_server': 'http://localhost:8070',
-        'queue_size': 10,
+        # None means "follow the concurrency n at processing time", so the
+        # default queue never starves the thread pool
+        'queue_size': None,
         'sleep_time': 5,
         'timeout': 180,
         'coordinates': [
@@ -161,6 +163,14 @@ def _set_config_params(self, params: dict) -> None:
             if value is not None:
                 self.config[key] = value
 
+    def _effective_queue_size(self, n: int) -> int:
+        """Return the configured queue_size, defaulting to the concurrency n.
+
+        A queue smaller than the thread pool leaves workers idle, so when no
+        explicit value is configured the chunk size follows n.
+        """
+        return self.config.get("queue_size") or n
+
     def _warn_on_consolidation_timeout(self, consolidate_citations: bool) -> None:
         """Warn when citation consolidation is enabled with a low client timeout.
 
@@ -859,7 +869,7 @@ def _run_file_batches(
 
         Returns the aggregated (processed, errors, skipped) counts.
         """
-        queue_size = self.config["queue_size"]
+        queue_size = self._effective_queue_size(n)
         processed_files_count = 0
         errors_files_count = 0
         skipped_files_count = 0
@@ -1117,7 +1127,7 @@ def _process_archive_core(
         Does not print the final summary (the caller does), so it can be
         aggregated with other inputs when resolving a glob pattern.
         """
-        queue_size = self.config["queue_size"]
+        queue_size = self._effective_queue_size(n)
 
         # Results must survive the temporary extraction directories, so when no
         # output is given we default to a directory named after the archive. For
@@ -1278,7 +1288,7 @@ def _process_remote_files(
         if output is None:
             output = "."
 
-        queue_size = self.config["queue_size"]
+        queue_size = self._effective_queue_size(n)
         print(f"Found {total} remote file(s) to process")
         processed_count = 0
         error_count = 0

tests/test_grobid_client.py

@@ -43,12 +43,24 @@ def test_init_default_values(self, mock_configure_logging, mock_test_server):
         client = GrobidClient(check_server=False)
 
         assert client.config['grobid_server'] == 'http://localhost:8070'
-        assert client.config['queue_size'] == 10
+        assert client.config['queue_size'] is None
         assert client.config['sleep_time'] == 5
         assert client.config['timeout'] == 180
         assert 'persName' in client.config['coordinates']
         mock_configure_logging.assert_called_once()
 
+    @patch('grobid_client.grobid_client.GrobidClient._test_server_connection')
+    @patch('grobid_client.grobid_client.GrobidClient._configure_logging')
+    def test_effective_queue_size(self, mock_configure_logging, mock_test_server):
+        """Test that an unset queue_size defaults to the concurrency n."""
+        mock_test_server.return_value = (True, 200)
+
+        client = GrobidClient(check_server=False)
+        assert client._effective_queue_size(40) == 40
+
+        client.config['queue_size'] = 100
+        assert client._effective_queue_size(40) == 100
+
     @patch('grobid_client.grobid_client.GrobidClient._test_server_connection')
     @patch('grobid_client.grobid_client.GrobidClient._configure_logging')
     def test_init_custom_values(self, mock_configure_logging, mock_test_server):