Important Bugfix. v1.5.1

Block ID cache was inverting block ids with file offsets

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>org.warp</groupId>
     <artifactId>jcwdb</artifactId>
-    <version>1.5.0</version>
+    <version>1.5.1</version>
 
     <name>jcwdb</name>
     <url>https://git.ignuranza.net/andreacavalli/JCWDB</url>

src/main/java/org/warp/cowdb/database/DatabaseBlocksIO.java

@@ -21,6 +21,9 @@ public class DatabaseBlocksIO implements IBlocksIO {
 
 	@Override
 	public long newBlock(int size, ByteBuffer data) throws IOException {
+		if (size == 0) {
+			return EMPTY_BLOCK_ID;
+		}
 		long index = fileIO.writeAtEnd(size, data);
 		return blocksMetadata.newBlock(index, size);
 	}

src/main/java/org/warp/cowdb/database/DatabaseBlocksMetadata.java

@@ -47,12 +47,15 @@ public class DatabaseBlocksMetadata implements IBlocksMetadata {
 		long index = buffer.getLong();
 		int size = buffer.getInt();
 		blockInfo = new BlockInfo(index, size);
-		cache.put(index, blockInfo);
+		cache.put(blockId, blockInfo);
 		return blockInfo;
 	}
 
 	@Override
 	public long newBlock(long index, int size) throws IOException {
+		if (size == 0) {
+			return EMPTY_BLOCK_ID;
+		}
 		long newBlockId = firstFreeBlock++;
 		BlockInfo blockInfo = new BlockInfo(index, size);
 		cache.put(newBlockId, blockInfo);

src/main/java/org/warp/cowdb/database/DatabaseBlocksMetadataCache.java

@@ -37,25 +37,16 @@ public class DatabaseBlocksMetadataCache {
 		}
 	}
 
-	public void put(long block, BlockInfo value) throws IOException {
+	public void put(long block, BlockInfo blockInfo) throws IOException {
 		if (closed) return;
 		synchronized (readAccessLock) {
 			synchronized (writeAccessLock) {
-				blocks2Info.put(block, value);
+				blocks2Info.put(block, blockInfo);
 			}
 		}
 		this.flush();
 	}
 
-	public void remove(long block) {
-		if (closed) return;
-		synchronized (readAccessLock) {
-			synchronized (writeAccessLock) {
-				blocks2Info.remove(block);
-			}
-		}
-	}
-
 	private void flush() throws IOException {
 		if (closed) return;
 		int blocks2InfoSize = blocks2Info.size();
@@ -68,17 +59,12 @@ public class DatabaseBlocksMetadataCache {
 					BlockInfo blockInfo = entry.getValue();
 					entriesToFlush.add(flusher.flush(entry.getLongKey(), blockInfo.getIndex(), blockInfo.getSize()));
 					entriesIterator.remove();
+					if (entriesToFlush.size() >= 1000) {
+						executeAsyncFlush(entriesToFlush);
+					}
 					blocks2InfoSize--;
 				}
-				try {
+				executeAsyncFlush(entriesToFlush);
-					for (Future<?> entryToFlush : entriesToFlush) {
-						entryToFlush.get();
-					}
-				} catch (InterruptedException e) {
-					throw new IOException(e);
-				} catch (ExecutionException e) {
-					throw new IOException(e.getCause());
-				}
 			}
 		}
 	}
@@ -95,8 +81,19 @@ public class DatabaseBlocksMetadataCache {
 						Long2ObjectMap.Entry<BlockInfo> entry = entriesIterator.next();
 						BlockInfo blockInfo = entry.getValue();
 						entriesToFlush.add(flusher.flush(entry.getLongKey(), blockInfo.getIndex(), blockInfo.getSize()));
+						entriesIterator.remove();
+						if (entriesToFlush.size() >= 1000) {
+							executeAsyncFlush(entriesToFlush);
+						}
 						blocks2InfoSize--;
 					}
