feat(#41): tackled thread pool starving situations

Automatic re-scheduling and transaction pipeline recreation.
FgForrest · Jun 23, 2024 · e27b0c1 · e27b0c1
1 parent 494cd15
commit e27b0c1
Show file tree

Hide file tree

Showing 7 changed files with 27 additions and 32 deletions.
diff --git a/documentation/user/en/operate/configure.md b/documentation/user/en/operate/configure.md
@@ -39,7 +39,6 @@ transaction:                                      # [see Transaction configurati
   transactionMemoryRegionCount: 256
   walFileSizeBytes: 16MB
   walFileCountKept: 8
-  maxQueueSize: 1K
   flushFrequencyInMillis: 1s
 
 cache:                                            # [see Cache configuration](#cache-configuration)
@@ -489,12 +488,6 @@ This section contains configuration options for the storage layer of the databas
         <p>Number of WAL files to keep. Increase this number in combination with `walFileSizeBytes` if you want to
             keep longer history of changes.</p>
     </dd>
-    <dt>maxQueueSize</dt>
-    <dd>
-        <p>**Default:** `1K`</p>
-        <p>Size of the catalog queue for parallel transaction. If there are more transaction than the number of free 
-            threads in the pool, the transaction are queued. If the queue is full, the transaction is rejected.</p>
-    </dd>
     <dt>flushFrequencyInMillis</dt>
     <dd>
         <p>**Default:** `1s`</p>

diff --git a/documentation/user/en/operate/reference/metrics.md b/documentation/user/en/operate/reference/metrics.md
@@ -3,12 +3,6 @@
 <UsedTerms>
   <h4>Labels used in metrics</h4>
   <dl>
-    <dt>taskName</dt>
-    <dd><strong>N/A</strong>: N/A</dd>
-    <dt>fileType</dt>
-    <dd><strong>File type</strong>: N/A</dd>
-    <dt>name</dt>
-    <dd><strong>Logical file name</strong>: N/A</dd>
     <dt>fileType</dt>
     <dd><strong>File type</strong>: N/A</dd>
     <dt>name</dt>
@@ -99,6 +93,12 @@
     <dd><strong>Logical file name</strong>: N/A</dd>
     <dt>recordType</dt>
     <dd><strong>Record type</strong>: N/A</dd>
+    <dt>taskName</dt>
+    <dd><strong>N/A</strong>: N/A</dd>
+    <dt>fileType</dt>
+    <dd><strong>File type</strong>: N/A</dd>
+    <dt>name</dt>
+    <dd><strong>Logical file name</strong>: N/A</dd>
   </dl>
 </UsedTerms>
 

diff --git a/evita_engine/src/main/java/io/evitadb/core/transaction/TransactionManager.java b/evita_engine/src/main/java/io/evitadb/core/transaction/TransactionManager.java
@@ -470,17 +470,19 @@ public void updateLastFinalizedCatalog(@Nonnull Catalog lastFinalizedCatalog, lo
 	 */
 	public void notifyCatalogPresentInLiveView(@Nonnull Catalog livingCatalog) {
 		final Catalog theLivingCatalog = getLivingCatalog();
-		Assert.isPremiseValid(
-			theLivingCatalog.getVersion() < livingCatalog.getVersion(),
-			"Catalog versions must be in order! " +
-				"Expected " + theLivingCatalog.getVersion() + ", got " + livingCatalog.getVersion() + "."
-		);
-		final long theLastFinalizedVersion = getLastFinalizedCatalogVersion();
-		Assert.isPremiseValid(
-			theLastFinalizedVersion >= livingCatalog.getVersion(),
-			"Catalog versions must be in order! " +
-				"Expected " + theLastFinalizedVersion + ", got " + livingCatalog.getVersion() + "."
-		);
+		if (livingCatalog.getVersion() > 0L) {
+			Assert.isPremiseValid(
+				theLivingCatalog.getVersion() < livingCatalog.getVersion(),
+				"Catalog versions must be in order! " +
+					"Expected " + theLivingCatalog.getVersion() + ", got " + livingCatalog.getVersion() + "."
+			);
+			final long theLastFinalizedVersion = getLastFinalizedCatalogVersion();
+			Assert.isPremiseValid(
+				theLastFinalizedVersion >= livingCatalog.getVersion(),
+				"Catalog versions must be in order! " +
+					"Expected " + theLastFinalizedVersion + ", got " + livingCatalog.getVersion() + "."
+			);
+		}
 		this.lastAssignedCatalogVersion.updateAndGet(current -> Math.max(current, livingCatalog.getVersion()));
 		this.livingCatalog.set(livingCatalog);
 

diff --git a/...ctional_tests/src/test/java/io/evitadb/store/wal/CatalogWriteAheadLogIntegrationTest.java b/...ctional_tests/src/test/java/io/evitadb/store/wal/CatalogWriteAheadLogIntegrationTest.java
@@ -251,10 +251,12 @@ void shouldCorrectlyReportFirstAvailableTimestamp() {
 
 		final OffsetDateTime initialTimestamp = OffsetDateTime.now();
 		writeWal(bigOffHeapMemoryManager, transactionSizes, initialTimestamp);
+		this.wal.walProcessedUntil(Long.MAX_VALUE);
+		this.wal.removeWalFiles();
 
-		assertEquals(2, offsetConsumer.getCatalogVersions().size());
-		assertEquals(initialTimestamp.plusMinutes(1), offsetConsumer.getCatalogVersions().get(0));
-		assertEquals(initialTimestamp.plusMinutes(2), offsetConsumer.getCatalogVersions().get(1));
+		// only one call would occur with the latest version possible
+		assertEquals(1, offsetConsumer.getCatalogVersions().size());
+		assertEquals(3, offsetConsumer.getCatalogVersions().get(0));
 	}
 
 	@Nonnull
@@ -271,7 +273,7 @@ private CatalogWriteAheadLog createCatalogWriteAheadLogOfSmallSize() {
 				.build(),
 			Mockito.mock(Scheduler.class),
 			offsetConsumer,
-			null
+			firstActiveCatalogVersion -> {}
 		);
 	}
 
@@ -286,7 +288,7 @@ private CatalogWriteAheadLog createCatalogWriteAheadLogOfLargeEnoughSize() {
 			TransactionOptions.builder().walFileSizeBytes(Long.MAX_VALUE).build(),
 			Mockito.mock(Scheduler.class),
 			offsetConsumer,
-			null
+			firstActiveCatalogVersion -> {}
 		);
 	}
 

diff --git a/evita_server/src/main/resources/evita-configuration.yaml b/evita_server/src/main/resources/evita-configuration.yaml
@@ -38,7 +38,6 @@ transaction:
   transactionMemoryRegionCount: ${transaction.memoryRegionCount:256}
   walFileSizeBytes: ${transaction.walFilSize:16M}
   walFileCountKept: ${transaction.walFileCountKept:8}
-  maxQueueSize: ${transaction.maxQueueSize:1024}
   flushFrequencyInMillis: ${transaction.flushFrequencyInMillis:1000}
 
 cache:

diff --git a/evita_store/evita_store_server/src/main/java/io/evitadb/store/wal/CatalogWriteAheadLog.java b/evita_store/evita_store_server/src/main/java/io/evitadb/store/wal/CatalogWriteAheadLog.java
@@ -1036,7 +1036,7 @@ MutationSupplier createSupplier(long startCatalogVersion, @Nullable Long request
 	/**
 	 * Removes the obsolete WAL files from the catalog storage path.
 	 */
-	private long removeWalFiles() {
+	long removeWalFiles() {
 		synchronized (this.pendingRemovals) {
 			final long catalogVersion = this.processedCatalogVersion.get();
 			final Set<PendingRemoval> toRemove = new HashSet<>(64);

diff --git a/evita_test_support/src/main/resources/evita-configuration.yaml b/evita_test_support/src/main/resources/evita-configuration.yaml
@@ -38,7 +38,6 @@ transaction:
   transactionMemoryRegionCount: ${transaction.memoryRegionCount:256}
   walFileSizeBytes: ${transaction.walFilSize:16M}
   walFileCountKept: ${transaction.walFileCountKept:8}
-  maxQueueSize: ${transaction.maxQueueSize:1024}
   flushFrequencyInMillis: ${transaction.flushFrequencyInMillis:1000}
 
 cache: