jody/yacy_search_server
1
0 Fork 0
mirror of https://github.com/yacy/yacy_search_server.git synced 2026-10-25 05:54:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Introduced immutable indexes to solve the problem of

Browse Source 
dying-peers-when-growing problem:

HeapReader can now open immutable indexes as SSTableHandleMap instances.
Keys and offsets remain file-backed instead of occupying the Java heap.
Lazy working copies avoid additional files until a mutation occurs.
Negative offsets record deletions without tombstone sidecars or RAM
maps. Mutable heaps retain RowHandleMap, while sealed segments save heap
memory.

With immutable indexes the memory consumption is not stopped, it is only
moved from heap memory to system memory. The result should be that there
is no need to assign more memory to YaCy when it gets bigger, it simply
takes it itself from the system memory.

This should also fix
https://github.com/yacy/yacy_search_server/issues/731
This commit is contained in:
Michael Peter Christen committed 2026-08-30 14:55:29 +02:00
1 parent a1be12f5b3
commit d05db9e6bb
22 files changed
+4456 -586
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

source/net/yacy/cora/storage/HandleMap.java
+6 -49
View File
@@ -20,38 +20,11 @@
package net.yacy.cora.storage;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.util.SpaceExceededException;
public interface HandleMap extends Iterable<Map.Entry<byte[], Long>> {
public long mem();
public void optimize();
/**
* write a dump of the index to a file. All entries are written in order
* which makes it possible to read them again in a fast way
* @param file
* @return the number of written entries
* @throws IOException
*/
public int dump(final File file) throws IOException;
public void clear();
public byte[] smallestKey();
public byte[] largestKey();
public boolean has(final byte[] key);
public long get(final byte[] key);
public interface HandleMap extends ImmutableHandleMap {
/**
* Adds the key-value pair to the index.
@@ -61,30 +34,14 @@ public interface HandleMap extends Iterable<Map.Entry<byte[], Long>> {
* @throws IOException
* @throws SpaceExceededException
*/
public long put(final byte[] key, final long l) throws SpaceExceededException;
public long put(final byte[] key, final long l) throws SpaceExceededException; // mutable operation();
public void putUnique(final byte[] key, final long l) throws SpaceExceededException;
public void putUnique(final byte[] key, final long l) throws SpaceExceededException; // mutable operation();
public long add(final byte[] key, final long a) throws SpaceExceededException;
public long add(final byte[] key, final long a) throws SpaceExceededException; // mutable operation();
public long inc(final byte[] key) throws SpaceExceededException;
public long inc(final byte[] key) throws SpaceExceededException; // mutable operation();
public long dec(final byte[] key) throws SpaceExceededException;
public ArrayList<long[]> removeDoubles() throws SpaceExceededException;
public ArrayList<byte[]> top(final int count);
public long remove(final byte[] key);
public long removeone();
public int size();
public boolean isEmpty();
public CloneableIterator<byte[]> keys(final boolean up, final byte[] firstKey);
public void close();
public long dec(final byte[] key) throws SpaceExceededException; // mutable operation();
}
source/net/yacy/cora/storage/ImmutableHandleMap.java
+72
View File
@@ -0,0 +1,72 @@
/**
* ImmutableHandleMap
* Copyright 2009 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
* First published 26.07.2012 on http://yacy.net
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program in the file lgpl21.txt
* If not, see <http://www.gnu.org/licenses/>.
*/
package net.yacy.cora.storage;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.util.SpaceExceededException;
public interface ImmutableHandleMap extends Iterable<Map.Entry<byte[], Long>> {
public long mem();
public void optimize();
/**
* write a dump of the index to a file. All entries are written in order
* which makes it possible to read them again in a fast way
* @param file
* @return the number of written entries
* @throws IOException
*/
public int dump(final File file) throws IOException;
public void clear();
public byte[] smallestKey();
public byte[] largestKey();
public boolean has(final byte[] key);
public long get(final byte[] key);
public ArrayList<long[]> removeDoubles() throws SpaceExceededException;
public ArrayList<byte[]> top(final int count);
public long remove(final byte[] key);
public long removeone();
public int size();
public boolean isEmpty();
public CloneableIterator<byte[]> keys(final boolean up, final byte[] firstKey);
public void close();
}
source/net/yacy/cora/util/ChunkedBytes.java
+317 -87
View File
@@ -25,8 +25,10 @@ import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.ReadOnlyBufferException;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.OpenOption;
@@ -34,6 +36,7 @@ import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import net.yacy.cora.document.encoding.UTF8;
@@ -49,10 +52,12 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
private final List<Segment> segments;
private long size;
private volatile boolean closed;
public ChunkedBytes() {
this.segments = new ArrayList<>();
this.size = 0;
this.closed = false;
}
public ChunkedBytes(InputStream in) throws IOException {
@@ -70,10 +75,13 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
this.append(UTF8.getBytes(initialData));
}
private ChunkedBytes(List<Segment> segments, long size) {
this.segments = new ArrayList<>(segments);
this.size = size;
}
private ChunkedBytes(final List<Segment> segments, final long size,
final List<FileBacking> fileBackings) {
this.segments = new ArrayList<>(segments);
this.size = size;
this.fileBackings.addAll(fileBackings);
this.closed = false;
}
/** Represents one contiguous region in the logical address space. */
private static final class Segment implements Closeable, Cloneable {
@@ -86,6 +94,7 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
@Override public Object clone() {
return new Segment((Chunk) this.chunk.clone(), this.start, this.length);
}
void force() { this.chunk.force(); }
@Override public void close() throws IOException { this.chunk.close(); }
}
@@ -97,7 +106,45 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
void set(long relPos, byte b);
int length();
Object clone();
@Override default void close() { /* no-op by default */ }
default void force() { /* heap and read-only chunks need no synchronization */ }
@Override default void close() throws IOException { /* no-op by default */ }
}
/**
* One channel ownership shared by all segments of an appended file and by
* read-only clones. A reference belongs to a ChunkedBytes instance, not to
* every segment, so a multi-chunk file is closed exactly once per owner.
*/
private static final class FileBacking implements Closeable {
final FileChannel channel;
private final AtomicInteger references;
FileBacking(final FileChannel channel) {
this.channel = channel;
this.references = new AtomicInteger(1);
}
FileBacking retain() {
int current;
do {
current = this.references.get();
if (current <= 0) throw new IllegalStateException("File backing is closed");
if (current == Integer.MAX_VALUE) {
throw new IllegalStateException("Too many references to file backing");
}
} while (!this.references.compareAndSet(current, current + 1));
return this;
}
@Override
public void close() throws IOException {
final int remaining = this.references.decrementAndGet();
if (remaining == 0) {
this.channel.close();
} else if (remaining < 0) {
throw new IOException("File backing closed more than once");
}
}
}
/** On-heap chunk. */
@@ -131,14 +178,14 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
/** File-backed chunk using MappedByteBuffer (lazy mapped). */
private static final class FileChunk implements Chunk, Cloneable {
final FileChannel ch;
final FileBacking backing;
final long fileOffset; // offset in the file where this chunk starts
final int len;
final boolean writable;
private volatile MappedByteBuffer mm; // lazily created
FileChunk(FileChannel ch, long fileOffset, int len, boolean writable) {
this.ch = ch; this.fileOffset = fileOffset; this.len = len; this.writable = writable;
FileChunk(FileBacking backing, long fileOffset, int len, boolean writable) {
this.backing = backing; this.fileOffset = fileOffset; this.len = len; this.writable = writable;
}
private MappedByteBuffer map() {
MappedByteBuffer local = this.mm;
@@ -148,9 +195,9 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
if (local == null) {
final FileChannel.MapMode mapMode = this.writable ? FileChannel.MapMode.READ_WRITE : FileChannel.MapMode.READ_ONLY;
try {
this.mm = local = this.ch.map(mapMode, this.fileOffset, this.len);
this.mm = local = this.backing.channel.map(mapMode, this.fileOffset, this.len);
} catch (final IOException e) {
throw new RuntimeException(e);
throw new UncheckedIOException(e);
}
}
}
@@ -166,7 +213,7 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
return take;
}
@Override public int write(long p, byte[] src, int off, int len) {
if (!this.writable) throw new RuntimeException("FileChunk is read-only");
if (!this.writable) throw new ReadOnlyBufferException();
if (p >= this.len) return -1;
final int take = Math.min(len, this.len - (int)p);
final ByteBuffer dup = this.map().duplicate();
@@ -178,18 +225,30 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
return this.map().get((int)p);
}
@Override public void set(long p, byte b) {
if (!this.writable) throw new RuntimeException("FileChunk is read-only");
if (!this.writable) throw new ReadOnlyBufferException();
this.map().put((int)p, b);
}
@Override public int length() { return this.len; }
@Override public Object clone() { return new FileChunk(this.ch, this.fileOffset, this.len, this.writable); }
@Override public void close() {
// Best-effort explicit unmap to release file handles promptly.
@Override public Object clone() { return new FileChunk(this.backing, this.fileOffset, this.len, this.writable); }
@Override public void force() {
final MappedByteBuffer local = this.mm;
if (local != null) {
try { Unmapper.unmap(local); } catch (final Throwable ignored) {}
if (this.writable && local != null) local.force();
}
@Override public void close() throws IOException {
/* Clear the reference before cleaning so close cannot clean it twice. */
final MappedByteBuffer local;
synchronized (this) {
local = this.mm;
this.mm = null;
}
// Channel is closed by owner (we don’t own it here).
if (local != null) {
try {
Unmapper.unmap(local);
} catch (final Exception e) {
throw new IOException("Cannot unmap file chunk at " + this.fileOffset, e);
}
}
// The shared FileBacking is released by the owning ChunkedBytes.
}
}
@@ -203,7 +262,7 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
final var theUnsafe = unsafeClass.getDeclaredField("theUnsafe");
theUnsafe.setAccessible(true);
final Object unsafe = theUnsafe.get(null);
unsafeClass.getMethod("invokeCleaner", MappedByteBuffer.class).invoke(unsafe, bb);
unsafeClass.getMethod("invokeCleaner", ByteBuffer.class).invoke(unsafe, bb);
return;
} catch (final Throwable ignore) {}
// Java 8: DirectBuffer.cleaner().clean()
@@ -218,22 +277,38 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
// ------------------- public API -------------------
/** Total logical size in bytes. */
public long size() { return this.size; }
public long size() {
requireOpen();
return this.size;
}
/** Append bytes from an InputStream into on-heap chunks. */
public void writeFrom(InputStream in) throws IOException {
public void writeFrom(final InputStream in) throws IOException {
requireOpen();
if (in == null) throw new NullPointerException("in");
final byte[] tmp = new byte[64 * 1024];
int r;
while ((r = in.read(tmp)) != -1) {
// InputStream.read(byte[]) may legally return 0
if (r == 0) continue;
/*
* InputStream.read() is foreign code and must run without holding the
* ChunkedBytes monitor. Otherwise an input implementation which owns
* another lock can invert that lock order when its owner concurrently
* calls append() or close(). append() provides the required, narrowly
* scoped synchronization for the actual state change.
*/
this.append(tmp, 0, r);
}
}
/** Append a byte array (copied into on-heap chunks). */
public void append(byte[] src, int off, int len) {
public synchronized void append(final byte[] src, int off, int len) {
requireOpen();
checkBounds(src, off, len, "src");
if (len == 0) return;
Math.addExact(this.size, len);
while (len > 0) {
int space = this.spaceInTailHeapChunk();
if (space == 0) {
@@ -256,51 +331,65 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
}
}
public void append(byte[] src) {
public void append(final byte[] src) {
if (src == null) throw new NullPointerException("src");
this.append(src, 0, src.length);
}
public void append(String src, int off, int len) {
public void append(final String src, final int off, final int len) {
if (src == null) throw new NullPointerException("src");
this.append(src.substring(off, off + len));
}
public void append(String src) {
public void append(final String src) {
if (src == null) throw new NullPointerException("src");
this.append(UTF8.getBytes(src));
}
/** Adopt an entire file as zero-copy file-backed segments (read-only). */
public void appendFile(Path path) {
public void appendFile(final Path path) {
this.appendFile(path, false);
}
/** Adopt a file; set writable=true to allow modifications to the mapped bytes. */
public void appendFile(Path path, boolean writable) {
public synchronized void appendFile(final Path path, final boolean writable) {
requireOpen();
if (path == null) throw new NullPointerException("path");
FileChannel ch = null;
try {
ch = FileChannel.open(path, writable
? new OpenOption[]{StandardOpenOption.READ, StandardOpenOption.WRITE}
: new OpenOption[]{StandardOpenOption.READ});
final long fileSize = ch.size();
if (fileSize == 0) return;
final long newSize = Math.addExact(this.size, fileSize);
final FileBacking backing = new FileBacking(ch);
final List<Segment> fileSegments = new ArrayList<>();
long pos = 0;
while (pos < fileSize) {
final int len = (int)Math.min(CHUNK_SIZE, fileSize - pos);
this.segments.add(new Segment(new FileChunk(ch, pos, len, writable), this.size, len));
this.size += len;
fileSegments.add(new Segment(
new FileChunk(backing, pos, len, writable), this.size + pos, len));
pos += len;
}
// We keep the FileChannel open until this ChunkedBytes is closed.
this.fileChannels.add(ch);
this.segments.addAll(fileSegments);
this.size = newSize;
// The shared backing owns the channel until this instance and all clones close.
this.fileBackings.add(backing);
ch = null; // prevent closing in finally
} catch (final IOException e) {
throw new RuntimeException(e);
throw new UncheckedIOException(e);
} finally {
if (ch != null) try { ch.close(); } catch (final IOException ignore) {}
}
}
/** Read into dst starting at logical position pos. Returns bytes read or -1 at EOF. */
public int read(long pos, byte[] dst, int off, int len) {
public int read(final long pos, final byte[] dst, final int off, final int len) {
requireOpen();
if (pos < 0) throw new IllegalArgumentException("pos < 0");
checkBounds(dst, off, len, "dst");
if (len == 0) return 0;
if (pos >= this.size) return -1;
long remaining = Math.min(len, this.size - pos);
int done = 0;
@@ -319,8 +408,11 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
}
/** Write bytes at logical position pos (requires writable backing for those ranges). */
public int write(long pos, byte[] src, int off, int len) {
public int write(final long pos, final byte[] src, final int off, final int len) {
requireOpen();
if (pos < 0) throw new IllegalArgumentException("pos < 0");
checkBounds(src, off, len, "src");
if (len == 0) return 0;
if (pos >= this.size) return -1;
long remaining = Math.min(len, this.size - pos);
int done = 0; int idx = this.findSegment(pos); long p = pos;
@@ -338,31 +430,48 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
/** InputStream view (no copying), supports >2 GB. */
public InputStream openStream() {
requireOpen();
return new InputStream() {
long pos = 0;
boolean streamClosed = false;
private void requireStreamOpen() throws IOException {
if (this.streamClosed) throw new IOException("Stream is closed");
}
@Override public int read() throws IOException {
requireStreamOpen();
final byte[] one = new byte[1];
final int n = this.read(one, 0, 1);
return n < 0 ? -1 : (one[0] & 0xFF);
}
@Override public int read(byte[] b, int off, int len) throws IOException {
requireStreamOpen();
final int n = ChunkedBytes.this.read(this.pos, b, off, len);
if (n > 0) this.pos += n;
return n;
}
@Override public long skip(long n) {
@Override public long skip(long n) throws IOException {
requireStreamOpen();
ChunkedBytes.this.requireOpen();
if (n <= 0) return 0;
final long k = Math.min(n, ChunkedBytes.this.size - this.pos);
this.pos += k; return k;
}
@Override public int available() {
@Override public int available() throws IOException {
requireStreamOpen();
ChunkedBytes.this.requireOpen();
final long rem = ChunkedBytes.this.size - this.pos;
return rem > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) rem;
}
@Override public void close() {
this.streamClosed = true;
}
};
}
/** Write the whole content to an OutputStream. */
public void writeTo(OutputStream out) {
public void writeTo(final OutputStream out) {
requireOpen();
if (out == null) throw new NullPointerException("out");
final byte[] tmp = new byte[256 * 1024];
long p = 0;
try {
@@ -373,13 +482,14 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
p += n;
}
} catch (final IOException e) {
throw new RuntimeException(e);
throw new UncheckedIOException(e);
}
}
/** Materialize as a single byte[] (only if total size fits in an int). */
public byte[] toByteArray() {
if (this.size > Integer.MAX_VALUE) throw new RuntimeException("Size > Integer.MAX_VALUE");
requireOpen();
if (this.size > Integer.MAX_VALUE) throw new IllegalStateException("Size > Integer.MAX_VALUE");
final byte[] all = new byte[(int) this.size];
this.writeTo(new ByteArrayOutputStream() {
int offset = 0;
@@ -393,6 +503,7 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
@Override
public synchronized void write(int b) {
requireOpen();
// Append single byte at end (OutputStream semantics)
final byte[] one = new byte[] { (byte) b };
this.append(one, 0, 1);
@@ -400,33 +511,63 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
@Override
public synchronized void write(byte[] b, int off, int len) {
if (b == null) throw new NullPointerException("b");
if (off < 0 || len < 0 || off + len > b.length) throw new IndexOutOfBoundsException();
requireOpen();
checkBounds(b, off, len, "b");
// Append to end; grows with heap chunks as needed
this.append(b, off, len);
}
public void writeBytes(byte[] b) {
if (b == null) throw new NullPointerException("b");
this.write(b, 0, b.length);
}
@Override
public void flush() {
// No-op for heap chunks; file-backed segments via MappedByteBuffer are
// written eagerly. If you need a hard sync to disk, expose a separate
// method to force() mapped buffers.
force();
}
@Override public void close() {
// Close/unmap all segments and channels.
Exception first = null;
/** Force changes in every writable mapped chunk to its storage device. */
public synchronized void force() {
requireOpen();
for (final Segment s : this.segments) {
try { s.close(); } catch (final Exception e) { if (first == null) first = e; }
s.force();
}
for (final FileChannel ch : this.fileChannels) {
try { ch.close(); } catch (final Exception e) { if (first == null) first = e; }
}
@Override
public synchronized void close() {
if (this.closed) return;
Exception failure = null;
try {
/* Mapped writes must be forced before the buffers are unmapped. */
force();
} catch (final RuntimeException e) {
failure = e;
}
if (first != null) throw new RuntimeException(first.getMessage());
this.closed = true;
for (final Segment s : this.segments) {
try {
s.close();
} catch (final Exception e) {
failure = addFailure(failure, e);
}
}
for (final FileBacking backing : this.fileBackings) {
try {
backing.close();
} catch (final Exception e) {
failure = addFailure(failure, e);
}
}
if (failure instanceof UncheckedIOException) throw (UncheckedIOException) failure;
if (failure instanceof IOException) {
throw new UncheckedIOException("Cannot close ChunkedBytes", (IOException) failure);
}
if (failure instanceof RuntimeException) throw (RuntimeException) failure;
if (failure != null) throw new RuntimeException("Cannot close ChunkedBytes", failure);
}
@Override
@@ -435,8 +576,9 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
}
public byte get(long pos) {
requireOpen();
if (pos < 0) throw new IllegalArgumentException("pos < 0");
if (pos >= this.size) throw new RuntimeException("pos >= size");
if (pos >= this.size) throw new IndexOutOfBoundsException("pos >= size");
final int idx = this.findSegment(pos);
final Segment s = this.segments.get(idx);
final long rel = pos - s.start;
@@ -444,11 +586,22 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
}
@Override
public Object clone() {
List<Segment> list = new ArrayList<>(this.segments.size());
for (Segment s: this.segments) list.add((Segment) s.clone());
final ChunkedBytes cb = new ChunkedBytes(list, this.size);
return cb;
public synchronized Object clone() {
requireOpen();
final List<FileBacking> retainedBackings = new ArrayList<>(this.fileBackings.size());
try {
for (final FileBacking backing : this.fileBackings) {
retainedBackings.add(backing.retain());
}
final List<Segment> clonedSegments = new ArrayList<>(this.segments.size());
for (final Segment segment : this.segments) {
clonedSegments.add((Segment) segment.clone());
}
return new ChunkedBytes(clonedSegments, this.size, retainedBackings);
} catch (final RuntimeException | Error e) {
releaseRetainedBackings(retainedBackings, e);
throw e;
}
}
@Override
@@ -457,44 +610,47 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
if (o instanceof ChunkedBytes) {
final ChunkedBytes cb = (ChunkedBytes)o;
if (this.size != cb.size) return false;
for (long i = 0; i < this.size; i++) {
if (this.get(i) != cb.get(i)) return false;
}
return true;
}
if (o instanceof byte[]) {
final byte[] b = (byte[])o;
if (this.size != b.length) return false;
for (int i = 0; i < b.length; i++) {
if (this.get(i) != b[i]) return false;
}
return true;
}
if (o instanceof String) {
return this.equals(UTF8.getBytes((String) o));
return compareChunked(cb) == 0;
}
return false;
}
@Override
public int hashCode() {
requireOpen();
int result = 1;
final byte[] buffer = new byte[64 * 1024];
long position = 0;
while (position < this.size) {
final int length = (int) Math.min(buffer.length, this.size - position);
final int read = this.read(position, buffer, 0, length);
if (read != length) throw new IllegalStateException("Short read while hashing");
for (int i = 0; i < read; i++) result = 31 * result + buffer[i];
position += read;
}
return result;
}
/** Content comparison retained for callers which previously used equals(byte[]). */
public boolean contentEquals(final byte[] bytes) {
requireOpen();
if (bytes == null || this.size != bytes.length) return false;
return compareBytes(bytes) == 0;
}
/** Content comparison retained for callers which previously used equals(String). */
public boolean contentEquals(final String text) {
requireOpen();
return text != null && contentEquals(UTF8.getBytes(text));
}
@Override
public int compareTo(Object o) {
if (o instanceof ChunkedBytes) {
final ChunkedBytes cb = (ChunkedBytes)o;
final int minLen = (int)Math.min(this.size, cb.size);
for (int i = 0; i < minLen; i++) {
final int diff = (this.get(i) & 0xFF) - (cb.get(i) & 0xFF);
if (diff != 0) return diff;
}
return Long.compare(this.size, cb.size);
return compareChunked((ChunkedBytes) o);
}
if (o instanceof byte[]) {
final byte[] b = (byte[])o;
final int minLen = (int)Math.min(this.size, b.length);
for (int i = 0; i < minLen; i++) {
final int diff = (this.get(i) & 0xFF) - (b[i] & 0xFF);
if (diff != 0) return diff;
}
return Long.compare(this.size, b.length);
return compareBytes((byte[]) o);
}
if (o instanceof String) {
return this.compareTo(UTF8.getBytes((String) o));
@@ -504,7 +660,81 @@ public final class ChunkedBytes extends OutputStream implements Comparable<Objec
// ------------------- internals -------------------
private final List<FileChannel> fileChannels = new ArrayList<>();
private final List<FileBacking> fileBackings = new ArrayList<>();
private void requireOpen() {
if (this.closed) throw new IllegalStateException("ChunkedBytes is closed");
}
private static void checkBounds(final byte[] bytes, final int offset,
final int length, final String name) {
if (bytes == null) throw new NullPointerException(name);
if (offset < 0 || length < 0 || offset > bytes.length - length) {
throw new IndexOutOfBoundsException(
"offset=" + offset + ", length=" + length
+ ", array length=" + bytes.length);
}
}
private static Exception addFailure(final Exception current, final Exception added) {
if (current == null) return added;
current.addSuppressed(added);
return current;
}
private static void releaseRetainedBackings(
final List<FileBacking> retainedBackings,
final Throwable failure) {
for (final FileBacking backing : retainedBackings) {
try {
backing.close();
} catch (final IOException closeFailure) {
failure.addSuppressed(closeFailure);
}
}
}
private int compareChunked(final ChunkedBytes other) {
requireOpen();
other.requireOpen();
final long minimumLength = Math.min(this.size, other.size);
final byte[] left = new byte[64 * 1024];
final byte[] right = new byte[left.length];
long position = 0;
while (position < minimumLength) {
final int length = (int) Math.min(left.length, minimumLength - position);
if (this.read(position, left, 0, length) != length
|| other.read(position, right, 0, length) != length) {
throw new IllegalStateException("Short read while comparing ChunkedBytes");
}
for (int i = 0; i < length; i++) {
final int difference = (left[i] & 0xff) - (right[i] & 0xff);
if (difference != 0) return difference;
}
position += length;
}
return Long.compare(this.size, other.size);
}
private int compareBytes(final byte[] bytes) {
requireOpen();
final long minimumLength = Math.min(this.size, bytes.length);
final byte[] left = new byte[64 * 1024];
long position = 0;
while (position < minimumLength) {
final int length = (int) Math.min(left.length, minimumLength - position);
if (this.read(position, left, 0, length) != length) {
throw new IllegalStateException("Short read while comparing bytes");
}
final int byteOffset = (int) position;
for (int i = 0; i < length; i++) {
final int difference = (left[i] & 0xff) - (bytes[byteOffset + i] & 0xff);
if (difference != 0) return difference;
}
position += length;
}
return Long.compare(this.size, bytes.length);
}
private int findSegment(long pos) {
int lo = 0, hi = this.segments.size() - 1;
source/net/yacy/kelondro/blob/ArrayStack.java
+126 -30
View File
@@ -95,6 +95,7 @@ public class ArrayStack implements BLOB {
private long repositoryAgeMax;
private long repositorySizeMax;
private List<blobItem> blobs;
private blobItem writableBlob;
private final String prefix;
private final int buffersize;
private final boolean trimall;
@@ -123,6 +124,7 @@ public class ArrayStack implements BLOB {
this.fileSizeLimit = maxFileSize;
this.repositoryAgeMax = Long.MAX_VALUE;
this.repositorySizeMax = Long.MAX_VALUE;
this.writableBlob = null;
this.trimall = trimall;
// init the thread pool for the keeperOf executor service
@@ -182,7 +184,6 @@ public class ArrayStack implements BLOB {
// find maximum time: the file with this time will be given a write buffer
final TreeMap<Long, blobItem> sortedItems = new TreeMap<Long, blobItem>();
BLOB oneBlob;
File f;
long maxtime = 0;
for (final String file : files) {
@@ -203,13 +204,17 @@ public class ArrayStack implements BLOB {
f = new File(heapLocation, file);
time = d.getTime();
try {
final blobItem item;
if (time == maxtime && !trimall) {
oneBlob = new Heap(f, keylength, ordering, buffersize);
final Heap heap = new Heap(f, keylength, ordering, buffersize);
item = new blobItem(d, f, heap);
this.writableBlob = item;
} else {
oneBlob = new HeapModifier(f, keylength, ordering);
oneBlob.optimize(); // no writings here, can be used with minimum memory
final ImmutableBLOB blob = new HeapModifier(f, keylength, ordering);
blob.optimize(); // no writings here, can be used with minimum memory
item = new blobItem(d, f, blob);
}
sortedItems.put(Long.valueOf(time), new blobItem(d, f, oneBlob));
sortedItems.put(Long.valueOf(time), item);
} catch (final IOException e) {
if (deleteonfail) {
ConcurrentLog.warn("KELONDRO", "ArrayStack: cannot read file " + f.getName() + ", deleting it (smart fail; alternative would be: crash; required user action would be same as deletion)");
@@ -257,14 +262,18 @@ public class ArrayStack implements BLOB {
} catch (final ParseException e) {
throw new IOException("date parse problem with file " + location.toString() + ": " + e.getMessage());
}
BLOB oneBlob;
if (full && this.buffersize > 0 && !this.trimall) {
oneBlob = new Heap(location, this.keylength, this.ordering, this.buffersize);
final blobItem item;
if (full && !this.trimall) {
sealWriter();
final Heap heap = new Heap(location, this.keylength, this.ordering, this.buffersize);
item = new blobItem(d, location, heap);
this.writableBlob = item;
} else {
oneBlob = new HeapModifier(location, this.keylength, this.ordering);
oneBlob.optimize();
final ImmutableBLOB blob = new HeapModifier(location, this.keylength, this.ordering);
blob.optimize();
item = new blobItem(d, location, blob);
}
this.blobs.add(new blobItem(d, location, oneBlob));
this.blobs.add(item);
}
private synchronized void unmountBLOB(final File location, final boolean writeIDX) {
@@ -273,6 +282,7 @@ public class ArrayStack implements BLOB {
b = this.blobs.get(i);
if (b.location.getAbsolutePath().equals(location.getAbsolutePath())) {
this.blobs.remove(i);
clearWritableReference(b);
b.blob.close(writeIDX);
b.blob = null;
b.location = null;
@@ -284,6 +294,7 @@ public class ArrayStack implements BLOB {
private File unmount(final int idx) {
final blobItem b = this.blobs.remove(idx);
clearWritableReference(b);
b.blob.close(false);
b.blob = null;
final File f = b.location;
@@ -413,6 +424,7 @@ public class ArrayStack implements BLOB {
while (!this.blobs.isEmpty() && System.currentTimeMillis() - this.blobs.get(0).creation.getTime() - this.fileAgeLimit > this.repositoryAgeMax) {
// too old
final blobItem oldestBLOB = this.blobs.remove(0);
clearWritableReference(oldestBLOB);
oldestBLOB.blob.close(false);
oldestBLOB.blob = null;
FileUtils.deletedelete(oldestBLOB.location);
@@ -422,6 +434,7 @@ public class ArrayStack implements BLOB {
while (!this.blobs.isEmpty() && length() > this.repositorySizeMax) {
// too large
final blobItem oldestBLOB = this.blobs.remove(0);
clearWritableReference(oldestBLOB);
oldestBLOB.blob.close(false);
FileUtils.deletedelete(oldestBLOB.location);
}
@@ -445,18 +458,89 @@ public class ArrayStack implements BLOB {
private class blobItem {
Date creation;
File location;
BLOB blob;
ImmutableBLOB blob;
BLOB writer;
public blobItem(final Date creation, final File location, final ImmutableBLOB blob) {
assert blob != null;
this.creation = creation;
this.location = location;
this.blob = blob;
this.writer = null;
}
public blobItem(final Date creation, final File location, final BLOB blob) {
assert blob != null;
this.creation = creation;
this.location = location;
this.blob = blob;
this.writer = blob;
}
public blobItem(final int buffer) throws IOException {
// make a new blob file and assign it in this item
this.creation = new Date();
this.location = newBLOB(this.creation);
this.blob = (buffer == 0) ? new HeapModifier(this.location, ArrayStack.this.keylength, ArrayStack.this.ordering) : new Heap(this.location, ArrayStack.this.keylength, ArrayStack.this.ordering, buffer);
final Heap heap = new Heap(this.location, ArrayStack.this.keylength, ArrayStack.this.ordering, buffer);
this.blob = heap;
this.writer = heap;
}
}
private void clearWritableReference(final blobItem item) {
if (item == this.writableBlob) this.writableBlob = null;
item.writer = null;
}
/**
* Close the current writable heap and reopen it with only the capabilities
* required by a sealed segment.
*/
private void sealWriter() throws IOException {
final blobItem item = this.writableBlob;
if (item == null) return;
if (item.writer == null) {
this.writableBlob = null;
return;
}
try {
item.writer.close(true);
} catch (final RuntimeException failure) {
/*
* HeapReader.close() releases the file, index and gap structures even
* when publishing their fingerprint pair fails. Keeping that Heap as
* writer would therefore leave ArrayStack pointing at a closed object.
* Reopen the authoritative blob before reporting the publish failure,
* so the caller may retry or continue writing to this ArrayStack.
*/
restoreWriterAfterFailedSeal(item, failure);
throw failure;
}
item.writer = null;
this.writableBlob = null;
try {
final ImmutableBLOB sealed = new HeapModifier(item.location, this.keylength, this.ordering);
sealed.optimize();
item.blob = sealed;
} catch (final IOException | RuntimeException failure) {
restoreWriterAfterFailedSeal(item, failure);
throw failure;
}
}
/** Restore the mutable state after either phase of sealing has failed. */
private void restoreWriterAfterFailedSeal(
final blobItem item, final Throwable failure) {
clearWritableReference(item);
try {
final Heap restored = new Heap(
item.location, this.keylength, this.ordering, this.buffersize);
item.blob = restored;
item.writer = restored;
this.writableBlob = item;
} catch (final IOException | RuntimeException restoreFailure) {
failure.addSuppressed(restoreFailure);
}
}
@@ -478,6 +562,7 @@ public class ArrayStack implements BLOB {
for (final blobItem bi: this.blobs) {
bi.blob.clear();
bi.blob.close(false);
clearWritableReference(bi);
HeapWriter.delete(bi.location);
}
this.blobs.clear();
@@ -695,7 +780,7 @@ public class ArrayStack implements BLOB {
@Override
protected byte[] next0() {
while (this.bii.hasNext()) {
final BLOB b = this.bii.next().blob;
final ImmutableBLOB b = this.bii.next().blob;
if (b == null) continue;
try {
final byte[] n = b.get(this.key);
@@ -752,7 +837,7 @@ public class ArrayStack implements BLOB {
@Override
protected Long next0() {
while (this.bii.hasNext()) {
final BLOB b = this.bii.next().blob;
final ImmutableBLOB b = this.bii.next().blob;
if (b == null) continue;
try {
final long l = b.length(this.key);
@@ -789,22 +874,18 @@ public class ArrayStack implements BLOB {
*/
@Override
public synchronized void insert(final byte[] key, final byte[] b) throws IOException {
blobItem bi = (this.blobs.isEmpty()) ? null : this.blobs.get(this.blobs.size() - 1);
/*
if (bi == null)
System.out.println("bi == null");
else if (System.currentTimeMillis() - bi.creation.getTime() > this.fileAgeLimit)
System.out.println("System.currentTimeMillis() - bi.creation.getTime() > this.maxage");
else if (bi.location.length() > this.fileSizeLimit)
System.out.println("bi.location.length() > this.maxsize");
*/
if ((bi == null) || (System.currentTimeMillis() - bi.creation.getTime() > this.fileAgeLimit) || (bi.location.length() > this.fileSizeLimit && this.fileSizeLimit >= 0)) {
blobItem bi = this.writableBlob;
if (bi == null
|| System.currentTimeMillis() - bi.creation.getTime() > this.fileAgeLimit
|| bi.location.length() > this.fileSizeLimit && this.fileSizeLimit >= 0) {
sealWriter();
// add a new blob to the array
bi = new blobItem(this.buffersize);
this.blobs.add(bi);
this.writableBlob = bi;
}
assert bi.blob instanceof Heap;
bi.blob.insert(key, b);
assert bi.writer != null;
bi.writer.insert(key, b);
executeLimits();
}
@@ -816,9 +897,12 @@ public class ArrayStack implements BLOB {
*/
@Override
public synchronized int replace(final byte[] key, final Rewriter rewriter) throws IOException, SpaceExceededException {
final Reducer reducer = rewriter::rewrite;
int d = 0;
for (final blobItem bi: this.blobs) {
d += bi.blob.replace(key, rewriter);
d += bi.writer == null
? bi.blob.reduce(key, reducer)
: bi.writer.replace(key, rewriter);
}
return d;
}
@@ -833,7 +917,9 @@ public class ArrayStack implements BLOB {
public synchronized int reduce(final byte[] key, final Reducer reduce) throws IOException, SpaceExceededException {
int d = 0;
for (final blobItem bi: this.blobs) {
d += bi.blob.reduce(key, reduce);
d += bi.writer == null
? bi.blob.reduce(key, reduce)
: bi.writer.replace(key, reduce);
}
return d;
}
@@ -903,10 +989,20 @@ public class ArrayStack implements BLOB {
*/
@Override
public synchronized void close(final boolean writeIDX) {
for (final blobItem bi: this.blobs) bi.blob.close(writeIDX);
RuntimeException failure = null;
for (final blobItem bi: this.blobs) {
clearWritableReference(bi);
try {
bi.blob.close(writeIDX);
} catch (final RuntimeException closeFailure) {
if (failure == null) failure = closeFailure;
else failure.addSuppressed(closeFailure);
}
}
this.blobs.clear();
this.blobs = null;
this.executor.shutdown();
if (failure != null) throw failure;
}
/**
source/net/yacy/kelondro/blob/BLOB.java
+4 -155
View File
@@ -4,10 +4,6 @@
//
// This is a part of YaCy, a peer-to-peer based web search engine
//
// $LastChangedDate$
// $LastChangedRevision$
// $LastChangedBy$
//
// LICENSE
//
// This program is free software; you can redistribute it and/or modify
@@ -28,120 +24,10 @@ package net.yacy.kelondro.blob;
import java.io.IOException;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.util.SpaceExceededException;
public interface BLOB {
/**
* return a name of the BLOB; can be the file name
* @return
*/
public String name();
/**
* ask for the length of the primary key
* @return the length of the key
*/
public int keylength();
/**
* return the underlying odering
* @return
*/
public ByteOrder ordering();
/**
* clears the content of the database
* @throws IOException
*/
public void clear() throws IOException;
/**
* trim the index of the database: this releases memory not currently used
* @throws IOException
*/
public void optimize();
/**
* calculate the memory in RAM that the BLOB occupies
* @return number of bytes that is used
*/
public long mem();
/**
* ask for the number of entries
* @return the number of entries in the table
*/
public int size();
/**
* ask if the BLOB is empty
* @return true iff size() == 0
*/
public boolean isEmpty();
/**
* iterator over all keys
* @param up
* @param rotating
* @return
* @throws IOException
*/
public CloneableIterator<byte[]> keys(boolean up, boolean rotating) throws IOException;
/**
* iterate over all keys
* @param up
* @param firstKey
* @return
* @throws IOException
*/
public CloneableIterator<byte[]> keys(boolean up, byte[] firstKey) throws IOException;
/**
* check if a specific key is in the database
* @param key the primary key
* @return
* @throws IOException
*/
public boolean containsKey(byte[] key);
/**
* retrieve the whole BLOB from the table
* @param key the primary key
* @return
* @throws IOException
*/
public byte[] get(byte[] key) throws IOException, SpaceExceededException;
public byte[] get(Object key);
/**
* retrieve the size of the database
* @param
* @return the size of the database in bytes
* @throws IOException
*/
public long length() throws IOException;
/**
* retrieve the size of the BLOB
* @param key
* @return the size of the BLOB or -1 if the BLOB does not exist
* @throws IOException
*/
public long length(byte[] key) throws IOException;
/**
* write a whole byte array as BLOB to the table
* @param key the primary key
* @param b
* @throws IOException
* @throws SpaceExceededException
*/
public void insert(byte[] key, byte[] b) throws IOException;
public interface BLOB extends ImmutableBLOB {
/**
* replace an existing entry in the BLOB with a new entry
@@ -160,49 +46,12 @@ public interface BLOB {
public int replace(byte[] key, Rewriter rewriter) throws IOException, SpaceExceededException;
/**
* a reduce method is the same as the replace. A replace subsumes a reduce method. A reduce method may be more simple.
* write a whole byte array as BLOB to the table
* @param key the primary key
* @param reducer
* @return the number of bytes that the rewriter reduced the BLOB
* @param b
* @throws IOException
* @throws SpaceExceededException
*/
public int reduce(byte[] key, Reducer reducer) throws IOException, SpaceExceededException;
/**
* remove a BLOB
* @param key the primary key
* @throws IOException
*/
public void delete(byte[] key) throws IOException;
/**
* close the BLOB table
*/
public void close(boolean writeIDX);
public interface Rewriter {
/**
* a rewrite method that is used in the replace functionality of a BLOB
* the result of such a rewrite must be always smaller or equal in size
* of the input
* @param b
* @return an array that is equal or smaller in size than b
* @throws SpaceExceededException
*/
public byte[] rewrite(byte[] b) throws SpaceExceededException;
}
public interface Reducer extends Rewriter {
/**
* A Reducer is a rewriter that reduced the content. There are no additional methods in this interface.
* The interface shall be used in replacement of a Rewriter to simply state the fact that the rewritement
* also reduces the content of a BLOB entry or may also keep the size the same;
*/
}
public void insert(byte[] key, byte[] b) throws IOException;
}
source/net/yacy/kelondro/blob/Gap.java
+41 -27
View File
@@ -34,6 +34,9 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.AtomicMoveNotSupportedException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;
@@ -101,36 +104,47 @@ public class Gap extends TreeMap<Long, Integer> {
* @return
* @throws IOException
*/
public int dump(File file) throws IOException {
File tmp = new File(file.getParentFile(), file.getName() + ".prt");
Iterator<Map.Entry<Long, Integer>> i = this.entrySet().iterator();
public int dump(final File file) throws IOException {
final File tmp = new File(file.getParentFile(), file.getName() + ".prt");
final Iterator<Map.Entry<Long, Integer>> i = this.entrySet().iterator();
DataOutputStream os;
int c = 0;
try (/* Resource automatically closed by this try-with-resources statement */
final FileOutputStream fileStream = new FileOutputStream(tmp);
) {
try {
os = new DataOutputStream(new BufferedOutputStream(fileStream, (Integer.SIZE + Long.SIZE) * 1024)); // = 16*1024*recordsize
} catch (final OutOfMemoryError e) {
os = new DataOutputStream(fileStream);
}
try {
Map.Entry<Long, Integer> e;
while (i.hasNext()) {
e = i.next();
os.writeLong(e.getKey().longValue());
os.writeInt(e.getValue().intValue());
c++;
}
os.flush();
} finally {
os.close();
}
try {
try (final FileOutputStream fileStream = new FileOutputStream(tmp)) {
try {
os = new DataOutputStream(new BufferedOutputStream(
fileStream, (Integer.SIZE + Long.SIZE) * 1024));
} catch (final OutOfMemoryError e) {
os = new DataOutputStream(fileStream);
}
try {
while (i.hasNext()) {
final Map.Entry<Long, Integer> entry = i.next();
os.writeLong(entry.getKey().longValue());
os.writeInt(entry.getValue().intValue());
c++;
}
os.flush();
} finally {
os.close();
}
}
try {
Files.move(tmp.toPath(), file.toPath(),
StandardCopyOption.ATOMIC_MOVE,
StandardCopyOption.REPLACE_EXISTING);
} catch (final AtomicMoveNotSupportedException e) {
Files.move(tmp.toPath(), file.toPath(),
StandardCopyOption.REPLACE_EXISTING);
}
} catch (final IOException e) {
try {
Files.deleteIfExists(tmp.toPath());
} catch (final IOException cleanupFailure) {
e.addSuppressed(cleanupFailure);
}
throw e;
}
tmp.renameTo(file);
assert file.exists() : file.toString();
assert !tmp.exists() : tmp.toString();
return c;
}
source/net/yacy/kelondro/blob/Heap.java
+41 -4
View File
@@ -28,6 +28,7 @@ package net.yacy.kelondro.blob;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.HashMap;
@@ -41,13 +42,14 @@ import net.yacy.cora.document.encoding.UTF8;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.cora.storage.HandleMap;
import net.yacy.cora.util.ConcurrentLog;
import net.yacy.cora.util.SpaceExceededException;
import net.yacy.kelondro.io.AbstractWriter;
import net.yacy.kelondro.util.MemoryControl;
public final class Heap extends HeapModifier implements BLOB {
public final class Heap extends AbstractHeapModifier<HandleMap> implements BLOB {
private SortedMap<byte[], byte[]> buffer; // a write buffer to limit IO to the file
private int buffersize; // bytes that are buffered in buffer
@@ -88,7 +90,7 @@ public final class Heap extends HeapModifier implements BLOB {
final int keylength,
final ByteOrder ordering,
int buffermax) throws IOException {
super(heapFile, keylength, ordering);
super(heapFile, keylength, ordering, HeapReader.mutable());
this.buffermax = buffermax;
this.buffer = new TreeMap<byte[], byte[]>(ordering);
this.buffersize = 0;
@@ -344,20 +346,48 @@ public final class Heap extends HeapModifier implements BLOB {
/**
* close the BLOB table
* @throws UncheckedIOException when buffered entries cannot be flushed
*/
@Override
public synchronized void close(final boolean writeIDX) {
ConcurrentLog.info("KELONDRO", "Heap: closing heap " + this.name());
IOException flushFailure = null;
if (this.file != null && this.buffer != null) {
try {
flushBufferInternal();
} catch (final IOException e) {
ConcurrentLog.logException(e);
flushFailure = e;
}
}
this.buffer = null;
super.close(writeIDX);
Throwable closeFailure = null;
try {
/*
* A failed flush may have changed the in-memory index without writing
* every corresponding record. Publishing that index would turn a
* recoverable I/O failure into a persistent, invalid fingerprint.
* Closing without a dump makes the next opener rebuild from the blob.
*/
super.close(writeIDX && flushFailure == null);
} catch (final Throwable failure) {
closeFailure = failure;
}
assert this.file == null;
if (flushFailure != null) {
if (closeFailure != null) flushFailure.addSuppressed(closeFailure);
throw new UncheckedIOException(
"Cannot flush heap buffer before closing " + this.name(),
flushFailure);
}
if (closeFailure instanceof RuntimeException) {
throw (RuntimeException) closeFailure;
}
if (closeFailure instanceof Error) throw (Error) closeFailure;
if (closeFailure != null) {
throw new IllegalStateException("Cannot close heap " + this.name(), closeFailure);
}
}
@Override
@@ -369,6 +399,13 @@ public final class Heap extends HeapModifier implements BLOB {
return this.buffermax;
}
@Override
public synchronized int replace(final byte[] key, final Rewriter rewriter) throws IOException, SpaceExceededException {
/* The shared in-place implementation operates on persisted records. */
flushBufferInternal();
return rewrite(key, rewriter);
}
/**
* write a whole byte array as BLOB to the table
* @param key the primary key
source/net/yacy/kelondro/blob/HeapModifier.java
+35 -16
View File
@@ -30,14 +30,24 @@ import java.util.SortedMap;
import net.yacy.cora.document.encoding.UTF8;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.storage.ImmutableHandleMap;
import net.yacy.cora.util.ConcurrentLog;
import net.yacy.cora.util.SpaceExceededException;
import net.yacy.kelondro.index.SSTableHandleMap;
import net.yacy.kelondro.io.CachedFileWriter;
import net.yacy.kelondro.util.FileUtils;
import net.yacy.kelondro.util.MemoryControl;
public class HeapModifier extends HeapReader implements BLOB {
public class HeapModifier extends AbstractHeapModifier<SSTableHandleMap> {
public HeapModifier(final File heapFile, final int keylength, final ByteOrder ordering) throws IOException {
super(heapFile, keylength, ordering, HeapReader.immutable());
}
}
abstract class AbstractHeapModifier<I extends ImmutableHandleMap> extends HeapReader<I> implements ImmutableBLOB {
/*
* This class adds a remove operation to a BLOBHeapReader. That means that a BLOBModifier can
@@ -54,8 +64,12 @@ public class HeapModifier extends HeapReader implements BLOB {
* @param ordering
* @throws IOException
*/
public HeapModifier(final File heapFile, final int keylength, final ByteOrder ordering) throws IOException {
super(heapFile, keylength, ordering);
protected AbstractHeapModifier(
final File heapFile,
final int keylength,
final ByteOrder ordering,
final HeapIndexFactory<I> indexFactory) throws IOException {
super(heapFile, keylength, ordering, indexFactory);
}
/**
@@ -118,6 +132,7 @@ public class HeapModifier extends HeapReader implements BLOB {
ConcurrentLog.severe("KELONDRO", "BLOBHeap: " + this.heapFile.getName() + ": too long size " + size + " in record at " + seek);
throw new IOException("BLOBHeap: " + this.heapFile.getName() + ": too long size " + size + " in record at " + seek);
}
prepareIndexForHeapMutation();
super.deleteFingerprint();
// add entry to free array
@@ -236,18 +251,16 @@ public class HeapModifier extends HeapReader implements BLOB {
}
}
@Override
public void insert(byte[] key, byte[] b) throws IOException {
throw new UnsupportedOperationException("put is not supported in BLOBHeapModifier");
}
@Override
public int replace(byte[] key, final Rewriter rewriter) throws IOException {
throw new UnsupportedOperationException();
}
@Override
public int reduce(byte[] key, final Reducer reducer) throws IOException, SpaceExceededException {
/**
* Rewrite an existing BLOB in place without changing its index entry.
*
* @param key the primary key
* @param rewriter produces content no larger than the existing content
* @return the number of bytes removed from the BLOB
* @throws IOException when the rewritten content cannot be stored in place
* @throws SpaceExceededException when the rewriter cannot allocate its result
*/
protected int rewrite(byte[] key, final Rewriter rewriter) throws IOException, SpaceExceededException {
key = normalizeKey(key);
assert key.length == this.keylength;
@@ -271,6 +284,7 @@ public class HeapModifier extends HeapReader implements BLOB {
if (MemoryControl.available() < len) {
if (!MemoryControl.request(len, true)) return 0; // not enough memory available for this blob
}
prepareIndexForHeapMutation();
super.deleteFingerprint();
// read the key
@@ -283,7 +297,7 @@ public class HeapModifier extends HeapReader implements BLOB {
this.file.readFully(blob, 0, blob.length);
// rewrite the entry
blob = reducer.rewrite(blob);
blob = rewriter.rewrite(blob);
int reduction = len - blob.length;
if (reduction == 0) {
// even if the reduction is zero then it is still be possible that the record has been changed
@@ -320,4 +334,9 @@ public class HeapModifier extends HeapReader implements BLOB {
}
}
@Override
public int reduce(final byte[] key, final Reducer reducer) throws IOException, SpaceExceededException {
return rewrite(key, reducer);
}
}
source/net/yacy/kelondro/blob/HeapReader.java
+505 -157
View File
@@ -29,6 +29,8 @@ import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.util.Arrays;
import java.util.Date;
import java.util.Iterator;
@@ -43,10 +45,12 @@ import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.order.Digest;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.cora.storage.HandleMap;
import net.yacy.cora.storage.ImmutableHandleMap;
import net.yacy.cora.util.ConcurrentLog;
import net.yacy.cora.util.LookAheadIterator;
import net.yacy.cora.util.SpaceExceededException;
import net.yacy.kelondro.index.RowHandleMap;
import net.yacy.kelondro.index.SSTableHandleMap;
import net.yacy.kelondro.io.CachedFileWriter;
import net.yacy.kelondro.io.Writer;
import net.yacy.kelondro.util.FileUtils;
@@ -54,12 +58,13 @@ import net.yacy.kelondro.util.MemoryControl;
import net.yacy.kelondro.util.RotateIterator;
public class HeapReader {
public abstract class HeapReader<I extends ImmutableHandleMap> {
//public final static long keepFreeMem = 20 * 1024 * 1024;
private final static ConcurrentLog log = new ConcurrentLog("KELONDRO");
private final HeapIndexFactory<I> indexFactory;
// input values
protected int keylength; // the length of the primary key
protected File heapFile; // the file of the heap
@@ -67,64 +72,320 @@ public class HeapReader {
// computed values
protected Writer file; // a random access to the file
protected HandleMap index; // key/seek relation for used records
protected I index; // key/seek relation for used records
protected Gap free; // set of {seek, size} pairs denoting space and position of free records
private File fingerprintFileIdx, fingerprintFileGap; // files with dumped indexes. Will be deleted if file is written
private Date closeDate; // records a time when the file was closed; used for debugging
public HeapReader(
interface HeapIndexFactory<I extends ImmutableHandleMap> {
I load(
int keylength,
ByteOrder ordering,
File dump)
throws IOException, SpaceExceededException;
HeapIndexBuilder<I> newBuilder(
File heapFile,
int keylength,
ByteOrder ordering,
int expectedspace) throws IOException;
void prepareForHeapMutation(I index) throws IOException;
}
interface HeapIndexBuilder<I extends ImmutableHandleMap> extends AutoCloseable {
void consume(byte[] key, long offset) throws IOException;
I finish() throws IOException;
@Override
void close() throws IOException;
}
@FunctionalInterface
interface HeapFileFactory {
Writer open(File heapFile) throws IOException;
}
static HeapIndexFactory<HandleMap> mutable() {
return new HeapIndexFactory<HandleMap>() {
@Override
public HandleMap load(
final int keylength,
final ByteOrder ordering,
final File dump)
throws IOException, SpaceExceededException {
return new RowHandleMap(keylength, ordering, 8, dump);
}
@Override
public HeapIndexBuilder<HandleMap> newBuilder(
final File heapFile,
final int keylength,
final ByteOrder ordering,
final int expectedspace) {
return new HeapIndexBuilder<HandleMap>() {
private RowHandleMap.initDataConsumer building =
RowHandleMap.asynchronusInitializer(
heapFile.getName() + ".initializer",
keylength, ordering, 8, expectedspace);
private boolean finishRequested = false;
@Override
public void consume(final byte[] key, final long offset) {
if (this.building == null) throw new IllegalStateException("index builder is closed");
this.building.consume(key, offset);
}
@Override
public HandleMap finish() throws IOException {
if (this.building == null) throw new IllegalStateException("index builder is closed");
requestFinish();
final RowHandleMap result = awaitResult();
this.building = null;
return result;
}
@Override
public void close() throws IOException {
if (this.building == null) return;
requestFinish();
try {
awaitResultForClose();
} finally {
this.building.close();
this.building = null;
}
}
private void requestFinish() {
if (this.finishRequested) return;
this.building.finish();
this.finishRequested = true;
}
private RowHandleMap awaitResult() throws IOException {
try {
return this.building.result();
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException("Interrupted while rebuilding mutable heap index", e);
} catch (final ExecutionException e) {
throw new IOException("Cannot rebuild mutable heap index", e.getCause());
}
}
private RowHandleMap awaitResultForClose() throws IOException {
boolean interrupted = false;
try {
while (true) {
try {
return this.building.result();
} catch (final InterruptedException e) {
interrupted = true;
} catch (final ExecutionException e) {
throw new IOException("Cannot rebuild mutable heap index", e.getCause());
}
}
} finally {
if (interrupted) Thread.currentThread().interrupt();
}
}
};
}
@Override
public void prepareForHeapMutation(final HandleMap index) {
/* RowHandleMap has no mapped fingerprint backing to detach. */
}
};
}
/**
* Index strategy for heaps which never add or replace keys.
*
* <p>The concrete generic type is intentional: this is not merely a factory
* returning some {@link ImmutableHandleMap}. Both loading a fingerprint and
* rebuilding from the heap are guaranteed to produce an
* {@link SSTableHandleMap}. A future accidental RowHandleMap fallback would
* therefore fail at compile time instead of silently restoring the RAM index.</p>
*/
static HeapIndexFactory<SSTableHandleMap> immutable() {
return new HeapIndexFactory<SSTableHandleMap>() {
@Override
public SSTableHandleMap load(
final int keylength,
final ByteOrder ordering,
final File dump)
throws IOException, SpaceExceededException {
/*
* Fingerprint dumps are published snapshots. SSTable removals write
* negative offsets, so the dump must never be mapped writable itself.
* openWorkingCopy() maps it read-only and promotes it to an owned
* temporary table only when the heap or index is first mutated.
*/
return SSTableHandleMap.openWorkingCopy(
keylength, ordering, 8, dump);
}
@Override
public HeapIndexBuilder<SSTableHandleMap> newBuilder(
final File heapFile,
final int keylength,
final ByteOrder ordering,
final int expectedspace) throws IOException {
final SSTableHandleMap.Builder builder = new SSTableHandleMap.Builder(
keylength, ordering, 8, heapFile.getParentFile(), heapFile.getName());
return new HeapIndexBuilder<SSTableHandleMap>() {
@Override
public void consume(final byte[] key, final long offset) throws IOException {
builder.consume(key, offset);
}
@Override
public SSTableHandleMap finish() throws IOException {
return builder.finish();
}
@Override
public void close() throws IOException {
builder.close();
}
};
}
@Override
public void prepareForHeapMutation(final SSTableHandleMap index)
throws IOException {
index.prepareForMutation();
}
};
}
protected HeapReader(
final File heapFile,
final int keylength,
final ByteOrder ordering) throws IOException {
final ByteOrder ordering,
final HeapIndexFactory<I> indexFactory) throws IOException {
this(heapFile, keylength, ordering, indexFactory, CachedFileWriter::new);
}
/** Constructor variant for an alternate heap-file ownership implementation. */
protected HeapReader(
final File heapFile,
final int keylength,
final ByteOrder ordering,
final HeapIndexFactory<I> indexFactory,
final HeapFileFactory heapFileFactory) throws IOException {
if (indexFactory == null) throw new IllegalArgumentException("indexFactory must not be null");
if (heapFileFactory == null) throw new IllegalArgumentException("heapFileFactory must not be null");
this.indexFactory = indexFactory;
this.ordering = ordering;
this.heapFile = heapFile;
this.keylength = keylength;
this.index = null; // will be created as result of initialization process
this.free = null; // will be initialized later depending on existing idx/gap file
this.heapFile.getParentFile().mkdirs();
this.file = new CachedFileWriter(this.heapFile);
this.file = heapFileFactory.open(this.heapFile);
this.closeDate = null;
// read or initialize the index
this.fingerprintFileIdx = null;
this.fingerprintFileGap = null;
if (initIndexReadDump()) {
// verify that everything worked just fine
// pick some elements of the index
Iterator<byte[]> i = this.index.keys(true, null);
int c = 3;
byte[] b, b1 = new byte[this.keylength];
long pos;
boolean ok = true;
while (i.hasNext() && c-- > 0) {
b = i.next();
pos = this.index.get(b);
this.file.seek(pos + 4);
this.file.readFully(b1, 0, b1.length);
if (!this.ordering.equal(b, b1)) {
ok = false;
break;
try {
// read or initialize the index
this.fingerprintFileIdx = null;
this.fingerprintFileGap = null;
if (initIndexReadDump()) {
if (!verifyLoadedIndex()) {
log.warn("HeapReader: verification of idx file for " + heapFile.toString() + " failed, re-building index");
/*
* Close the working index before invalidating its source fingerprint.
* This matters for mapped files on Windows and also guarantees that a
* rejected dump cannot leave a temporary SSTable behind.
*/
discardIndexState();
deleteFingerprint();
initIndexReadFromHeap();
} else {
log.info("HeapReader: using a dump of the index of " + heapFile.toString() + ".");
}
}
if (!ok) {
log.warn("HeapReader: verification of idx file for " + heapFile.toString() + " failed, re-building index");
initIndexReadFromHeap();
} else {
log.info("HeapReader: using a dump of the index of " + heapFile.toString() + ".");
// if we did not have a dump, create a new index
initIndexReadFromHeap();
}
} else {
// if we did not have a dump, create a new index
initIndexReadFromHeap();
// merge gaps that follow directly
mergeFreeEntries();
// after the initial initialization of the heap, we close the file again
// to make more room to file pointers which may run out if the number
// of file descriptors is too low and the number of files is too high
this.file.close();
// the file will be opened again automatically when the next access to it comes.
} catch (final IOException | RuntimeException | Error failure) {
cleanupFailedInitialization(failure);
throw failure;
}
}
// merge gaps that follow directly
mergeFreeEntries();
/** Release every resource acquired by a constructor which cannot return. */
private void cleanupFailedInitialization(final Throwable failure) {
if (this.index != null) {
try {
this.index.close();
} catch (final Throwable cleanupFailure) {
failure.addSuppressed(cleanupFailure);
} finally {
this.index = null;
}
}
if (this.free != null) {
try {
this.free.clear();
} catch (final Throwable cleanupFailure) {
failure.addSuppressed(cleanupFailure);
} finally {
this.free = null;
}
}
if (this.file != null) {
try {
this.file.close();
} catch (final Throwable cleanupFailure) {
failure.addSuppressed(cleanupFailure);
} finally {
this.file = null;
}
}
this.closeDate = new Date();
}
// after the initial initialization of the heap, we close the file again
// to make more room to file pointers which may run out if the number
// of file descriptors is too low and the number of files is too high
this.file.close();
// the file will be opened again automatically when the next access to it comes.
/**
* Verify a small sample before the loaded fingerprint becomes normal runtime state.
*
* <p>The index and gap dump form a cache, while the heap file remains the source of
* truth. A key mismatch, an invalid offset, or a short heap read therefore rejects
* the complete cached generation and lets the constructor rebuild it from the heap.
* The iterator is closed here so this validation phase owns all resources it opens.</p>
*/
private boolean verifyLoadedIndex() {
final CloneableIterator<byte[]> keys = this.index.keys(true, null);
try {
int remaining = 3;
final byte[] storedKey = new byte[this.keylength];
while (keys.hasNext() && remaining-- > 0) {
final byte[] indexedKey = keys.next();
final long position = this.index.get(indexedKey);
this.file.seek(position + 4);
this.file.readFully(storedKey, 0, storedKey.length);
if (!this.ordering.equal(indexedKey, storedKey)) return false;
}
return true;
} catch (final IOException | RuntimeException e) {
log.warn("HeapReader: cannot verify loaded index for " + this.heapFile, e);
return false;
} finally {
keys.close();
}
}
public long mem() {
@@ -135,6 +396,11 @@ public class HeapReader {
this.index.optimize();
}
/** Detach a mapped fingerprint before changing the heap generation. */
protected final void prepareIndexForHeapMutation() throws IOException {
this.indexFactory.prepareForHeapMutation(this.index);
}
protected byte[] normalizeKey(byte[] key) {
// check size of key: zero-filled keys are only possible of the ordering is
// an instance of the natural ordering. Base64-orderings cannot use zeros in keys.
@@ -170,39 +436,71 @@ public class HeapReader {
if (!this.fingerprintFileGap.exists()) this.fingerprintFileGap = new File(this.fingerprintFileGap.getAbsolutePath() + ".gz");
if (!this.fingerprintFileIdx.exists() || !this.fingerprintFileGap.exists()) {
deleteAllFingerprints(this.heapFile, this.fingerprintFileIdx.getName(), this.fingerprintFileGap.getName());
/* An incomplete pair must not be mistaken for a publishable snapshot. */
deleteFingerprint();
return false;
}
// there is an index and a gap file:
// read the index file:
/*
* Load both parts transactionally. In the immutable mode indexFactory.load()
* maps the published fingerprint read-only; it creates a working copy only
* on a later mutation. The loaded index must be closed immediately when
* loading the gap file or validating the pair fails; assigning fields early
* would leak that mapping when the heap scan starts its replacement builder.
*/
I loadedIndex = null;
try {
this.index = new RowHandleMap(this.keylength, this.ordering, 8, this.fingerprintFileIdx);
loadedIndex = this.indexFactory.load(
this.keylength, this.ordering, this.fingerprintFileIdx);
} catch (final IOException e) {
ConcurrentLog.logException(e);
deleteFingerprint();
return false;
} catch (final SpaceExceededException e) {
ConcurrentLog.logException(e);
deleteFingerprint();
return false;
}
// check saturation
if (this.index instanceof RowHandleMap) {
int[] saturation = ((RowHandleMap) this.index).saturation(); // {<the maximum length of consecutive equal-beginning bytes in the key>, <the minimum number of leading zeros in the second column>}
log.info("HeapReader: saturation of " + this.fingerprintFileIdx.getName() + ": keylength = " + saturation[0] + ", vallength = " + saturation[1] + ", size = " + this.index.size() +
", maximum saving for index-compression = " + (saturation[0] * this.index.size() / 1024 / 1024) + " MB" +
", exact saving for value-compression = " + (saturation[1] * this.index.size() / 1024 / 1024) + " MB");
if (loadedIndex instanceof RowHandleMap) {
final int[] saturation = ((RowHandleMap) loadedIndex).saturation(); // {<the maximum length of consecutive equal-beginning bytes in the key>, <the minimum number of leading zeros in the second column>}
log.info("HeapReader: saturation of " + this.fingerprintFileIdx.getName() + ": keylength = " + saturation[0] + ", vallength = " + saturation[1] + ", size = " + loadedIndex.size() +
", maximum saving for index-compression = " + (saturation[0] * loadedIndex.size() / 1024 / 1024) + " MB" +
", exact saving for value-compression = " + (saturation[1] * loadedIndex.size() / 1024 / 1024) + " MB");
}
// read the gap file:
final Gap loadedFree;
try {
this.free = new Gap(this.fingerprintFileGap);
loadedFree = new Gap(this.fingerprintFileGap);
} catch (final IOException e) {
ConcurrentLog.logException(e);
loadedIndex.close();
ConcurrentLog.logException(e);
deleteFingerprint();
return false;
}
// everything is fine now
return !this.index.isEmpty();
if (loadedIndex.isEmpty() && this.heapFile.length() > 0) {
loadedIndex.close();
loadedFree.clear();
deleteFingerprint();
return false;
}
this.index = loadedIndex;
this.free = loadedFree;
return true;
}
private void discardIndexState() {
if (this.index != null) {
this.index.close();
this.index = null;
}
if (this.free != null) {
this.free.clear();
this.free = null;
}
}
/**
@@ -252,65 +550,60 @@ public class HeapReader {
log.info("HeapReader: generating index for " + this.heapFile.toString() + ", " + (this.file.length() / 1024 / 1024) + " MB. Please wait.");
this.free = new Gap();
RowHandleMap.initDataConsumer indexready = RowHandleMap.asynchronusInitializer(this.name() + ".initializer", this.keylength, this.ordering, 8, Math.max(10, (int) (Runtime.getRuntime().freeMemory() / (10 * 1024 * 1024))));
byte[] key = new byte[this.keylength];
int reclen;
long seek = 0;
if (this.file.length() > 0) {
loop: while (true) { // don't test available() here because this does not work for files > 2GB
final int expectedspace = Math.max(10,
(int) (Runtime.getRuntime().freeMemory() / (10 * 1024 * 1024)));
try (HeapIndexBuilder<I> indexBuilder = this.indexFactory.newBuilder(
this.heapFile, this.keylength, this.ordering, expectedspace)) {
byte[] key = new byte[this.keylength];
int reclen;
long seek = 0;
if (this.file.length() > 0) {
loop: while (true) { // don't test available() here because this does not work for files > 2GB
try {
// go to seek position
this.file.seek(seek);
try {
// go to seek position
this.file.seek(seek);
// read length of the following record without the length of the record size bytes
reclen = this.file.readInt();
//assert reclen > 0 : " reclen == 0 at seek pos " + seek;
if (reclen == 0) {
// very bad file inconsistency
log.severe("HeapReader: reclen == 0 at seek pos " + seek + " in file " + this.heapFile);
this.file.setLength(seek); // delete everything else at the remaining of the file :-(
break loop;
// read length of the following record without the length of the record size bytes
reclen = this.file.readInt();
//assert reclen > 0 : " reclen == 0 at seek pos " + seek;
if (reclen == 0) {
// very bad file inconsistency
log.severe("HeapReader: reclen == 0 at seek pos " + seek + " in file " + this.heapFile);
this.file.setLength(seek); // delete everything else at the remaining of the file :-(
break loop;
}
// read key
this.file.readFully(key, 0, key.length);
} catch (final IOException e) {
// EOF reached
break loop; // terminate loop
}
// read key
this.file.readFully(key, 0, key.length);
} catch (final IOException e) {
// EOF reached
break loop; // terminate loop
}
// check if this record is empty
if (key == null || key[0] == 0) {
// it is an empty record, store to free list
if (reclen > 0) this.free.put(seek, reclen);
} else {
if (this.ordering.wellformed(key)) {
indexready.consume(key, seek);
key = new byte[this.keylength];
// check if this record is empty
if (key == null || key[0] == 0) {
// it is an empty record, store to free list
if (reclen > 0) this.free.put(seek, reclen);
} else {
// free the lost space
this.free.put(seek, reclen);
this.file.seek(seek + 4);
Arrays.fill(key, (byte) 0);
this.file.write(key); // mark the place as empty record
log.warn("HeapReader: BLOB " + this.heapFile.getName() + ": skiped not wellformed key " + UTF8.String(key) + " at seek pos " + seek);
if (this.ordering.wellformed(key)) {
indexBuilder.consume(key, seek);
key = new byte[this.keylength];
} else {
// free the lost space
this.free.put(seek, reclen);
this.file.seek(seek + 4);
Arrays.fill(key, (byte) 0);
this.file.write(key); // mark the place as empty record
log.warn("HeapReader: BLOB " + this.heapFile.getName() + ": skiped not wellformed key " + UTF8.String(key) + " at seek pos " + seek);
}
}
// new seek position
seek += 4L + reclen;
}
// new seek position
seek += 4L + reclen;
}
}
indexready.finish();
// finish the index generation
try {
this.index = indexready.result();
} catch (final InterruptedException e) {
ConcurrentLog.logException(e);
} catch (final ExecutionException e) {
ConcurrentLog.logException(e);
}
this.index = indexBuilder.finish();
}
log.info("HeapReader: finished index generation for " + this.heapFile.toString() + ", " + this.index.size() + " entries, " + this.free.size() + " gaps.");
}
@@ -328,6 +621,7 @@ public class HeapReader {
//System.out.println("*** DEBUG BLOB: free-seek = " + nextFree.seek + ", size = " + nextFree.size);
// check if they follow directly
if (lastFree.getKey() + lastFree.getValue() + 4 == nextFree.getKey()) {
if (merged == 0) prepareIndexForHeapMutation();
// merge those records
this.file.seek(lastFree.getKey());
lastFree.setValue(lastFree.getValue() + nextFree.getValue() + 4); // this updates also the free map
@@ -502,6 +796,7 @@ public class HeapReader {
log.severe("HeapReader: file " + this.file.file() + " corrupted at " + pos + ": negative len. len = " + len + ", pk.len = " + this.keylength);
// to get lazy over that problem (who wants to tell the user to stop operation and delete the file???) we work on like the entry does not exist
this.index.remove(key);
deleteFingerprint();
return null;
}
long memr = len + this.keylength + 64;
@@ -523,6 +818,7 @@ public class HeapReader {
// this is a severe operation, it should never happen.
// remove entry from index because keeping that element in the index would not make sense
this.index.remove(key);
deleteFingerprint();
// nothing to return
return null;
// but if the process ends in this state, it would completely fail
@@ -599,67 +895,119 @@ public class HeapReader {
/**
* close the BLOB table
* @throws UncheckedIOException when the heap cannot be closed or its complete
* fingerprint pair cannot be published
*/
public void close(boolean writeIDX) {
if (this.index == null) return;
synchronized (this.index) {
final I closingIndex = this.index;
if (closingIndex == null) return;
Throwable failure = null;
synchronized (closingIndex) {
try {
if (this.file != null)
try {
this.file.close();
} catch (final IOException e) {
ConcurrentLog.logException(e);
}
this.file = null;
if (writeIDX && this.index != null && this.free != null && (this.index.size() > 3 || this.free.size() > 3)) {
// now we can create a dump of the index and the gap information
// to speed up the next start
try {
String fingerprint = fingerprintFileHash(this.heapFile);
if (fingerprint == null) {
log.severe("HeapReader: cannot write a dump for " + this.heapFile.getName()+ ": fingerprint is null");
} else {
File newFingerprintFileGap = HeapWriter.fingerprintGapFile(this.heapFile, fingerprint);
if (this.fingerprintFileGap != null &&
this.fingerprintFileGap.getName().equals(newFingerprintFileGap.getName()) &&
this.fingerprintFileGap.exists()) {
log.info("HeapReader: using existing gap dump instead of writing a new one: " + this.fingerprintFileGap.getName());
} else {
long start = System.currentTimeMillis();
this.free.dump(newFingerprintFileGap);
log.info("HeapReader: wrote a dump for the " + this.free.size() + " gap entries of " + this.heapFile.getName()+ " in " + (System.currentTimeMillis() - start) + " milliseconds.");
}
}
this.free.clear();
this.free = null;
if (fingerprint != null) {
File newFingerprintFileIdx = HeapWriter.fingerprintIndexFile(this.heapFile, fingerprint);
if (this.fingerprintFileIdx != null &&
this.fingerprintFileIdx.getName().equals(newFingerprintFileIdx.getName()) &&
this.fingerprintFileIdx.exists()) {
log.info("HeapReader: using existing idx dump instead of writing a new one: " + this.fingerprintFileIdx.getName());
} else {
long start = System.currentTimeMillis();
this.index.dump(newFingerprintFileIdx);
log.info("HeapReader: wrote a dump for the " + this.index.size() + " index entries of " + this.heapFile.getName()+ " in " + (System.currentTimeMillis() - start) + " milliseconds.");
}
}
this.index.close();
this.index = null;
} catch (final IOException e) {
ConcurrentLog.logException(e);
if (this.file != null) this.file.close();
this.file = null;
if (writeIDX && this.free != null
&& (closingIndex.size() > 3 || this.free.size() > 3)) {
publishFingerprintGeneration(
this.heapFile, closingIndex, this.free,
this.fingerprintFileIdx, this.fingerprintFileGap,
"HeapReader");
}
} catch (final Throwable e) {
failure = e;
} finally {
this.file = null;
if (this.free != null) this.free.clear();
this.free = null;
try {
closingIndex.close();
} catch (final Throwable closeFailure) {
if (failure == null) failure = closeFailure;
else failure.addSuppressed(closeFailure);
}
if (this.index == closingIndex) this.index = null;
this.closeDate = new Date();
}
if (this.free != null) this.free.clear();
this.free = null;
if (this.index != null) this.index.close();
this.index = null;
this.closeDate = new Date();
} catch (Throwable e) {ConcurrentLog.logException(e);}
log.info("HeapReader: close HeapFile " + this.heapFile.getName());
log.fine("trace: " + ConcurrentLog.stackTrace());
}
if (failure instanceof IOException) {
final IOException ioFailure = (IOException) failure;
log.severe("HeapReader: cannot close and publish fingerprint for "
+ this.heapFile, ioFailure);
throw new UncheckedIOException(
"Cannot close and publish fingerprint for " + this.heapFile,
ioFailure);
}
if (failure instanceof RuntimeException) throw (RuntimeException) failure;
if (failure instanceof Error) throw (Error) failure;
if (failure != null) {
throw new IllegalStateException("Cannot close heap " + this.heapFile, failure);
}
}
/**
* Publish one coherent cache generation. The index is installed first and the
* gap file last; the latter acts as the commit marker because readers require
* both files. A reported failure removes every part installed by this attempt.
*/
static void publishFingerprintGeneration(
final File heapFile, final ImmutableHandleMap index, final Gap free,
final File reusableIndex, final File reusableGap,
final String publisher) throws IOException {
final String fingerprint = fingerprintFileHash(heapFile);
if (fingerprint == null) {
throw new IOException(publisher + ": cannot publish fingerprint for "
+ heapFile + ": fingerprint is null");
}
final File targetIndex = HeapWriter.fingerprintIndexFile(heapFile, fingerprint);
final File targetGap = HeapWriter.fingerprintGapFile(heapFile, fingerprint);
final boolean reusePair = sameFingerprintFile(reusableIndex, targetIndex)
&& sameFingerprintFile(reusableGap, targetGap);
if (reusePair) {
log.info(publisher + ": using existing fingerprint pair for "
+ heapFile.getName());
return;
}
final long start = System.currentTimeMillis();
boolean indexPublished = false;
boolean gapPublished = false;
try {
index.dump(targetIndex);
indexPublished = true;
free.dump(targetGap);
gapPublished = true;
log.info(publisher + ": published a fingerprint with " + index.size()
+ " index entries and " + free.size() + " gap entries for "
+ heapFile.getName() + " in "
+ (System.currentTimeMillis() - start) + " milliseconds.");
} catch (final IOException | RuntimeException e) {
final IOException failure = new IOException(
publisher + ": cannot publish complete fingerprint pair for "
+ heapFile, e);
if (gapPublished) deleteFailedPublish(targetGap, failure);
if (indexPublished) deleteFailedPublish(targetIndex, failure);
throw failure;
}
}
private static boolean sameFingerprintFile(
final File reusable, final File target) {
return reusable != null && reusable.exists()
&& reusable.getAbsoluteFile().equals(target.getAbsoluteFile());
}
private static void deleteFailedPublish(
final File file, final IOException failure) {
try {
Files.deleteIfExists(file.toPath());
} catch (final IOException cleanupFailure) {
failure.addSuppressed(cleanupFailure);
}
}
public synchronized void close() {
source/net/yacy/kelondro/blob/HeapWriter.java
+18 -17
View File
@@ -152,25 +152,26 @@ public final class HeapWriter {
if (this.heapFileTMP.exists()) throw new IOException("HeapWriter: renaming to " + this.heapFileTMP.toString() + " failed: file does not exist");
// generate index and gap files
if (writeIDX && this.index.size() > 3) {
// now we can create a dump of the index and the gap information
// to speed up the next start
long start = System.currentTimeMillis();
String fingerprint = HeapReader.fingerprintFileHash(this.heapFileREADY);
if (fingerprint == null) {
log.severe("HeapWriter: cannot write a dump for " + this.heapFileREADY.getName()+ ": fingerprint is null");
} else {
new Gap().dump(fingerprintGapFile(this.heapFileREADY, fingerprint));
this.index.dump(fingerprintIndexFile(this.heapFileREADY, fingerprint));
log.info("HeapWriter: wrote a dump for the " + this.index.size() + " index entries of " + this.heapFileREADY.getName()+ " in " + (System.currentTimeMillis() - start) + " milliseconds.");
IOException publishFailure = null;
try {
if (writeIDX && this.index.size() > 3) {
HeapReader.publishFingerprintGeneration(
this.heapFileREADY, this.index, new Gap(),
null, null, "HeapWriter");
}
} catch (final IOException e) {
publishFailure = e;
} finally {
try {
this.index.close();
} catch (final RuntimeException closeFailure) {
if (publishFailure == null) throw closeFailure;
publishFailure.addSuppressed(closeFailure);
} finally {
this.index = null;
}
this.index.close();
this.index = null;
} else {
// this is small.. just free resources, do not write index
this.index.close();
this.index = null;
}
if (publishFailure != null) throw publishFailure;
}
public static void delete(File f) {
source/net/yacy/kelondro/blob/ImmutableBLOB.java
+180
View File
@@ -0,0 +1,180 @@
// ImmutableBLOB.java
// (C) 2008 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
// first published 08.06.2008 on http://yacy.net
//
// This is a part of YaCy, a peer-to-peer based web search engine
//
// LICENSE
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
package net.yacy.kelondro.blob;
import java.io.IOException;
import java.io.UncheckedIOException;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.util.SpaceExceededException;
public interface ImmutableBLOB {
/**
* return a name of the BLOB; can be the file name
* @return
*/
public String name();
/**
* ask for the length of the primary key
* @return the length of the key
*/
public int keylength();
/**
* return the underlying odering
* @return
*/
public ByteOrder ordering();
/**
* clears the content of the database
* @throws IOException
*/
public void clear() throws IOException;
/**
* trim the index of the database: this releases memory not currently used
* @throws IOException
*/
public void optimize();
/**
* calculate the memory in RAM that the BLOB occupies
* @return number of bytes that is used
*/
public long mem();
/**
* ask for the number of entries
* @return the number of entries in the table
*/
public int size();
/**
* ask if the BLOB is empty
* @return true iff size() == 0
*/
public boolean isEmpty();
/**
* iterator over all keys
* @param up
* @param rotating
* @return
* @throws IOException
*/
public CloneableIterator<byte[]> keys(boolean up, boolean rotating) throws IOException;
/**
* iterate over all keys
* @param up
* @param firstKey
* @return
* @throws IOException
*/
public CloneableIterator<byte[]> keys(boolean up, byte[] firstKey) throws IOException;
/**
* check if a specific key is in the database
* @param key the primary key
* @return
* @throws IOException
*/
public boolean containsKey(byte[] key);
/**
* retrieve the whole BLOB from the table
* @param key the primary key
* @return
* @throws IOException
*/
public byte[] get(byte[] key) throws IOException, SpaceExceededException;
public byte[] get(Object key);
/**
* retrieve the size of the database
* @param
* @return the size of the database in bytes
* @throws IOException
*/
public long length() throws IOException;
/**
* retrieve the size of the BLOB
* @param key
* @return the size of the BLOB or -1 if the BLOB does not exist
* @throws IOException
*/
public long length(byte[] key) throws IOException;
/**
* remove a BLOB
* @param key the primary key
* @throws IOException
*/
public void delete(byte[] key) throws IOException;
/**
* a reduce method is the same as the replace. A replace subsumes a reduce method. A reduce method may be more simple.
* @param key the primary key
* @param reducer
* @return the number of bytes that the rewriter reduced the BLOB
* @throws IOException
* @throws SpaceExceededException
*/
public int reduce(byte[] key, Reducer reducer) throws IOException, SpaceExceededException;
public interface Rewriter {
/**
* a rewrite method that is used in the replace functionality of a BLOB
* the result of such a rewrite must be always smaller or equal in size
* of the input
* @param b
* @return an array that is equal or smaller in size than b
* @throws SpaceExceededException
*/
public byte[] rewrite(byte[] b) throws SpaceExceededException;
}
public interface Reducer extends Rewriter {
/**
* A Reducer is a rewriter that reduced the content. There are no additional methods in this interface.
* The interface shall be used in replacement of a Rewriter to simply state the fact that the rewritement
* also reduces the content of a BLOB entry or may also keep the size the same;
*/
}
/**
* Close the BLOB table and optionally publish its index fingerprint.
*
* @throws UncheckedIOException when closing storage or publishing the complete
* fingerprint pair fails; the signature remains unchecked for compatibility
* with the existing BLOB lifecycle
*/
public void close(boolean writeIDX);
}
source/net/yacy/kelondro/index/RowHandleMap.java
+52 -40
View File
@@ -32,6 +32,9 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.AtomicMoveNotSupportedException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Iterator;
@@ -52,7 +55,6 @@ import java.util.zip.GZIPOutputStream;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.storage.HandleMap;
import net.yacy.cora.util.ConcurrentLog;
import net.yacy.cora.util.SpaceExceededException;
import net.yacy.kelondro.util.NamePrefixThreadFactory;
import net.yacy.kelondro.workflow.WorkflowProcessor;
@@ -200,34 +202,48 @@ public final class RowHandleMap implements HandleMap, Iterable<Map.Entry<byte[],
final File tmp = new File(file.getParentFile(), file.getName() + ".prt");
final Iterator<Row.Entry> i = this.index.rows(true, null);
int c = 0;
final FileOutputStream fileStream = new FileOutputStream(tmp);
OutputStream os = null;
try {
try {
os = new BufferedOutputStream(fileStream, 4 * 1024 * 1024);
} catch (final OutOfMemoryError e) {
os = fileStream;
}
if (file.getName().endsWith(".gz")) os = new GZIPOutputStream(os, 65536){{def.setLevel(Deflater.BEST_COMPRESSION);}};
while (i.hasNext()) {
os.write(i.next().bytes());
c++;
}
os.flush();
} finally {
try {
if(os != null) {
os.close();
}
} finally {
if(fileStream != os) {
fileStream.close();
}
}
final FileOutputStream fileStream = new FileOutputStream(tmp);
OutputStream os = null;
try {
try {
os = new BufferedOutputStream(fileStream, 4 * 1024 * 1024);
} catch (final OutOfMemoryError e) {
os = fileStream;
}
if (file.getName().endsWith(".gz")) {
os = new GZIPOutputStream(os, 65536) {{
this.def.setLevel(Deflater.BEST_COMPRESSION);
}};
}
while (i.hasNext()) {
os.write(i.next().bytes());
c++;
}
os.flush();
} finally {
try {
if (os != null) os.close();
} finally {
if (fileStream != os) fileStream.close();
}
}
try {
Files.move(tmp.toPath(), file.toPath(),
StandardCopyOption.ATOMIC_MOVE,
StandardCopyOption.REPLACE_EXISTING);
} catch (final AtomicMoveNotSupportedException e) {
Files.move(tmp.toPath(), file.toPath(),
StandardCopyOption.REPLACE_EXISTING);
}
} catch (final IOException e) {
try {
Files.deleteIfExists(tmp.toPath());
} catch (final IOException cleanupFailure) {
e.addSuppressed(cleanupFailure);
}
throw e;
}
tmp.renameTo(file);
assert file.exists() : file.toString();
assert !tmp.exists() : tmp.toString();
return c;
}
@@ -490,21 +506,17 @@ public final class RowHandleMap implements HandleMap, Iterable<Map.Entry<byte[],
}
@Override
public final RowHandleMap call() throws IOException {
try {
finishloop: while (true) {
entry c;
try {
while ((c = this.cache.take()) != poisonEntry) {
this.map.putUnique(c.key, c.l);
}
break finishloop;
} catch (final InterruptedException e) {
continue finishloop;
public final RowHandleMap call() throws SpaceExceededException {
finishloop: while (true) {
entry c;
try {
while ((c = this.cache.take()) != poisonEntry) {
this.map.putUnique(c.key, c.l);
}
break finishloop;
} catch (final InterruptedException e) {
continue finishloop;
}
} catch (final SpaceExceededException e) {
ConcurrentLog.logException(e);
}
return this.map;
}
source/net/yacy/kelondro/index/SSTableHandleMap.java
+1420
View File
File diff suppressed because it is too large. Load diff
source/net/yacy/kelondro/rwi/ReferenceContainerArray.java
+2 -2
View File
@@ -35,7 +35,7 @@ import net.yacy.cora.sorting.Rating;
import net.yacy.cora.util.ConcurrentLog;
import net.yacy.cora.util.SpaceExceededException;
import net.yacy.kelondro.blob.ArrayStack;
import net.yacy.kelondro.blob.BLOB;
import net.yacy.kelondro.blob.ImmutableBLOB;
import net.yacy.kelondro.data.word.Word;
import net.yacy.kelondro.index.Row;
import net.yacy.kelondro.index.RowSet;
@@ -364,7 +364,7 @@ public final class ReferenceContainerArray<ReferenceType extends Reference> {
return this.array.reduce(termHash, new BLOBReducer(termHash, reducer));
}
public class BLOBReducer implements BLOB.Reducer {
public class BLOBReducer implements ImmutableBLOB.Reducer {
ContainerReducer<ReferenceType> rewriter;
byte[] wordHash;
source/net/yacy/search/Switchboard.java
+20 -2
View File
@@ -185,6 +185,7 @@ import net.yacy.kelondro.blob.Tables;
import net.yacy.kelondro.blob.Tables.SortDirection;
import net.yacy.kelondro.data.meta.URIMetadataNode;
import net.yacy.kelondro.data.word.Word;
import net.yacy.kelondro.index.SSTableHandleMap;
import net.yacy.kelondro.logging.GuiHandler;
import net.yacy.kelondro.logging.ThreadDump;
import net.yacy.kelondro.rwi.ReferenceContainer;
@@ -366,6 +367,25 @@ public final class Switchboard extends serverSwitch {
this.log.config("Index Primary Path: " + indexPath.toString());
final File archivePath = this.getDataPath(SwitchboardConstants.INDEX_ARCHIVE_PATH, SwitchboardConstants.INDEX_ARCHIVE_DEFAULT);
this.log.config("Index Archive Path: " + archivePath.toString());
this.htCachePath =
this.getDataPath(SwitchboardConstants.HTCACHE_PATH, SwitchboardConstants.HTCACHE_PATH_DEFAULT);
/*
* SSTable working and builder files are private scratch state, never a
* published fingerprint. A hard process stop can prevent normal cleanup.
* Recover all configured repository trees before any ArrayStack opens them;
* exclusive file locks protect tables still owned by another YaCy process.
*/
final SSTableHandleMap.RecoveryReport sstableRecovery =
SSTableHandleMap.recoverWorkingCopies(
dataPath, indexPath, archivePath, this.htCachePath);
if (sstableRecovery.candidates() > 0 || sstableRecovery.failures() > 0) {
this.log.info("SSTable crash recovery: deleted "
+ sstableRecovery.deleted() + " of "
+ sstableRecovery.candidates() + " temporary artifacts; "
+ sstableRecovery.inUse() + " still in use, "
+ sstableRecovery.failures() + " failures");
}
this.listsPath = this.getDataPath(SwitchboardConstants.LISTS_PATH, SwitchboardConstants.LISTS_PATH_DEFAULT);
this.log.config("Lists Path: " + this.listsPath.toString());
this.htRootPath = this.getDataPath(SwitchboardConstants.HTROOT_PATH, SwitchboardConstants.HTROOT_PATH_DEFAULT); // path to the servlets
@@ -770,8 +790,6 @@ public final class Switchboard extends serverSwitch {
this.log.config("Starting HT Cache Manager");
// create the cache directory
this.htCachePath =
this.getDataPath(SwitchboardConstants.HTCACHE_PATH, SwitchboardConstants.HTCACHE_PATH_DEFAULT);
this.log.info("HTCACHE Path = " + this.htCachePath.getAbsolutePath());
final long maxCacheSize =
1024L * 1024L * Long.parseLong(this.getConfig(SwitchboardConstants.PROXY_CACHE_SIZE, "2")); // this is megabyte
test/java/net/yacy/cora/util/ChunkedBytesTest.java
+232
View File
@@ -0,0 +1,232 @@
/*
* ChunkedBytesTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.cora.util;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
public class ChunkedBytesTest {
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void inputStreamHonorsZeroLengthAndNegativeSkipContracts() throws Exception {
try (ChunkedBytes empty = new ChunkedBytes();
InputStream input = empty.openStream()) {
assertEquals(0, input.read(new byte[1], 0, 0));
}
try (ChunkedBytes bytes = new ChunkedBytes(new byte[] {1, 2, 3});
InputStream input = bytes.openStream()) {
assertEquals(0L, input.skip(-1L));
assertEquals(1, input.read());
input.close();
assertThrows(IOException.class, input::read);
}
}
@Test
public void validatesRangesBeforeReadingOrWriting() {
try (ChunkedBytes bytes = new ChunkedBytes(new byte[] {1, 2, 3})) {
assertEquals(0, bytes.read(bytes.size(), new byte[0], 0, 0));
assertEquals(0, bytes.write(bytes.size(), new byte[0], 0, 0));
assertThrows(NullPointerException.class,
() -> bytes.read(bytes.size(), null, 0, 1));
assertThrows(IndexOutOfBoundsException.class,
() -> bytes.read(0, new byte[1], 1, 1));
assertThrows(IndexOutOfBoundsException.class,
() -> bytes.write(0, new byte[1], 0, -1));
assertThrows(IndexOutOfBoundsException.class,
() -> bytes.append(new byte[1], Integer.MAX_VALUE, 1));
}
}
@Test
public void equalContentHasEqualHashCodeAndExplicitForeignComparisons() {
try (ChunkedBytes left = new ChunkedBytes(new byte[] {1, 2, 3});
ChunkedBytes right = new ChunkedBytes(new byte[] {1, 2, 3})) {
assertEquals(left, right);
assertEquals(left.hashCode(), right.hashCode());
assertFalse(left.equals(new byte[] {1, 2, 3}));
assertFalse(left.equals("\u0001\u0002\u0003"));
assertTrue(left.contentEquals(new byte[] {1, 2, 3}));
assertTrue(left.contentEquals("\u0001\u0002\u0003"));
}
}
@Test
public void closeIsIdempotentAndRejectsFurtherAccess() {
final ChunkedBytes bytes = new ChunkedBytes(new byte[] {1});
bytes.close();
bytes.close();
assertThrows(IllegalStateException.class, bytes::size);
assertThrows(IllegalStateException.class, () -> bytes.get(0));
assertThrows(IllegalStateException.class, () -> bytes.append(new byte[] {2}));
assertThrows(IllegalStateException.class, bytes::openStream);
}
@Test
public void fileBackedCloneOutlivesOriginalOwner() throws Exception {
final File file = this.temporaryFolder.newFile("clone.bin");
Files.write(file.toPath(), new byte[] {42, 43});
final ChunkedBytes original = new ChunkedBytes();
original.appendFile(file.toPath());
final ChunkedBytes clone = (ChunkedBytes) original.clone();
original.close();
try {
assertEquals(42, clone.get(0));
assertArrayEquals(new byte[] {42, 43}, clone.toByteArray());
} finally {
clone.close();
}
}
@Test(timeout = 15000L)
public void comparesDifferentContentBeyondTwoGiB() throws Exception {
final long length = (long) Integer.MAX_VALUE + 1L;
final File leftFile = this.temporaryFolder.newFile("large-left.bin");
final File rightFile = this.temporaryFolder.newFile("large-right.bin");
createSparseFile(leftFile, length, 1);
createSparseFile(rightFile, length, 2);
try (ChunkedBytes left = new ChunkedBytes();
ChunkedBytes right = new ChunkedBytes()) {
left.appendFile(leftFile.toPath());
right.appendFile(rightFile.toPath());
assertEquals(length, left.size());
assertEquals(length, right.size());
assertTrue(left.compareTo(right) < 0);
assertTrue(right.compareTo(left) > 0);
}
}
@Test
public void writableMappingPersistsCrossChunkChangesOnFlushAndClose() throws Exception {
final File file = this.temporaryFolder.newFile("writable.bin");
final long patchPosition = ChunkedBytes.CHUNK_SIZE - 4L;
final byte[] patch = new byte[] {9, 8, 7, 6, 5, 4, 3, 2};
createSparseFile(file, ChunkedBytes.CHUNK_SIZE + 4L, 0);
try (ChunkedBytes bytes = new ChunkedBytes()) {
bytes.appendFile(file.toPath(), true);
assertEquals(patch.length,
bytes.write(patchPosition, patch, 0, patch.length));
bytes.flush();
}
final byte[] stored = new byte[patch.length];
try (RandomAccessFile input = new RandomAccessFile(file, "r")) {
input.seek(patchPosition);
input.readFully(stored);
}
assertArrayEquals(patch, stored);
}
@Test
public void appendFileReportsIoFailuresConsistently() {
final File missing = new File(this.temporaryFolder.getRoot(), "missing.bin");
try (ChunkedBytes bytes = new ChunkedBytes()) {
assertThrows(UncheckedIOException.class,
() -> bytes.appendFile(missing.toPath()));
}
}
@Test(timeout = 5000L)
public void writeFromDoesNotHoldMonitorWhileCallingInputStream() throws Exception {
final ChunkedBytes bytes = new ChunkedBytes();
final Object inputLock = new Object();
final CountDownLatch inputLockHeld = new CountDownLatch(1);
final CountDownLatch readEntered = new CountDownLatch(1);
final AtomicReference<Throwable> failure = new AtomicReference<>();
final InputStream input = new InputStream() {
@Override
public int read() {
throw new AssertionError("Bulk read expected");
}
@Override
public int read(final byte[] target, final int offset, final int length) {
readEntered.countDown();
synchronized (inputLock) {
return -1;
}
}
};
final Thread appender = new Thread(() -> {
synchronized (inputLock) {
inputLockHeld.countDown();
try {
if (!readEntered.await(2L, TimeUnit.SECONDS)) {
throw new AssertionError("writeFrom did not enter InputStream.read");
}
bytes.append(new byte[] {1});
} catch (final Throwable e) {
failure.compareAndSet(null, e);
}
}
}, "chunked-bytes-lock-owner");
appender.setDaemon(true);
appender.start();
assertTrue("Input lock was not acquired",
inputLockHeld.await(2L, TimeUnit.SECONDS));
final Thread writer = new Thread(() -> {
try {
bytes.writeFrom(input);
} catch (final Throwable e) {
failure.compareAndSet(null, e);
}
}, "chunked-bytes-stream-writer");
writer.setDaemon(true);
writer.start();
writer.join(2000L);
appender.join(2000L);
assertFalse("writeFrom and append deadlocked", writer.isAlive());
assertFalse("writeFrom and append deadlocked", appender.isAlive());
if (failure.get() != null) throw new AssertionError(failure.get());
assertArrayEquals(new byte[] {1}, bytes.toByteArray());
bytes.close();
}
private static void createSparseFile(final File file, final long length,
final int firstByte) throws IOException {
try (RandomAccessFile output = new RandomAccessFile(file, "rw")) {
output.setLength(length);
if (length > 0) {
output.seek(0L);
output.write(firstByte);
}
}
}
}
test/java/net/yacy/kelondro/blob/ArrayStackSSTableLifecycleTest.java
+308
View File
@@ -0,0 +1,308 @@
/*
* ArrayStackSSTableLifecycleTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.blob;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.UncheckedIOException;
import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.document.encoding.ASCII;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.kelondro.data.citation.CitationReference;
import net.yacy.kelondro.data.citation.CitationReferenceFactory;
import net.yacy.kelondro.index.RowHandleMap;
import net.yacy.kelondro.index.RowSet;
import net.yacy.kelondro.index.SSTableHandleMap;
import net.yacy.kelondro.rwi.ReferenceContainer;
public class ArrayStackSSTableLifecycleTest {
private static final int KEY_LENGTH = 12;
private static final int WRITE_BUFFER = 4096;
private static final String PREFIX = "segment";
private static final byte[] DELETE_KEY = ASCII.getBytes("AAAAAAAAAAAA");
private static final byte[] FILLER_B_KEY = ASCII.getBytes("BBBBBBBBBBBB");
private static final byte[] REDUCE_KEY = ASCII.getBytes("CCCCCCCCCCCC");
private static final byte[] SHARED_KEY = ASCII.getBytes("DDDDDDDDDDDD");
private static final byte[] FILLER_E_KEY = ASCII.getBytes("EEEEEEEEEEEE");
private static final byte[] ACTIVE_KEY = ASCII.getBytes("FFFFFFFFFFFF");
private static final byte[] FILLER_G_KEY = ASCII.getBytes("GGGGGGGGGGGG");
private static final byte[] FILLER_H_KEY = ASCII.getBytes("HHHHHHHHHHHH");
private static final byte[] FILLER_I_KEY = ASCII.getBytes("IIIIIIIIIIII");
private static final byte[] RECOVERED_KEY = ASCII.getBytes("JJJJJJJJJJJJ");
private static final CitationReferenceFactory REFERENCE_FACTORY =
new CitationReferenceFactory();
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void sealedSegmentsStaySSTablesAcrossRestartReduceDeleteAndMerge() throws Exception {
final File directory = this.temporaryFolder.newFolder("array-stack-sstable");
ArrayStack stack = newStack(directory);
try {
final byte[] twoReferences = container(
REDUCE_KEY, "000000000001", "000000000002");
final byte[] oneReference = container(REDUCE_KEY, "000000000001");
/*
* The first five inserts create the initial writable Heap. Five entries
* are intentional: Heap.close(true) then publishes an idx/gap pair which
* sealWriter() must reopen through HeapReader.immutable().
*/
stack.insert(DELETE_KEY, container(DELETE_KEY, "100000000001"));
stack.insert(FILLER_B_KEY, container(FILLER_B_KEY, "200000000001"));
stack.insert(REDUCE_KEY, twoReferences);
stack.insert(SHARED_KEY, container(SHARED_KEY, "300000000001"));
stack.insert(FILLER_E_KEY, container(FILLER_E_KEY, "400000000001"));
/*
* Mounting a newer full segment is the deterministic rollover path:
* ArrayStack seals the old writer and keeps only this new segment mutable.
*/
final File secondSegment = stack.newBLOB(
new Date(System.currentTimeMillis() + 60_000L));
writeSecondSegment(secondSegment);
stack.mountBLOB(secondSegment, true);
assertSegmentTypes(stack, HeapModifier.class, Heap.class);
assertTrue(((HeapModifier) segmentBlobs(stack).get(0)).index
instanceof SSTableHandleMap);
assertTrue(((Heap) segmentBlobs(stack).get(1)).index
instanceof RowHandleMap);
assertEquals(0L, countWorkingFiles(directory.toPath()));
assertArrayEquals(container(SHARED_KEY, "500000000001"),
stack.get(SHARED_KEY));
assertEquals(2, countOccurrences(stack, SHARED_KEY));
/* reduce() must use HeapModifier's in-place path on the sealed segment. */
assertEquals(twoReferences.length - oneReference.length,
stack.reduce(REDUCE_KEY, ignored -> oneReference));
assertEquals(1L, countWorkingFiles(directory.toPath()));
assertContainerSize(stack.get(REDUCE_KEY), 1);
stack.close(true);
stack = null;
assertEquals(0L, countWorkingFiles(directory.toPath()));
/* A process restart must reconstruct the same mutable/sealed boundary. */
stack = newStack(directory);
assertSegmentTypes(stack, HeapModifier.class, Heap.class);
assertTrue(((HeapModifier) segmentBlobs(stack).get(0)).index
instanceof SSTableHandleMap);
assertTrue(((Heap) segmentBlobs(stack).get(1)).index
instanceof RowHandleMap);
assertContainerSize(stack.get(REDUCE_KEY), 1);
assertEquals(0L, countWorkingFiles(directory.toPath()));
/*
* Exercise the actual ArrayStack merge protocol: unmount both sources,
* merge their ReferenceContainers, and mount the result as a sealed heap.
*/
final File[] mergeSources = stack.unmountSmallest(Long.MAX_VALUE);
assertNotNull(mergeSources);
assertEquals(0, stack.entries());
assertEquals(0L, countWorkingFiles(directory.toPath()));
final File mergedFile = stack.newBLOB(
new Date(System.currentTimeMillis() + 120_000L));
assertEquals(mergedFile, stack.mergeMount(
mergeSources[0], mergeSources[1], REFERENCE_FACTORY,
mergedFile, WRITE_BUFFER));
assertSegmentTypes(stack, HeapModifier.class);
assertTrue(((HeapModifier) segmentBlobs(stack).get(0)).index
instanceof SSTableHandleMap);
assertEquals(0L, countWorkingFiles(directory.toPath()));
assertContainerSize(stack.get(SHARED_KEY), 2);
assertContainerSize(stack.get(REDUCE_KEY), 1);
assertEquals(9, stack.size());
/* With one sealed segment delete() reaches HeapModifier without a RAM index. */
stack.delete(DELETE_KEY);
assertEquals(1L, countWorkingFiles(directory.toPath()));
assertFalse(stack.containsKey(DELETE_KEY));
assertEquals(8, stack.size());
stack.close(true);
stack = null;
assertEquals(0L, countWorkingFiles(directory.toPath()));
/* The newest (now sole) merged segment is writable after the next restart. */
stack = newStack(directory);
assertSegmentTypes(stack, Heap.class);
assertTrue(((Heap) segmentBlobs(stack).get(0)).index
instanceof RowHandleMap);
assertFalse(stack.containsKey(DELETE_KEY));
assertContainerSize(stack.get(SHARED_KEY), 2);
assertContainerSize(stack.get(REDUCE_KEY), 1);
assertEquals(8, stack.size());
assertEquals(0L, countWorkingFiles(directory.toPath()));
} finally {
if (stack != null) stack.close(false);
}
}
@Test
public void failedFingerprintPublishRestoresWritableHeap() throws Exception {
final File directory = this.temporaryFolder.newFolder("array-stack-publish-failure");
ArrayStack stack = new ArrayStack(
directory, PREFIX, NaturalOrder.naturalOrder,
KEY_LENGTH, 0, false, false);
try {
/* More than three entries make close(true) publish a fingerprint pair. */
stack.insert(DELETE_KEY, ASCII.getBytes("value-a"));
stack.insert(FILLER_B_KEY, ASCII.getBytes("value-b"));
stack.insert(REDUCE_KEY, ASCII.getBytes("value-c"));
stack.insert(SHARED_KEY, ASCII.getBytes("value-d"));
stack.insert(FILLER_E_KEY, ASCII.getBytes("value-e"));
final File activeBlob = onlyBlobFile(directory);
final String fingerprint = HeapReader.fingerprintFileHash(activeBlob);
final File gapTarget = HeapWriter.fingerprintGapFile(activeBlob, fingerprint);
/*
* A non-empty directory cannot be replaced by Gap.dump(). This forces
* publishing to fail after Heap.close() has already released its file
* and in-memory index, reproducing the stale-writer failure mode.
*/
assertTrue(gapTarget.mkdir());
Files.write(gapTarget.toPath().resolve("blocker"), new byte[] {1});
final File futureSegment = stack.newBLOB(
new Date(System.currentTimeMillis() + 60_000L));
UncheckedIOException publishFailure = null;
try {
stack.mountBLOB(futureSegment, true);
} catch (final UncheckedIOException expected) {
publishFailure = expected;
}
assertNotNull("The forced fingerprint publish must fail", publishFailure);
/* The original segment must already be reopened and writable. */
final byte[] recoveredValue = ASCII.getBytes("after-failure");
stack.insert(RECOVERED_KEY, recoveredValue);
assertArrayEquals(recoveredValue, stack.get(RECOVERED_KEY));
stack.close(true);
stack = null;
stack = new ArrayStack(
directory, PREFIX, NaturalOrder.naturalOrder,
KEY_LENGTH, 0, false, false);
assertArrayEquals(recoveredValue, stack.get(RECOVERED_KEY));
} finally {
if (stack != null) stack.close(false);
}
}
private static ArrayStack newStack(final File directory) throws Exception {
return new ArrayStack(directory, PREFIX, NaturalOrder.naturalOrder,
KEY_LENGTH, WRITE_BUFFER, false, false);
}
private static File onlyBlobFile(final File directory) {
final File[] blobs = directory.listFiles(
file -> file.getName().endsWith(".blob"));
assertNotNull(blobs);
assertEquals(1, blobs.length);
return blobs[0];
}
private static void writeSecondSegment(final File target) throws Exception {
final File temporary = new File(target.getParentFile(), target.getName() + ".prt");
final HeapWriter writer = new HeapWriter(
temporary, target, KEY_LENGTH, NaturalOrder.naturalOrder, WRITE_BUFFER);
writer.add(SHARED_KEY, container(SHARED_KEY, "500000000001"));
writer.add(ACTIVE_KEY, container(ACTIVE_KEY, "600000000001"));
writer.add(FILLER_G_KEY, container(FILLER_G_KEY, "700000000001"));
writer.add(FILLER_H_KEY, container(FILLER_H_KEY, "800000000001"));
writer.add(FILLER_I_KEY, container(FILLER_I_KEY, "900000000001"));
writer.close(true);
}
private static byte[] container(final byte[] termHash, final String... urlHashes)
throws Exception {
final ReferenceContainer<CitationReference> container =
new ReferenceContainer<>(REFERENCE_FACTORY, termHash);
long modified = 1_700_000_000_000L;
for (final String urlHash : urlHashes) {
container.add(new CitationReference(ASCII.getBytes(urlHash), modified++));
}
return container.exportCollection();
}
private static void assertContainerSize(final byte[] payload, final int expected)
throws Exception {
assertNotNull(payload);
assertEquals(expected,
RowSet.importRowSet(payload, REFERENCE_FACTORY.getRow()).size());
}
private static int countOccurrences(final ArrayStack stack, final byte[] key)
throws Exception {
int count = 0;
for (final byte[] payload : stack.getAll(key)) {
assertContainerSize(payload, 1);
count++;
}
return count;
}
private static void assertSegmentTypes(
final ArrayStack stack, final Class<?>... expectedTypes) throws Exception {
final List<ImmutableBLOB> segments = segmentBlobs(stack);
assertEquals(expectedTypes.length, segments.size());
for (int index = 0; index < expectedTypes.length; index++) {
assertEquals(expectedTypes[index], segments.get(index).getClass());
}
}
/**
* ArrayStack deliberately keeps blobItem private. Reflection is limited to
* this regression assertion; production code still exposes no mutable segment list.
*/
private static List<ImmutableBLOB> segmentBlobs(final ArrayStack stack) throws Exception {
final Field blobsField = ArrayStack.class.getDeclaredField("blobs");
blobsField.setAccessible(true);
final List<?> items = (List<?>) blobsField.get(stack);
final List<ImmutableBLOB> result = new ArrayList<>(items.size());
for (final Object item : items) {
final Field blobField = item.getClass().getDeclaredField("blob");
blobField.setAccessible(true);
result.add((ImmutableBLOB) blobField.get(item));
}
return result;
}
private static long countWorkingFiles(final Path directory) throws Exception {
try (java.util.stream.Stream<Path> files = Files.list(directory)) {
return files.filter(path -> path.getFileName().toString()
.endsWith(".sstwork")).count();
}
}
}
test/java/net/yacy/kelondro/blob/HeapCloseFailureTest.java
+104
View File
@@ -0,0 +1,104 @@
/*
* HeapCloseFailureTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.blob;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.document.encoding.ASCII;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.kelondro.io.Writer;
public class HeapCloseFailureTest {
private static final int KEY_LENGTH = 12;
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void closePropagatesFlushFailureWithoutPublishingStaleIndex() throws Exception {
final File directory = this.temporaryFolder.newFolder("heap-close-failure");
final File heapFile = new File(directory, "data.blob");
final Heap heap = new Heap(
heapFile, KEY_LENGTH, NaturalOrder.naturalOrder, 4096);
for (int index = 0; index < 5; index++) {
heap.insert(key(index), ASCII.getBytes("buffered-value-" + index));
}
/*
* Keep length(), seek() and close() operational, but fail the bulk write.
* At that point flushBufferInternal() has already assigned offsets to all
* buffered keys, so close(true) must not publish this stale RAM index.
*/
final Writer delegate = heap.file;
heap.file = (Writer) Proxy.newProxyInstance(
Writer.class.getClassLoader(),
new Class<?>[] {Writer.class},
(proxy, method, arguments) -> {
if (method.getName().equals("write")) {
throw new IOException("forced buffered write failure");
}
try {
return method.invoke(delegate, arguments);
} catch (final InvocationTargetException failure) {
throw failure.getCause();
}
});
try {
heap.close(true);
throw new AssertionError("flush failure was swallowed");
} catch (final UncheckedIOException expected) {
assertTrue(expected.getMessage().contains(heapFile.getName()));
assertEquals("forced buffered write failure",
expected.getCause().getMessage());
} finally {
heap.close(false);
}
assertFalse(hasFingerprintFile(directory));
/* No stale offsets may survive: reopening must scan the authoritative blob. */
final Heap reopened = new Heap(
heapFile, KEY_LENGTH, NaturalOrder.naturalOrder, 4096);
try {
assertEquals(0, reopened.size());
} finally {
reopened.close(false);
}
}
private static byte[] key(final int index) {
return ASCII.getBytes(String.format("%012d", Integer.valueOf(index)));
}
private static boolean hasFingerprintFile(final File directory) {
final File[] fingerprints = directory.listFiles(file -> {
final String name = file.getName();
return name.endsWith(".idx") || name.endsWith(".gap")
|| name.endsWith(".idx.gz") || name.endsWith(".gap.gz");
});
return fingerprints != null && fingerprints.length > 0;
}
}
test/java/net/yacy/kelondro/blob/HeapModifierWorkingCopyTest.java
+208
View File
@@ -0,0 +1,208 @@
/*
* HeapModifierWorkingCopyTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.blob;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.kelondro.index.SSTableHandleMap;
public class HeapModifierWorkingCopyTest {
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void modifierUsesPrivateGenerationAndPublishesCompactedFingerprint() throws Exception {
final File directory = this.temporaryFolder.newFolder("heap-working-copy");
final File temporaryHeap = new File(directory, "data.heap.prt");
final File heap = new File(directory, "data.heap");
/* HeapWriter publishes the initial immutable fingerprint generation. */
final HeapWriter writer = new HeapWriter(
temporaryHeap, heap, 4, NaturalOrder.naturalOrder, 4096);
for (int index = 1; index <= 5; index++) {
writer.add(key(index), value(index));
}
writer.close(true);
final File initialFingerprint = HeapWriter.fingerprintIndexFile(
heap, HeapReader.fingerprintFileHash(heap));
assertTrue(initialFingerprint.isFile());
final byte[] publishedBytes = Files.readAllBytes(initialFingerprint.toPath());
final HeapModifier first = new HeapModifier(
heap, 4, NaturalOrder.naturalOrder);
HeapModifier second = null;
try {
assertTrue(first.index instanceof SSTableHandleMap);
assertEquals(0, countWorkingFiles(directory.toPath()));
assertArrayEquals(value(2), first.get(key(2)));
assertArrayEquals(publishedBytes, Files.readAllBytes(initialFingerprint.toPath()));
/* Multiple readers remain on the published generation without copying. */
second = new HeapModifier(heap, 4, NaturalOrder.naturalOrder);
assertTrue(second.index instanceof SSTableHandleMap);
assertEquals(0, countWorkingFiles(directory.toPath()));
assertArrayEquals(value(2), second.get(key(2)));
second.close(false);
second = null;
assertEquals(0, countWorkingFiles(directory.toPath()));
/* The first heap mutation creates exactly one private generation. */
first.delete(key(2));
assertEquals(1, countWorkingFiles(directory.toPath()));
assertFalse(first.containsKey(key(2)));
assertFalse(initialFingerprint.exists());
first.close(true);
} finally {
if (second != null) second.close(false);
first.close(false);
}
assertEquals(0, countWorkingFiles(directory.toPath()));
/* close(true) compacted tombstones and published the next fingerprint. */
final File compactedFingerprint = HeapWriter.fingerprintIndexFile(
heap, HeapReader.fingerprintFileHash(heap));
assertTrue(compactedFingerprint.isFile());
final HeapModifier reopened = new HeapModifier(
heap, 4, NaturalOrder.naturalOrder);
try {
assertTrue(reopened.index instanceof SSTableHandleMap);
assertEquals(4, reopened.size());
assertFalse(reopened.containsKey(key(2)));
assertArrayEquals(value(5), reopened.get(key(5)));
} finally {
reopened.close(true);
}
assertEquals(0, countWorkingFiles(directory.toPath()));
}
@Test
public void publishFailureIsPropagatedAndRollsBackIncompletePair() throws Exception {
final File directory = this.temporaryFolder.newFolder("heap-publish-failure");
final File temporaryHeap = new File(directory, "data.heap.prt");
final File heap = new File(directory, "data.heap");
final HeapWriter writer = new HeapWriter(
temporaryHeap, heap, 4, NaturalOrder.naturalOrder, 4096);
for (int index = 1; index <= 5; index++) {
writer.add(key(index), value(index));
}
writer.close(true);
final HeapModifier modifier = new HeapModifier(
heap, 4, NaturalOrder.naturalOrder);
modifier.delete(key(2));
final String fingerprint = HeapReader.fingerprintFileHash(heap);
final File targetIndex = HeapWriter.fingerprintIndexFile(heap, fingerprint);
final File targetGap = HeapWriter.fingerprintGapFile(heap, fingerprint);
/* A non-empty directory cannot be replaced by the final gap dump. The
* index has already been published at that point and must be rolled back. */
assertTrue(targetGap.mkdir());
Files.write(targetGap.toPath().resolve("blocker"), new byte[] {1});
try {
modifier.close(true);
throw new AssertionError("publish failure was swallowed");
} catch (final UncheckedIOException expected) {
assertTrue(expected.getMessage().contains(heap.getName()));
assertTrue(expected.getCause().getMessage().contains("complete fingerprint pair"));
} finally {
modifier.close(false);
}
assertFalse(targetIndex.exists());
assertTrue(targetGap.isDirectory());
assertFalse(new File(targetIndex.getPath() + ".prt").exists());
assertFalse(new File(targetGap.getPath() + ".prt").exists());
assertEquals(0, countWorkingFiles(directory.toPath()));
}
@Test
public void incompleteFingerprintPairIsRebuiltFromHeapAndRepublished() throws Exception {
final File directory = this.temporaryFolder.newFolder("heap-pair-recovery");
final File temporaryHeap = new File(directory, "data.heap.prt");
final File heap = new File(directory, "data.heap");
final HeapWriter writer = new HeapWriter(
temporaryHeap, heap, 4, NaturalOrder.naturalOrder, 4096);
for (int index = 1; index <= 6; index++) {
writer.add(key(index), value(index));
}
writer.close(true);
final String fingerprint = HeapReader.fingerprintFileHash(heap);
final File indexDump = HeapWriter.fingerprintIndexFile(heap, fingerprint);
final File gapDump = HeapWriter.fingerprintGapFile(heap, fingerprint);
assertTrue(indexDump.isFile());
assertTrue(gapDump.isFile());
/* Simulate a crash between publishing the index and installing the gap
* commit marker. The remaining index must never be accepted alone. */
Files.delete(gapDump.toPath());
assertFalse(gapDump.exists());
final HeapModifier recovered = new HeapModifier(
heap, 4, NaturalOrder.naturalOrder);
try {
assertEquals(6, recovered.size());
assertArrayEquals(value(1), recovered.get(key(1)));
assertArrayEquals(value(6), recovered.get(key(6)));
assertFalse(indexDump.exists());
} finally {
recovered.close(true);
}
assertTrue(indexDump.isFile());
assertTrue(gapDump.isFile());
final HeapModifier restarted = new HeapModifier(
heap, 4, NaturalOrder.naturalOrder);
try {
assertEquals(6, restarted.size());
assertArrayEquals(value(4), restarted.get(key(4)));
} finally {
restarted.close(false);
}
assertEquals(0, countWorkingFiles(directory.toPath()));
}
private static long countWorkingFiles(final Path directory) throws Exception {
try (java.util.stream.Stream<Path> files = Files.list(directory)) {
return files.filter(path -> path.getFileName().toString()
.endsWith(".sstwork")).count();
}
}
private static byte[] key(final int value) {
return String.format("%04d", value).getBytes(StandardCharsets.US_ASCII);
}
private static byte[] value(final int value) {
return ("value-" + value).getBytes(StandardCharsets.US_ASCII);
}
}
test/java/net/yacy/kelondro/blob/HeapReaderInitializationFailureTest.java
+141
View File
@@ -0,0 +1,141 @@
/*
* HeapReaderInitializationFailureTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.blob;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
import java.nio.file.Files;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.document.encoding.ASCII;
import net.yacy.cora.order.ByteOrder;
import net.yacy.cora.order.NaturalOrder;
import net.yacy.cora.storage.HandleMap;
import net.yacy.kelondro.io.CachedFileWriter;
import net.yacy.kelondro.io.Writer;
public class HeapReaderInitializationFailureTest {
private static final int KEY_LENGTH = 4;
private static final byte[] KEY = ASCII.getBytes("0001");
private static final byte[] VALUE = ASCII.getBytes("payload");
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void rebuildFailureClosesBuilderAndHeapFile() throws Exception {
final File directory = this.temporaryFolder.newFolder("failed-rebuild");
final File heapFile = new File(directory, "data.blob");
writeHeapRecord(heapFile);
final AtomicBoolean fileClosed = new AtomicBoolean();
final AtomicBoolean builderClosed = new AtomicBoolean();
final HeapReader.HeapFileFactory files = file -> {
final Writer delegate = new CachedFileWriter(file);
return (Writer) Proxy.newProxyInstance(
Writer.class.getClassLoader(),
new Class<?>[] {Writer.class},
(proxy, method, arguments) -> {
if (method.getName().equals("close")) fileClosed.set(true);
try {
return method.invoke(delegate, arguments);
} catch (final InvocationTargetException failure) {
throw failure.getCause();
}
});
};
final HeapReader.HeapIndexFactory<HandleMap> indexes =
new HeapReader.HeapIndexFactory<HandleMap>() {
@Override
public HandleMap load(final int keylength, final ByteOrder ordering,
final File dump) {
throw new AssertionError("No fingerprint should be loaded");
}
@Override
public HeapReader.HeapIndexBuilder<HandleMap> newBuilder(
final File file, final int keylength,
final ByteOrder ordering, final int expectedspace) {
return new HeapReader.HeapIndexBuilder<HandleMap>() {
@Override
public void consume(final byte[] key, final long offset)
throws IOException {
throw new IOException("forced index rebuild failure");
}
@Override
public HandleMap finish() throws IOException {
throw new IOException("finish must not be reached");
}
@Override
public void close() {
builderClosed.set(true);
}
};
}
@Override
public void prepareForHeapMutation(final HandleMap index) {
}
};
try {
new TestHeapReader(heapFile, indexes, files);
throw new AssertionError("index rebuild failure was swallowed");
} catch (final IOException expected) {
assertEquals("forced index rebuild failure", expected.getMessage());
}
assertTrue("The failed builder must be closed", builderClosed.get());
assertTrue("The heap file must be closed", fileClosed.get());
/* Cleanup must leave the authoritative heap usable for a normal retry. */
final Heap recovered = new Heap(
heapFile, KEY_LENGTH, NaturalOrder.naturalOrder, 0);
try {
assertArrayEquals(VALUE, recovered.get(KEY));
} finally {
recovered.close(false);
}
}
private static void writeHeapRecord(final File heapFile) throws IOException {
try (DataOutputStream output = new DataOutputStream(
new BufferedOutputStream(Files.newOutputStream(heapFile.toPath())))) {
output.writeInt(KEY_LENGTH + VALUE.length);
output.write(KEY);
output.write(VALUE);
}
}
private static final class TestHeapReader extends HeapReader<HandleMap> {
private TestHeapReader(final File heapFile,
final HeapIndexFactory<HandleMap> indexFactory,
final HeapFileFactory heapFileFactory) throws IOException {
super(heapFile, KEY_LENGTH, NaturalOrder.naturalOrder,
indexFactory, heapFileFactory);
}
}
}
test/java/net/yacy/kelondro/index/SSTableHandleMapRecoveryAndLoadTest.java
+280
View File
@@ -0,0 +1,280 @@
/*
* SSTableHandleMapRecoveryAndLoadTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.index;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.order.NaturalOrder;
public class SSTableHandleMapRecoveryAndLoadTest {
private static final int KEY_LENGTH = Long.BYTES;
private static final int LARGE_RECORD_COUNT = 200000;
private static final int LOOKUPS_PER_READER = 20000;
private static final int REMOVALS = 5000;
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
/**
* Helper entry point used by {@link #globalRecoveryRemovesARealCrashArtifact()}.
* The process deliberately bypasses every shutdown hook and finally block.
*/
public static void main(final String[] arguments) throws Exception {
if (arguments.length != 2) return;
if ("--crash-helper".equals(arguments[0])) {
final File fingerprint = new File(arguments[1]);
final SSTableHandleMap table = SSTableHandleMap.openWorkingCopy(
KEY_LENGTH, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
table.remove(key(1));
signalReadyAndWait();
/* Keep the owning table strongly reachable while the parent verifies its
* cross-process lock. The helper is normally killed before this executes. */
table.size();
Runtime.getRuntime().halt(0);
}
if ("--builder-crash-helper".equals(arguments[0])) {
final SSTableHandleMap.Builder builder = new SSTableHandleMap.Builder(
KEY_LENGTH, NaturalOrder.naturalOrder, Long.BYTES,
new File(arguments[1]), "crash-helper", 1);
builder.consume(key(1), offset(1));
signalReadyAndWait();
/* Preserve the Builder and its session lock until the forced process stop. */
builder.consume(key(2), offset(2));
Runtime.getRuntime().halt(0);
}
}
private static void signalReadyAndWait() throws IOException {
System.out.println("READY");
System.out.flush();
System.in.read();
}
@Test(timeout = 30000)
public void globalRecoveryRemovesARealCrashArtifact() throws Exception {
final File repository = this.temporaryFolder.newFolder("crash-repository");
final File fingerprint = new File(repository, "crashed.idx");
writeTable(fingerprint, 32);
final byte[] published = Files.readAllBytes(fingerprint.toPath());
final File java = new File(new File(System.getProperty("java.home"), "bin"),
isWindows() ? "java.exe" : "java");
final Process process = new ProcessBuilder(
java.getAbsolutePath(), "-cp", System.getProperty("java.class.path"),
SSTableHandleMapRecoveryAndLoadTest.class.getName(),
"--crash-helper", fingerprint.getAbsolutePath())
.redirectErrorStream(true)
.start();
final BufferedReader processOutput = new BufferedReader(new InputStreamReader(
process.getInputStream(), StandardCharsets.UTF_8));
try {
/* Newer JVMs may print runtime diagnostics (for example about explicit
* mapped-buffer cleanup) before the helper reaches its synchronization
* marker. These messages must not be mistaken for helper failure. */
String outputLine;
do {
outputLine = processOutput.readLine();
} while (outputLine != null && !"READY".equals(outputLine));
assertEquals("READY", outputLine);
assertEquals(1L, countWorkingFiles(repository.toPath()));
final SSTableHandleMap.RecoveryReport whileRunning =
SSTableHandleMap.recoverWorkingCopies(repository);
assertEquals(1L, whileRunning.candidates());
assertEquals(0L, whileRunning.deleted());
assertEquals(1L, whileRunning.inUse());
assertEquals(0L, whileRunning.failures());
} finally {
process.destroyForcibly();
if (!process.waitFor(20, TimeUnit.SECONDS)) {
process.destroyForcibly();
throw new AssertionError("crash helper did not terminate");
}
}
assertEquals(1L, countWorkingFiles(repository.toPath()));
final SSTableHandleMap.RecoveryReport recovery =
SSTableHandleMap.recoverWorkingCopies(repository);
assertEquals(1L, recovery.candidates());
assertEquals(1L, recovery.deleted());
assertEquals(0L, recovery.inUse());
assertEquals(0L, recovery.failures());
assertEquals(0L, countWorkingFiles(repository.toPath()));
assertArrayEquals(published, Files.readAllBytes(fingerprint.toPath()));
}
@Test(timeout = 30000)
public void globalRecoveryRemovesRealBuilderCrashArtifacts() throws Exception {
final File repository = this.temporaryFolder.newFolder("builder-crash-repository");
final File java = new File(new File(System.getProperty("java.home"), "bin"),
isWindows() ? "java.exe" : "java");
final Process process = new ProcessBuilder(
java.getAbsolutePath(), "-cp", System.getProperty("java.class.path"),
SSTableHandleMapRecoveryAndLoadTest.class.getName(),
"--builder-crash-helper", repository.getAbsolutePath())
.redirectErrorStream(true)
.start();
final BufferedReader processOutput = new BufferedReader(new InputStreamReader(
process.getInputStream(), StandardCharsets.UTF_8));
try {
String outputLine;
do {
outputLine = processOutput.readLine();
} while (outputLine != null && !"READY".equals(outputLine));
assertEquals("READY", outputLine);
assertEquals(2L, countBuilderArtifacts(repository.toPath()));
final SSTableHandleMap.RecoveryReport whileRunning =
SSTableHandleMap.recoverWorkingCopies(repository);
assertEquals(2L, whileRunning.candidates());
assertEquals(0L, whileRunning.deleted());
assertEquals(2L, whileRunning.inUse());
assertEquals(0L, whileRunning.failures());
} finally {
process.destroyForcibly();
if (!process.waitFor(20, TimeUnit.SECONDS)) {
process.destroyForcibly();
throw new AssertionError("builder crash helper did not terminate");
}
}
final SSTableHandleMap.RecoveryReport recovery =
SSTableHandleMap.recoverWorkingCopies(repository);
assertEquals(2L, recovery.candidates());
assertEquals(2L, recovery.deleted());
assertEquals(0L, recovery.inUse());
assertEquals(0L, recovery.failures());
assertEquals(0L, countBuilderArtifacts(repository.toPath()));
}
@Test(timeout = 60000)
public void largeTablePromotesSafelyUnderConcurrentReadLoad() throws Exception {
final File repository = this.temporaryFolder.newFolder("load-repository");
final File fingerprint = new File(repository, "large.idx");
writeTable(fingerprint, LARGE_RECORD_COUNT);
final byte[] published = Files.readAllBytes(fingerprint.toPath());
final SSTableHandleMap table = SSTableHandleMap.openWorkingCopy(
KEY_LENGTH, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
final ExecutorService readers = Executors.newFixedThreadPool(4);
try {
assertEquals(LARGE_RECORD_COUNT, table.size());
assertEquals(0L, table.mem());
assertEquals(0L, countWorkingFiles(repository.toPath()));
final CountDownLatch start = new CountDownLatch(1);
final List<Future<?>> results = new ArrayList<>();
for (int reader = 0; reader < 4; reader++) {
final int seed = reader;
results.add(readers.submit(() -> {
final Random random = new Random(seed);
start.await();
for (int lookup = 0; lookup < LOOKUPS_PER_READER; lookup++) {
/* Readers stay in the untouched upper half while the lower
* half triggers and exercises the mapping transition. */
final int record = LARGE_RECORD_COUNT / 2
+ random.nextInt(LARGE_RECORD_COUNT / 2);
assertEquals(offset(record), table.get(key(record)));
}
return null;
}));
}
start.countDown();
for (int record = 0; record < REMOVALS; record++) {
assertEquals(offset(record), table.remove(key(record)));
}
for (final Future<?> result : results) result.get();
assertEquals(LARGE_RECORD_COUNT - REMOVALS, table.size());
assertEquals(-1L, table.get(key(REMOVALS - 1)));
assertEquals(offset(LARGE_RECORD_COUNT - 1),
table.get(key(LARGE_RECORD_COUNT - 1)));
assertEquals(1L, countWorkingFiles(repository.toPath()));
assertArrayEquals(published, Files.readAllBytes(fingerprint.toPath()));
} finally {
readers.shutdownNow();
try {
assertTrue(readers.awaitTermination(10, TimeUnit.SECONDS));
} finally {
table.close();
}
}
assertEquals(0L, countWorkingFiles(repository.toPath()));
}
private static void writeTable(final File target, final int records)
throws IOException {
try (DataOutputStream output = new DataOutputStream(
new BufferedOutputStream(Files.newOutputStream(target.toPath())))) {
for (int record = 0; record < records; record++) {
output.write(key(record));
output.write(NaturalOrder.encodeLong(offset(record), Long.BYTES));
}
}
}
private static long countWorkingFiles(final Path repository) throws IOException {
try (java.util.stream.Stream<Path> files = Files.walk(repository)) {
return files.filter(path -> path.getFileName().toString()
.endsWith(".sstwork")).count();
}
}
private static long countBuilderArtifacts(final Path repository) throws IOException {
try (java.util.stream.Stream<Path> files = Files.walk(repository)) {
return files.filter(path -> {
final String name = path.getFileName().toString();
return name.startsWith("sst-build-")
&& (name.endsWith(".sstlock") || name.endsWith(".run")
|| name.endsWith(".merge") || name.endsWith(".sst"));
}).count();
}
}
private static byte[] key(final long record) {
return NaturalOrder.encodeLong(record, KEY_LENGTH);
}
private static long offset(final long record) {
return record * 10L;
}
private static boolean isWindows() {
return System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win");
}
}
test/java/net/yacy/kelondro/index/SSTableHandleMapTest.java
+344
View File
@@ -0,0 +1,344 @@
/*
* SSTableHandleMapTest
* Copyright 2026 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*/
package net.yacy.kelondro.index;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import net.yacy.cora.order.CloneableIterator;
import net.yacy.cora.order.NaturalOrder;
public class SSTableHandleMapTest {
@Rule
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
@Test
public void builderMergesBoundedRunsAndRemovesItsWorkingFile() throws Exception {
final File directory = this.temporaryFolder.newFolder("builder");
final SSTableHandleMap table;
try (SSTableHandleMap.Builder builder = new SSTableHandleMap.Builder(
4, NaturalOrder.naturalOrder, Long.BYTES,
directory, "multipass", 1)) {
/* One buffered record per run forces both fan-in limiting and a second merge pass. */
for (int index = 129; index >= 0; index--) {
builder.consume(key(index), index);
}
builder.consume(key(50), 999L);
table = builder.finish();
}
try {
assertEquals(130, table.size());
assertEquals(0L, table.get(key(0)));
assertEquals(999L, table.get(key(50)));
assertEquals(129L, table.get(key(129)));
final CloneableIterator<byte[]> keys = table.keys(true, null);
int expected = 0;
while (keys.hasNext()) assertArrayEquals(key(expected++), keys.next());
assertEquals(130, expected);
assertEquals(1, directory.list().length);
} finally {
table.close();
}
assertEquals(0, directory.list().length);
}
@Test
public void workingCopyPromotesLazilyAndHonorsOwnershipLocks() throws Exception {
final File directory = this.temporaryFolder.newFolder("working-copy");
final File fingerprint = new File(directory, "heap.fingerprint.idx");
writeFingerprint(fingerprint);
final byte[] publishedSnapshot = Files.readAllBytes(fingerprint.toPath());
/* Simulate a working file left by a crashed process: it has no live lock. */
final String prefix = "sst-work-" + Integer.toHexString(
fingerprint.getAbsolutePath().hashCode()) + "-";
final Path crashedRepository = Files.createDirectories(
directory.toPath().resolve("repository/old-generation"));
Files.write(crashedRepository.resolve(prefix + "stale.sstwork"), new byte[] {1});
final SSTableHandleMap.RecoveryReport initialRecovery =
SSTableHandleMap.recoverWorkingCopies(directory, crashedRepository.toFile());
assertEquals(1L, initialRecovery.candidates());
assertEquals(1L, initialRecovery.deleted());
assertEquals(0L, initialRecovery.inUse());
assertEquals(0L, initialRecovery.failures());
final SSTableHandleMap first = SSTableHandleMap.openWorkingCopy(
4, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
SSTableHandleMap second = null;
try {
/* Reads stay directly on the published read-only mapping. */
assertEquals(0, countWorkingFiles(directory.toPath()));
assertEquals(20L, first.remove(key(2)));
assertEquals(1, countWorkingFiles(directory.toPath()));
/* The tombstone belongs to the private generation, never to the snapshot. */
assertArrayEquals(publishedSnapshot, Files.readAllBytes(fingerprint.toPath()));
second = SSTableHandleMap.openWorkingCopy(
4, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
assertEquals(1, countWorkingFiles(directory.toPath()));
assertEquals(20L, second.get(key(2)));
assertEquals(30L, second.remove(key(3)));
assertEquals(2, countWorkingFiles(directory.toPath()));
/* A global pass may run defensively without touching live generations. */
final SSTableHandleMap.RecoveryReport liveRecovery =
SSTableHandleMap.recoverWorkingCopies(directory);
assertEquals(2L, liveRecovery.candidates());
assertEquals(0L, liveRecovery.deleted());
assertEquals(2L, liveRecovery.inUse());
assertEquals(0L, liveRecovery.failures());
} finally {
if (second != null) second.close();
first.close();
}
assertEquals(0, countWorkingFiles(directory.toPath()));
assertArrayEquals(publishedSnapshot, Files.readAllBytes(fingerprint.toPath()));
}
@Test
public void recoveryCoversBuilderArtifactsAndPreservesUnrelatedFiles() throws Exception {
final File directory = this.temporaryFolder.newFolder("builder-recovery");
final String session = "sst-build-deadbeef-0123456789abcdef0123456789abcdef";
final String[] artifacts = {
session + ".sstlock",
session + "-1.run",
session + "-2.merge",
session + "-3.sst",
"sst-deadbeef-legacy.run",
"sst-deadbeef-legacy.merge",
"sst-deadbeef-legacy.sst"
};
for (final String artifact : artifacts) {
Files.write(directory.toPath().resolve(artifact), new byte[] {1});
}
final Path unrelatedRun = Files.write(
directory.toPath().resolve("application.run"), new byte[] {2});
final Path unrelatedSSTable = Files.write(
directory.toPath().resolve("sst-user-data.sst"), new byte[] {3});
final SSTableHandleMap.RecoveryReport recovery =
SSTableHandleMap.recoverWorkingCopies(directory);
assertEquals(artifacts.length, recovery.candidates());
assertEquals(artifacts.length, recovery.deleted());
assertEquals(0L, recovery.inUse());
assertEquals(0L, recovery.failures());
for (final String artifact : artifacts) {
assertFalse(Files.exists(directory.toPath().resolve(artifact)));
}
assertTrue(Files.isRegularFile(unrelatedRun));
assertTrue(Files.isRegularFile(unrelatedSSTable));
}
@Test
public void recoveryDoesNotDeleteActiveBuilderOrFinishedTable() throws Exception {
final File directory = this.temporaryFolder.newFolder("active-builder-recovery");
final SSTableHandleMap.Builder builder = new SSTableHandleMap.Builder(
4, NaturalOrder.naturalOrder, Long.BYTES,
directory, "active", 1);
SSTableHandleMap table = null;
try {
builder.consume(key(1), 10L);
final SSTableHandleMap.RecoveryReport whileBuilding =
SSTableHandleMap.recoverWorkingCopies(directory);
assertEquals(2L, whileBuilding.candidates());
assertEquals(0L, whileBuilding.deleted());
assertEquals(2L, whileBuilding.inUse());
assertEquals(0L, whileBuilding.failures());
table = builder.finish();
builder.close();
assertEquals(10L, table.get(key(1)));
final SSTableHandleMap.RecoveryReport whileOpen =
SSTableHandleMap.recoverWorkingCopies(directory);
assertEquals(1L, whileOpen.candidates());
assertEquals(0L, whileOpen.deleted());
assertEquals(1L, whileOpen.inUse());
assertEquals(0L, whileOpen.failures());
assertEquals(10L, table.get(key(1)));
} finally {
try {
builder.close();
} finally {
if (table != null) table.close();
}
}
assertEquals(0, directory.list().length);
}
@Test
public void workingCopyDecompressesGzipFingerprint() throws Exception {
final File directory = this.temporaryFolder.newFolder("working-copy-gzip");
final File fingerprint = new File(directory, "heap.fingerprint.idx.gz");
writeFingerprint(fingerprint);
final SSTableHandleMap table = SSTableHandleMap.openWorkingCopy(
4, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
try {
assertEquals(3, table.size());
assertEquals(30L, table.get(key(3)));
assertEquals(1, countWorkingFiles(directory.toPath()));
final SSTableHandleMap.RecoveryReport recovery =
SSTableHandleMap.recoverWorkingCopies(directory);
assertEquals(1L, recovery.candidates());
assertEquals(0L, recovery.deleted());
assertEquals(1L, recovery.inUse());
assertEquals(0L, recovery.failures());
} finally {
table.close();
}
assertEquals(0, countWorkingFiles(directory.toPath()));
}
@Test
public void clearAndRemoveOnePromoteOnlyWhenTheyChangeTheTable() throws Exception {
final File directory = this.temporaryFolder.newFolder("working-copy-mutations");
final File fingerprint = new File(directory, "heap.fingerprint.idx");
writeFingerprint(fingerprint);
final byte[] publishedSnapshot = Files.readAllBytes(fingerprint.toPath());
final SSTableHandleMap removeOne = SSTableHandleMap.openWorkingCopy(
4, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
try {
assertEquals(0, countWorkingFiles(directory.toPath()));
assertEquals(30L, removeOne.removeone());
assertEquals(1, countWorkingFiles(directory.toPath()));
} finally {
removeOne.close();
}
final SSTableHandleMap clear = SSTableHandleMap.openWorkingCopy(
4, NaturalOrder.naturalOrder, Long.BYTES, fingerprint);
try {
assertEquals(0, countWorkingFiles(directory.toPath()));
clear.clear();
assertEquals(0, clear.size());
assertEquals(1, countWorkingFiles(directory.toPath()));
clear.clear();
assertEquals(1, countWorkingFiles(directory.toPath()));
} finally {
clear.close();
}
assertEquals(0, countWorkingFiles(directory.toPath()));
assertArrayEquals(publishedSnapshot, Files.readAllBytes(fingerprint.toPath()));
}
@Test
public void rejectsTruncatedUnsortedAndDuplicateTables() throws Exception {
final File directory = this.temporaryFolder.newFolder("invalid-tables");
final File truncated = new File(directory, "truncated.idx");
Files.write(truncated.toPath(), new byte[11]);
assertInvalidTable(truncated, "not a multiple");
final File unsorted = new File(directory, "unsorted.idx");
writeRawTable(unsorted, new int[] {1, 3, 2, 4}, null);
assertInvalidTable(unsorted, "out of order");
final File duplicate = new File(directory, "duplicate.idx");
writeRawTable(duplicate, new int[] {1, 2, 2, 3}, null);
assertInvalidTable(duplicate, "duplicate key");
}
@Test
public void rejectsUnknownNegativeOffsetsButAcceptsTombstones() throws Exception {
final File directory = this.temporaryFolder.newFolder("invalid-offsets");
final File invalid = new File(directory, "negative.idx");
writeRawTable(invalid, new int[] {1, 2, 3}, new long[] {10L, -2L, 30L});
assertInvalidTable(invalid, "negative SSTable offset -2");
final File tombstoned = new File(directory, "tombstoned.idx");
writeRawTable(tombstoned, new int[] {1, 2, 3}, new long[] {10L, -1L, 30L});
final SSTableHandleMap table = new SSTableHandleMap(
4, NaturalOrder.naturalOrder, Long.BYTES, tombstoned);
try {
assertEquals(2, table.size());
assertEquals(-1L, table.get(key(2)));
assertEquals(30L, table.get(key(3)));
} finally {
table.close();
}
}
private static void writeFingerprint(final File target) throws Exception {
final RowHandleMap map = new RowHandleMap(
4, NaturalOrder.naturalOrder, Long.BYTES, 3, target.getName());
try {
map.putUnique(key(1), 10L);
map.putUnique(key(2), 20L);
map.putUnique(key(3), 30L);
map.dump(target);
} finally {
map.close();
}
}
private static void writeRawTable(final File target, final int[] keys,
final long[] offsets) throws IOException {
try (DataOutputStream output = new DataOutputStream(
new BufferedOutputStream(Files.newOutputStream(target.toPath())))) {
for (int index = 0; index < keys.length; index++) {
output.write(key(keys[index]));
output.write(NaturalOrder.encodeLong(
offsets == null ? keys[index] * 10L : offsets[index],
Long.BYTES));
}
}
}
private static void assertInvalidTable(final File tableFile,
final String expectedMessage) throws Exception {
SSTableHandleMap table = null;
try {
table = new SSTableHandleMap(
4, NaturalOrder.naturalOrder, Long.BYTES, tableFile);
throw new AssertionError("Invalid SSTable was accepted: " + tableFile);
} catch (final IOException expected) {
assertTrue(expected.getMessage(),
expected.getMessage().contains(expectedMessage));
} finally {
if (table != null) table.close();
}
}
private static long countWorkingFiles(final Path directory) throws Exception {
try (java.util.stream.Stream<Path> files = Files.walk(directory)) {
return files.filter(path -> path.getFileName().toString().endsWith(".sstwork")).count();
}
}
private static byte[] key(final int value) {
return String.format("%04d", value).getBytes(StandardCharsets.US_ASCII);
}
}