+					executeAsyncFlush(entriesToFlush);
+				}
+			}
+		}
+	}
+
+	private void executeAsyncFlush(List<Future<?>> entriesToFlush) throws IOException {
 		try {
 			for (Future<?> entryToFlush : entriesToFlush) {
 				entryToFlush.get();
@@ -105,9 +102,8 @@ public class DatabaseBlocksMetadataCache {
 			throw new IOException(e);
 		} catch (ExecutionException e) {
 			throw new IOException(e.getCause());
-					}
+		} finally {
-				}
+			entriesToFlush.clear();
-			}
 		}
 	}
 }

src/main/java/org/warp/cowdb/database/DatabaseReferencesMetadataCache.java

@@ -27,32 +27,25 @@ public class DatabaseReferencesMetadataCache {
 	}
 
 	public long get(long reference) throws IOException {
-		if (closed) throw new IOException("Cache already closed!");
 		synchronized (readAccessLock) {
+			if (closed) throw new IOException("Cache already closed!");
 			return references2Blocks.getOrDefault(reference, ERROR_BLOCK_ID);
 		}
 	}
 
-	public void put(long reference, long value) throws IOException {
+	public void put(long reference, long blockId) throws IOException {
-		if (closed) return;
 		synchronized (readAccessLock) {
 			synchronized (writeAccessLock) {
-				references2Blocks.put(reference, value);
+				if (closed) return;
+				references2Blocks.put(reference, blockId);
 			}
 		}
 		this.flush();
 	}
 
-	public void remove(long reference) {
+	private void flush() throws IOException {
-		if (closed) return;
 		synchronized (readAccessLock) {
 			synchronized (writeAccessLock) {
-				references2Blocks.remove(reference);
-			}
-		}
-	}
-
-	private void flush() throws IOException {
 				if (closed) return;
 				int references2BlocksSize = references2Blocks.size();
 				if (references2BlocksSize > MAX_CACHE_SIZE) {
@@ -63,16 +56,13 @@ public class DatabaseReferencesMetadataCache {
 							Long2LongMap.Entry entry = entriesIterator.next();
 							entriesToFlush.add(flusher.flush(entry.getLongKey(), entry.getLongValue()));
 							entriesIterator.remove();
+							if (entriesToFlush.size() >= 1000) {
+								executeAsyncFlush(entriesToFlush);
+							}
 							references2BlocksSize--;
 						}
-				try {
+						executeAsyncFlush(entriesToFlush);
-					for (Future<?> entryToFlush : entriesToFlush) {
-						entryToFlush.get();
 					}
-				} catch (InterruptedException e) {
-					throw new IOException(e);
-				} catch (ExecutionException e) {
-					throw new IOException(e.getCause());
 				}
 			}
 		}
@@ -89,8 +79,21 @@ public class DatabaseReferencesMetadataCache {
 					while (references2BlocksSize > 0) {
 						Long2LongMap.Entry entry = entriesIterator.next();
 						entriesToFlush.add(flusher.flush(entry.getLongKey(), entry.getLongValue()));
+						entriesIterator.remove();
+						if (entriesToFlush.size() >= 1000) {
+							executeAsyncFlush(entriesToFlush);
+						}
 						references2BlocksSize--;
 					}
+					executeAsyncFlush(entriesToFlush);
+				}
+			}
+		}
+	}
+
+	private void executeAsyncFlush(List<Future<?>> entriesToFlush) throws IOException {
+		synchronized (readAccessLock) {
+			synchronized (writeAccessLock) {
 				try {
 					for (Future<?> entryToFlush : entriesToFlush) {
 						entryToFlush.get();
@@ -99,7 +102,8 @@ public class DatabaseReferencesMetadataCache {
 					throw new IOException(e);
 				} catch (ExecutionException e) {
 					throw new IOException(e.getCause());
-					}
+				} finally {
+					entriesToFlush.clear();
 				}
 			}
 		}