diff --git a/src/main/java/com/gliwka/hyperscan/wrapper/Database.java b/src/main/java/com/gliwka/hyperscan/wrapper/Database.java index 87500ec..a263222 100644 --- a/src/main/java/com/gliwka/hyperscan/wrapper/Database.java +++ b/src/main/java/com/gliwka/hyperscan/wrapper/Database.java @@ -19,6 +19,7 @@ */ public class Database implements Closeable { private final Map expressions; + private final Expression[] expressionsById; private final int expressionCount; private NativeDatabase database; @@ -30,7 +31,14 @@ void registerDeallocator() { } } - private Database(NativeDatabase database, List expressions) { + private static final Map BITMASK_TO_FLAG = + Collections.unmodifiableMap(Arrays.stream(ExpressionFlag.values()) + .collect(Collectors.toMap(ExpressionFlag::getBits, identity()))); + + private final Mode mode; + + private Database(NativeDatabase database, List expressions, Mode mode) { + this.mode = mode; this.database = database; this.expressionCount = expressions.size(); database.registerDeallocator(); @@ -49,6 +57,40 @@ private Database(NativeDatabase database, List expressions) { this.expressions.put(i++, expression); } } + + this.expressionsById = buildExpressionsById(expressions, hasIds); + } + + private static Expression[] buildExpressionsById(List expressions, boolean hasIds) { + int maxId = -1; + if (hasIds) { + for (Expression expression : expressions) { + Integer id = expression.getId(); + if (id != null && id > maxId) { + maxId = id; + } + } + } else { + maxId = expressions.size() - 1; + } + if (maxId < 0 || maxId > Math.max(4 * expressions.size(), 1024)) { + return null; + } + Expression[] byId = new Expression[maxId + 1]; + if (hasIds) { + for (Expression expression : expressions) { + Integer id = expression.getId(); + if (id != null) { + byId[id] = expression; + } + } + } else { + int i = 0; + for (Expression expression : expressions) { + byId[i++] = expression; + } + } + return byId; } private static void handleErrors(int hsError, hs_compile_error_t compileError, List expressions) throws CompileErrorException { @@ -94,6 +136,21 @@ public static Database compile(Expression... expressions) throws CompileErrorExc * @throws CompileErrorException If any of the expressions cannot be compiled */ public static Database compile(List expressions) throws CompileErrorException { + return compile(expressions, Mode.BLOCK); + } + + /** + * Compiles a list of expressions into a database in the given mode. + * Block-mode databases work with the block scanning methods, stream-mode + * databases with {@link Scanner#openStream(Database)}, and vectored-mode + * databases with the vectored scanning methods. + * + * @param expressions List of expressions to compile + * @param mode Compilation mode + * @return Compiled database + * @throws CompileErrorException If any of the expressions cannot be compiled + */ + public static Database compile(List expressions, Mode mode) throws CompileErrorException { try ( NativeExpressionCollection nativeExpressions = new NativeExpressionCollection(expressions); hs_compile_error_t errorT = new hs_compile_error_t(); @@ -106,17 +163,44 @@ public static Database compile(List expressions) throws CompileError nativeExpressions.getNativeFlags(), nativeExpressions.getNativeIds(), nativeExpressions.getSize(), - HS_MODE_BLOCK, + nativeMode(mode), null, database, error); handleErrors(hsError, error.get(hs_compile_error_t.class), expressions); - return new Database(database.get(NativeDatabase.class), expressions); + return new Database(database.get(NativeDatabase.class), expressions, mode); } } + /** + * compile an expression into a database in the given mode to use for scanning + * + * @param expression Expression to compile + * @param mode Compilation mode + * @return Compiled database + * @throws CompileErrorException If the expression cannot be compiled + */ + public static Database compile(Expression expression, Mode mode) throws CompileErrorException { + return compile(singletonList(expression), mode); + } + + private static int nativeMode(Mode mode) { + switch (mode) { + case STREAM: + return HS_MODE_STREAM; + case VECTORED: + return HS_MODE_VECTORED; + default: + return HS_MODE_BLOCK; + } + } + + Mode getMode() { + return mode; + } + NativeDatabase getDatabase() { return database; } @@ -139,6 +223,13 @@ public long getSize() { } Expression getExpression(int id) { + Expression[] byId = expressionsById; + if (byId != null && id >= 0 && id < byId.length) { + Expression expression = byId[id]; + if (expression != null) { + return expression; + } + } return expressions.get(id); } @@ -249,10 +340,6 @@ public static Database load(InputStream expressionsIn, InputStream databaseIn) t int expressionCount = expressionsDataIn.readInt(); List expressions = new ArrayList<>(expressionCount); - // Setup a lookup map for expression flags - Map bitmaskToFlag = Arrays.stream(ExpressionFlag.values()) - .collect(Collectors.toMap(ExpressionFlag::getBits, identity())); - for (int i = 0; i < expressionCount; i++) { int id = expressionsDataIn.readInt(); String pattern = expressionsDataIn.readUTF(); @@ -260,7 +347,7 @@ public static Database load(InputStream expressionsIn, InputStream databaseIn) t EnumSet flags = EnumSet.noneOf(ExpressionFlag.class); for (int j = 0; j < flagCount; j++) { int bitmask = expressionsDataIn.readInt(); - flags.add(bitmaskToFlag.get(bitmask)); + flags.add(BITMASK_TO_FLAG.get(bitmask)); } expressions.add(new Expression(pattern, flags, id == -1 ? null : id)); @@ -281,7 +368,9 @@ public static Database load(InputStream expressionsIn, InputStream databaseIn) t throw HyperscanException.hsErrorToException(hsError); } - return new Database(database, expressions); + // The mode is not recoverable from the serialized form; leave it + // unknown so API-level mode validation is skipped for loaded databases. + return new Database(database, expressions, null); } @Override diff --git a/src/main/java/com/gliwka/hyperscan/wrapper/Mode.java b/src/main/java/com/gliwka/hyperscan/wrapper/Mode.java new file mode 100644 index 0000000..c7f51f9 --- /dev/null +++ b/src/main/java/com/gliwka/hyperscan/wrapper/Mode.java @@ -0,0 +1,23 @@ +package com.gliwka.hyperscan.wrapper; + +/** + * Database compilation mode, determining which scanning APIs the compiled + * database can be used with. + */ +public enum Mode { + /** + * Block mode: the whole input is scanned in one call via + * {@link Scanner#scan(Database, byte[], ByteMatchEventHandler)} and friends. + */ + BLOCK, + /** + * Streaming mode: input is fed in chunks via + * {@link Scanner#openStream(Database)}. + */ + STREAM, + /** + * Vectored mode: input is presented as a list of segments via + * {@link Scanner#scanVector(Database, byte[][], ByteMatchEventHandler)}. + */ + VECTORED +} diff --git a/src/main/java/com/gliwka/hyperscan/wrapper/Scanner.java b/src/main/java/com/gliwka/hyperscan/wrapper/Scanner.java index cd94043..9c17dbe 100644 --- a/src/main/java/com/gliwka/hyperscan/wrapper/Scanner.java +++ b/src/main/java/com/gliwka/hyperscan/wrapper/Scanner.java @@ -2,10 +2,13 @@ import com.gliwka.hyperscan.jni.hs_database_t; import com.gliwka.hyperscan.jni.hs_scratch_t; +import com.gliwka.hyperscan.jni.hs_stream_t; import com.gliwka.hyperscan.jni.match_event_handler; import com.gliwka.hyperscan.wrapper.mapping.ByteCharMapping; import org.bytedeco.javacpp.BytePointer; +import org.bytedeco.javacpp.IntPointer; import org.bytedeco.javacpp.Pointer; +import org.bytedeco.javacpp.PointerPointer; import org.bytedeco.javacpp.SizeTPointer; import java.io.Closeable; @@ -13,6 +16,7 @@ import java.nio.Buffer; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.LinkedList; import java.util.List; @@ -29,6 +33,9 @@ */ public class Scanner implements Closeable { private static final ThreadLocal activeCallback = new ThreadLocal<>(); + private static final ThreadLocal scanBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocateDirect(0)); + private static final ThreadLocal hasMatchBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocateDirect(0)); + private static final ThreadLocal rawScanBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocateDirect(0)); private static class NativeScratch extends hs_scratch_t { void registerDeallocator() { @@ -144,16 +151,23 @@ public List scan(final Database db, final String input) { * @param eventHandler Handler to receive match events with string indices. */ public void scan(final Database db, final String input, StringMatchEventHandler eventHandler) { - ByteBuffer byteBuffer = ByteBuffer.allocateDirect(input.length() * 4); + int required = input.length() * 4; + ByteBuffer byteBuffer = scanBuffer.get(); + if (byteBuffer.capacity() < required) { + byteBuffer = ByteBuffer.allocateDirect(required); + scanBuffer.set(byteBuffer); + } else { + ((Buffer) byteBuffer).clear(); + } final ByteCharMapping mapping = Utf8Encoder.encodeToBufferAndMap(byteBuffer, input); scan(db, byteBuffer, (expressionId, fromByteIdx, toByteIdx, flags) -> { Expression expression = db.getExpression(expressionId); - long fromStringIndex = mapping.getMappingSize() > 0 ? mapping.getCharIndex((int)fromByteIdx) : 0; + long fromStringIndex = mapping.getCharIndex((int)fromByteIdx); long toStringIndex = 0; if(toByteIdx > 0) { - toStringIndex = mapping.getMappingSize() > 0 ? mapping.getCharIndex((int)toByteIdx - 1) : 0; + toStringIndex = mapping.getCharIndex((int)toByteIdx - 1); } return eventHandler.onMatch(expression, fromStringIndex, toStringIndex); @@ -171,10 +185,80 @@ public void scan(final Database db, final String input, StringMatchEventHandler * @param eventHandler Handler to receive match events with byte indices. */ public void scan(final Database db, final byte[] input, ByteMatchEventHandler eventHandler) { - scan(db, ByteBuffer.wrap(input), - (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> - eventHandler.onMatch(db.getExpression(expressionId), fromByteIdx, toByteIdx) - ); + RawMatchEventHandler rawHandler = input == null || input.length == 0 + ? (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> true + : (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(db.getExpression(expressionId), fromByteIdx, toByteIdx); + scanRaw(db, input, rawHandler); + } + + /** + * Scans a {@link ByteBuffer} for matches using a compiled expression database + * and reports matches to the provided event handler using byte indices. + * Bytes from the buffer's current position to its limit are scanned; the + * position and limit are not modified. + * Direct buffers are scanned zero-copy. Heap buffers are first copied into + * a reused per-thread direct buffer. + * Can only be executed one at a time on a per-instance basis. + * + * @param db Database containing expressions to use for matching. + * @param input Buffer to match against. + * @param eventHandler Handler to receive match events with byte indices. + */ + public void scan(final Database db, final ByteBuffer input, ByteMatchEventHandler eventHandler) { + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(db.getExpression(expressionId), fromByteIdx, toByteIdx); + if (input.isDirect()) { + scan(db, input, rawHandler); + return; + } + int remaining = input.remaining(); + ByteBuffer directBuffer = rawScanBuffer.get(); + if (directBuffer.capacity() < remaining) { + directBuffer = ByteBuffer.allocateDirect(remaining); + rawScanBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } + directBuffer.put(input.duplicate()); + ((Buffer) directBuffer).flip(); + scan(db, directBuffer, rawHandler); + } + + private int scanRaw(final Database db, final byte[] input, RawMatchEventHandler eventHandler) { + if (input != null && input.length > 0) { + ByteBuffer directBuffer = rawScanBuffer.get(); + if (directBuffer.capacity() < input.length) { + directBuffer = ByteBuffer.allocateDirect(input.length); + rawScanBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } + directBuffer.put(input); + ((Buffer) directBuffer).flip(); + return scan(db, directBuffer, eventHandler); + } + if (scratch == null) { + throw new IllegalStateException("Scratch space has already been deallocated"); + } + if (activeCallback.get() != null) { + throw new IllegalStateException("Recursive scanning is not supported."); + } + activeCallback.set(eventHandler); + int hsError = 0; + try { + hs_database_t database = db.getDatabase(); + try (final BytePointer bytePointer = new BytePointer(input)) { + int length = input == null ? 4 : input.length; + hsError = hs_scan(database, bytePointer, length, 0, scratch, matchHandler, null); + if (hsError != 0 && hsError != HS_SCAN_TERMINATED) { + throw HyperscanException.hsErrorToException(hsError); + } + } + } finally { + activeCallback.remove(); + } + return hsError; } /** @@ -238,10 +322,16 @@ public boolean hasMatch(final Database db, final ByteBuffer input) { * @return true if at least one match is found, false otherwise. */ public boolean hasMatch(final Database db, final byte[] input) { - // Allocate a direct buffer and copy data - ByteBuffer directBuffer = ByteBuffer.allocateDirect(input.length); + int required = input.length; + ByteBuffer directBuffer = hasMatchBuffer.get(); + if (directBuffer.capacity() < required) { + directBuffer = ByteBuffer.allocateDirect(required); + hasMatchBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } directBuffer.put(input); - ((Buffer)directBuffer).flip(); + ((Buffer) directBuffer).flip(); return hasMatch(db, directBuffer); } @@ -254,11 +344,339 @@ public boolean hasMatch(final Database db, final byte[] input) { * @return true if at least one match is found, false otherwise. */ public boolean hasMatch(final Database db, final String input) { - ByteBuffer byteBuffer = ByteBuffer.allocateDirect(input.length() * 4); + int required = input.length() * 4; + ByteBuffer byteBuffer = scanBuffer.get(); + if (byteBuffer.capacity() < required) { + byteBuffer = ByteBuffer.allocateDirect(required); + scanBuffer.set(byteBuffer); + } else { + ((Buffer) byteBuffer).clear(); + } Utf8Encoder.encodeToBufferAndMap(byteBuffer, input); return hasMatch(db, byteBuffer); } + /** + * Scans a sequence of byte arrays as one logical input using the vectored + * scanning mode. The segments are matched as if concatenated, and reported + * byte indices are relative to the start of the first segment. + * Requires a database compiled with {@link Mode#VECTORED}. + * Neither the arrays nor their contents may be null. + * + * @param db Database containing expressions to use for matching. + * @param inputs Segments to match against. + * @param eventHandler Handler to receive match events with byte indices. + */ + public void scanVector(final Database db, final byte[][] inputs, ByteMatchEventHandler eventHandler) { + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(db.getExpression(expressionId), fromByteIdx, toByteIdx); + int total = 0; + for (byte[] input : inputs) { + total += input.length; + } + ByteBuffer directBuffer = rawScanBuffer.get(); + if (directBuffer.capacity() < total) { + directBuffer = ByteBuffer.allocateDirect(Math.max(total, 1)); + rawScanBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } + for (byte[] input : inputs) { + directBuffer.put(input); + } + ((Buffer) directBuffer).flip(); + int[] lengths = new int[inputs.length]; + for (int i = 0; i < inputs.length; i++) { + lengths[i] = inputs[i].length; + } + scanVectorRaw(db, directBuffer, lengths, rawHandler); + } + + /** + * Scans a sequence of {@link ByteBuffer}s as one logical input using the + * vectored scanning mode. For every segment, bytes from position to limit + * are scanned; positions and limits are not modified. + * Direct segments are scanned zero-copy; heap segments are first copied + * into a reused per-thread direct buffer. + * The segments are matched as if concatenated, and reported byte indices + * are relative to the start of the first segment. + * Requires a database compiled with {@link Mode#VECTORED}. + * Neither the array nor its elements may be null. + * + * @param db Database containing expressions to use for matching. + * @param inputs Segments to match against. + * @param eventHandler Handler to receive match events with byte indices. + */ + public void scanVector(final Database db, final ByteBuffer[] inputs, ByteMatchEventHandler eventHandler) { + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(db.getExpression(expressionId), fromByteIdx, toByteIdx); + int heapTotal = 0; + for (ByteBuffer input : inputs) { + if (!input.isDirect()) { + heapTotal += input.remaining(); + } + } + ByteBuffer bulk = null; + if (heapTotal > 0) { + bulk = rawScanBuffer.get(); + if (bulk.capacity() < heapTotal) { + bulk = ByteBuffer.allocateDirect(heapTotal); + rawScanBuffer.set(bulk); + } else { + ((Buffer) bulk).clear(); + } + } + int n = inputs.length; + int[] lengths = new int[n]; + ByteBuffer[] segments = new ByteBuffer[n]; + int[] segmentStarts = new int[n]; + for (int i = 0; i < n; i++) { + ByteBuffer input = inputs[i]; + lengths[i] = input.remaining(); + if (input.isDirect()) { + segments[i] = input; + segmentStarts[i] = input.position(); + } else { + segmentStarts[i] = bulk.position(); + bulk.put(input.duplicate()); + segments[i] = bulk; + } + } + scanVectorRaw(db, segments, segmentStarts, lengths, rawHandler); + } + + private int scanVectorRaw(final Database db, final ByteBuffer packed, final int[] lengths, + RawMatchEventHandler eventHandler) { + int n = lengths.length; + ByteBuffer[] segments = new ByteBuffer[n]; + Arrays.fill(segments, packed); + int[] starts = new int[n]; + int offset = 0; + for (int i = 0; i < n; i++) { + starts[i] = offset; + offset += lengths[i]; + } + return scanVectorRaw(db, segments, starts, lengths, eventHandler, packed); + } + + private int scanVectorRaw(final Database db, final ByteBuffer[] segments, final int[] starts, + final int[] lengths, RawMatchEventHandler eventHandler) { + return scanVectorRaw(db, segments, starts, lengths, eventHandler, null); + } + + private int scanVectorRaw(final Database db, final ByteBuffer[] segments, final int[] starts, + final int[] lengths, RawMatchEventHandler eventHandler, ByteBuffer packed) { + if (scratch == null) { + throw new IllegalStateException("Scratch space has already been deallocated"); + } + if (db.getMode() != null && db.getMode() != Mode.VECTORED) { + throw new IllegalArgumentException("Vectored scanning requires a database compiled with Mode.VECTORED"); + } + if (activeCallback.get() != null) { + throw new IllegalStateException("Recursive scanning is not supported."); + } + activeCallback.set(eventHandler); + int hsError; + int n = lengths.length; + ByteBuffer keepAlive = packed != null ? packed : segments[0]; + try { + hs_database_t database = db.getDatabase(); + try (PointerPointer data = new PointerPointer<>(n); + IntPointer lengthPtr = new IntPointer(lengths)) { + BytePointer base = new BytePointer(keepAlive); + for (int i = 0; i < n; i++) { + BytePointer element = new BytePointer(segments[i] == keepAlive ? base : new BytePointer(segments[i])); + data.put(i, element.position(starts[i])); + } + hsError = hs_scan_vector(database, data, lengthPtr, n, 0, scratch, matchHandler, null); + if (hsError != 0 && hsError != HS_SCAN_TERMINATED) { + throw HyperscanException.hsErrorToException(hsError); + } + } + } finally { + activeCallback.remove(); + } + return hsError; + } + + /** + * Opens a streaming scan session over the given database. The returned + * stream shares this scanner's scratch space and must be closed after use, + * preferably with try-with-resources. Closing without a handler discards + * pending matches; use {@link Stream#close(ByteMatchEventHandler)} to + * receive them. + * Not thread-safe, just like the owning scanner. + * + * @param db Database containing expressions to use for matching. + * @return Open stream ready for scanning. + */ + public Stream openStream(final Database db) { + if (scratch == null) { + throw new IllegalStateException("Scratch space has already been deallocated"); + } + if (db.getMode() != null && db.getMode() != Mode.STREAM) { + throw new IllegalArgumentException("Streaming requires a database compiled with Mode.STREAM"); + } + hs_stream_t nativeStream; + try (PointerPointer streamOut = new PointerPointer<>(1)) { + streamOut.put(0, new hs_stream_t()); + int hsError = hs_open_stream(db.getDatabase(), 0, streamOut); + if (hsError != 0) { + throw HyperscanException.hsErrorToException(hsError); + } + nativeStream = streamOut.get(hs_stream_t.class); + } + return new Stream(db, nativeStream); + } + + private int scanStreamRaw(final hs_stream_t stream, final ByteBuffer input, RawMatchEventHandler eventHandler) { + if (scratch == null) { + throw new IllegalStateException("Scratch space has already been deallocated"); + } + if (activeCallback.get() != null) { + throw new IllegalStateException("Recursive scanning is not supported."); + } + activeCallback.set(eventHandler); + int hsError = 0; + try { + try (final BytePointer bytePointer = new BytePointer(input)) { + hsError = hs_scan_stream(stream, bytePointer.position(input.position()), input.remaining(), 0, scratch, matchHandler, null); + if (hsError != 0 && hsError != HS_SCAN_TERMINATED) { + throw HyperscanException.hsErrorToException(hsError); + } + } + } finally { + activeCallback.remove(); + } + return hsError; + } + + /** + * A streaming scan session created by {@link Scanner#openStream(Database)}. + * Input is fed in chunks via {@link #scan(byte[], ByteMatchEventHandler)} + * or {@link #scan(ByteBuffer, ByteMatchEventHandler)}; matches are reported + * with byte offsets relative to the start of the stream, so patterns + * spanning chunk boundaries are matched transparently. + * Not thread-safe; the stream shares the owning scanner's scratch space. + */ + public class Stream implements Closeable { + private final Database database; + private hs_stream_t nativeStream; + + private Stream(final Database database, final hs_stream_t nativeStream) { + this.database = database; + this.nativeStream = nativeStream; + } + + /** + * Feeds one chunk of input to the stream. + * + * @param input Chunk to match against, may be empty (flush only). + * @param eventHandler Handler to receive match events with byte indices. + */ + public void scan(final byte[] input, ByteMatchEventHandler eventHandler) { + ensureOpen(); + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(database.getExpression(expressionId), fromByteIdx, toByteIdx); + int length = input == null ? 0 : input.length; + ByteBuffer directBuffer = rawScanBuffer.get(); + if (directBuffer.capacity() < Math.max(length, 1)) { + directBuffer = ByteBuffer.allocateDirect(Math.max(length, 1)); + rawScanBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } + if (length > 0) { + directBuffer.put(input); + } + ((Buffer) directBuffer).flip(); + scanStreamRaw(nativeStream, directBuffer, rawHandler); + } + + /** + * Feeds one chunk of input to the stream. Bytes from the buffer's + * current position to its limit are scanned; position and limit are + * not modified. Direct buffers are scanned zero-copy, heap buffers are + * first copied into a reused per-thread direct buffer. + * + * @param input Chunk to match against. + * @param eventHandler Handler to receive match events with byte indices. + */ + public void scan(final ByteBuffer input, ByteMatchEventHandler eventHandler) { + ensureOpen(); + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(database.getExpression(expressionId), fromByteIdx, toByteIdx); + if (input.isDirect()) { + scanStreamRaw(nativeStream, input, rawHandler); + return; + } + int remaining = input.remaining(); + ByteBuffer directBuffer = rawScanBuffer.get(); + if (directBuffer.capacity() < Math.max(remaining, 1)) { + directBuffer = ByteBuffer.allocateDirect(Math.max(remaining, 1)); + rawScanBuffer.set(directBuffer); + } else { + ((Buffer) directBuffer).clear(); + } + directBuffer.put(input.duplicate()); + ((Buffer) directBuffer).flip(); + scanStreamRaw(nativeStream, directBuffer, rawHandler); + } + + /** + * Closes the stream and discards any pending matches. + */ + @Override + public void close() { + if (nativeStream != null) { + hs_stream_t stream = nativeStream; + nativeStream = null; + int hsError = hs_close_stream(stream, scratch, null, null); + if (hsError != 0 && hsError != HS_SCAN_TERMINATED) { + throw HyperscanException.hsErrorToException(hsError); + } + } + } + + /** + * Closes the stream, reporting any matches still pending at the end of + * the stream to the given handler. + * + * @param eventHandler Handler to receive trailing match events. + */ + public void close(final ByteMatchEventHandler eventHandler) { + if (nativeStream == null) { + return; + } + ensureCallbackFree(); + RawMatchEventHandler rawHandler = (expressionId, fromByteIdx, toByteIdx, expressionFlags) -> + eventHandler.onMatch(database.getExpression(expressionId), fromByteIdx, toByteIdx); + hs_stream_t stream = nativeStream; + nativeStream = null; + activeCallback.set(rawHandler); + try { + int hsError = hs_close_stream(stream, scratch, matchHandler, null); + if (hsError != 0 && hsError != HS_SCAN_TERMINATED) { + throw HyperscanException.hsErrorToException(hsError); + } + } finally { + activeCallback.remove(); + } + } + + private void ensureOpen() { + if (nativeStream == null) { + throw new IllegalStateException("Stream is already closed"); + } + } + + private void ensureCallbackFree() { + if (activeCallback.get() != null) { + throw new IllegalStateException("Recursive scanning is not supported."); + } + } + } + @Override public void close() throws IOException { if(scratch != null) { diff --git a/src/main/java/com/gliwka/hyperscan/wrapper/Utf8Encoder.java b/src/main/java/com/gliwka/hyperscan/wrapper/Utf8Encoder.java index e3ff779..34c9d5e 100644 --- a/src/main/java/com/gliwka/hyperscan/wrapper/Utf8Encoder.java +++ b/src/main/java/com/gliwka/hyperscan/wrapper/Utf8Encoder.java @@ -19,6 +19,7 @@ package com.gliwka.hyperscan.wrapper; import com.gliwka.hyperscan.wrapper.mapping.ByteCharMapping; +import com.gliwka.hyperscan.wrapper.mapping.IdentityByteCharMapping; import java.nio.Buffer; import java.nio.ByteBuffer; @@ -38,20 +39,30 @@ class Utf8Encoder { /** * Encodes a Java String to a direct ByteBuffer containing UTF-8 bytes * and creates a mapping from byte index to character index. + * Pure-ASCII input needs no table (byte index == char index), so the shared + * identity mapping is returned and nothing is allocated. For other input the + * table is created lazily when the first non-ASCII character is encountered, + * sized by the worst-case encoded length rather than the buffer capacity. * * @param buffer The ByteBuffer to write the UTF-8 bytes to * @param string The Java String to encode * @return An array of int values representing the mapping from byte index to character index */ static ByteCharMapping encodeToBufferAndMap(ByteBuffer buffer, String string) { - ByteCharMapping mapping = ByteCharMapping.create(buffer.capacity(), string.length()); + ByteCharMapping mapping = null; int writerIndex = 0; int end = string.length(); for (int i = 0; i < end; i++) { char c = string.charAt(i); + if (mapping == null && c >= UTF8_1_BYTE_LIMIT) { + mapping = createMapping(string.length(), writerIndex, end - i); + } if (c < UTF8_1_BYTE_LIMIT) { - mapping.setCharIndex(writerIndex++, i); + if (mapping != null) { + mapping.setCharIndex(writerIndex, i); + } + writerIndex++; buffer.put((byte) c); } else if (c < UTF8_2_BYTE_LIMIT) { mapping.setCharIndex(writerIndex++, i); @@ -104,6 +115,16 @@ static ByteCharMapping encodeToBufferAndMap(ByteBuffer buffer, String string) { // Make JDK9+ compile for JDK 8 ((Buffer)buffer).flip(); + return mapping != null ? mapping : IdentityByteCharMapping.instance(); + } + + private static ByteCharMapping createMapping(int stringLength, int writerIndex, int remainingChars) { + // 3 bytes per remaining char is a safe upper bound (surrogate pairs are + // 4 bytes for 2 chars). The ASCII prefix is identity (byte index == char index). + ByteCharMapping mapping = ByteCharMapping.create(writerIndex + 3 * remainingChars, stringLength); + for (int b = 0; b < writerIndex; b++) { + mapping.setCharIndex(b, b); + } return mapping; } } \ No newline at end of file diff --git a/src/main/java/com/gliwka/hyperscan/wrapper/mapping/IdentityByteCharMapping.java b/src/main/java/com/gliwka/hyperscan/wrapper/mapping/IdentityByteCharMapping.java new file mode 100644 index 0000000..78fae03 --- /dev/null +++ b/src/main/java/com/gliwka/hyperscan/wrapper/mapping/IdentityByteCharMapping.java @@ -0,0 +1,32 @@ +package com.gliwka.hyperscan.wrapper.mapping; + +/** + * Shared identity mapping for pure-ASCII input, where byte index == char index. + * Allocates nothing; {@link #getMappingSize()} returns 0 to indicate that no + * table exists. + */ +public final class IdentityByteCharMapping implements ByteCharMapping { + private static final IdentityByteCharMapping INSTANCE = new IdentityByteCharMapping(); + + private IdentityByteCharMapping() { + } + + public static IdentityByteCharMapping instance() { + return INSTANCE; + } + + @Override + public void setCharIndex(int byteIndex, int charIndex) { + throw new UnsupportedOperationException("Identity mapping is immutable"); + } + + @Override + public int getCharIndex(int byteIndex) { + return byteIndex; + } + + @Override + public int getMappingSize() { + return 0; + } +} diff --git a/src/test/java/com/gliwka/hyperscan/benchmark/SimpleBenchmark.java b/src/test/java/com/gliwka/hyperscan/benchmark/SimpleBenchmark.java index fd4c5c5..8280e75 100644 --- a/src/test/java/com/gliwka/hyperscan/benchmark/SimpleBenchmark.java +++ b/src/test/java/com/gliwka/hyperscan/benchmark/SimpleBenchmark.java @@ -1,6 +1,7 @@ package com.gliwka.hyperscan.benchmark; import com.gliwka.hyperscan.wrapper.*; +import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.*; import org.openjdk.jmh.infra.Blackhole; diff --git a/src/test/java/com/gliwka/hyperscan/wrapper/ScannerTest.java b/src/test/java/com/gliwka/hyperscan/wrapper/ScannerTest.java index 69eef10..2cd0bff 100644 --- a/src/test/java/com/gliwka/hyperscan/wrapper/ScannerTest.java +++ b/src/test/java/com/gliwka/hyperscan/wrapper/ScannerTest.java @@ -194,6 +194,50 @@ void scanBytesWithHandler_utf8_shouldFindMatch() throws CompileErrorException { utfDb.close(); } + @Test + void scanDirectByteBufferWithHandler_shouldInvokeHandler() { + byte[] data = "test test1 test test3".getBytes(StandardCharsets.UTF_8); + java.nio.ByteBuffer buffer = java.nio.ByteBuffer.allocateDirect(data.length); + buffer.put(data); + ((java.nio.Buffer) buffer).flip(); + + AtomicInteger matchCount2 = new AtomicInteger(0); + final List matchedExprs = new ArrayList<>(); + + scanner.scan(database, buffer, (expression, from, to) -> { + matchCount2.incrementAndGet(); + matchedExprs.add(expression.getExpression()); + return true; + }); + + assertThat(matchCount2.get()).isEqualTo(6); + assertThat(matchedExprs).containsExactlyInAnyOrder("test", "test", "test1", "test", "test", "test3"); + assertThat(buffer.position()).isEqualTo(0); + assertThat(buffer.limit()).isEqualTo(data.length); + } + + @Test + void scanHeapByteBufferWithHandler_shouldInvokeHandler() { + java.nio.ByteBuffer buffer = java.nio.ByteBuffer.wrap("xx test yy".getBytes(StandardCharsets.UTF_8)); + ((java.nio.Buffer) buffer).position(3); + ((java.nio.Buffer) buffer).limit(7); + + AtomicInteger matchCount2 = new AtomicInteger(0); + final List matchPositions2 = new ArrayList<>(); + + scanner.scan(database, buffer, (expression, from, to) -> { + matchCount2.incrementAndGet(); + matchPositions2.add(from); + matchPositions2.add(to); + return true; + }); + + assertThat(matchCount2.get()).isEqualTo(1); + assertThat(matchPositions2).containsExactly(0L, 4L); + assertThat(buffer.position()).isEqualTo(3); + assertThat(buffer.limit()).isEqualTo(7); + } + @ParameterizedTest @ValueSource(strings = {"Test", "A test string", "Another TEST"}) void hasMatch_shouldReturnTrueIfMatchExists(String input) { diff --git a/src/test/java/com/gliwka/hyperscan/wrapper/StreamTest.java b/src/test/java/com/gliwka/hyperscan/wrapper/StreamTest.java new file mode 100644 index 0000000..0935e93 --- /dev/null +++ b/src/test/java/com/gliwka/hyperscan/wrapper/StreamTest.java @@ -0,0 +1,156 @@ +package com.gliwka.hyperscan.wrapper; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.*; + +class StreamTest { + + private Scanner scanner; + private Database streamDb; + private Database vectorDb; + + @BeforeEach + void setUp() { + scanner = new Scanner(); + try { + streamDb = Database.compile(Arrays.asList( + new Expression("test", 0), + new Expression("test1", 1)), Mode.STREAM); + scanner.allocScratch(streamDb); + vectorDb = Database.compile(Arrays.asList( + new Expression("hello", 0), + new Expression("world", 1)), Mode.VECTORED); + scanner.allocScratch(vectorDb); + } catch (CompileErrorException e) { + fail("Database compilation failed", e); + } + } + + @AfterEach + void tearDown() { + try { + if (streamDb != null) { + streamDb.close(); + } + if (vectorDb != null) { + vectorDb.close(); + } + if (scanner != null) { + scanner.close(); + } + } catch (Exception e) { + // Ignore cleanup exceptions in tests + } + } + + @Test + void streamingScanAcrossChunkBoundaryFindsMatches() { + List matches = new ArrayList<>(); + try (Scanner.Stream stream = scanner.openStream(streamDb)) { + ByteMatchEventHandler handler = (expression, from, to) -> { + matches.add(new long[]{expression.getId(), from, to}); + return true; + }; + stream.scan("tes".getBytes(StandardCharsets.UTF_8), handler); + stream.scan("t1".getBytes(StandardCharsets.UTF_8), handler); + } + + assertThat(matches).hasSize(2); + assertThat(matches).anySatisfy(m -> assertThat(m).containsExactly(0L, 0L, 4L)); + assertThat(matches).anySatisfy(m -> assertThat(m).containsExactly(1L, 0L, 5L)); + } + + @Test + void streamCloseWithHandlerReportsPendingMatch() throws CompileErrorException { + Database db = Database.compile(new Expression("c$", 0), Mode.STREAM); + try { + scanner.allocScratch(db); + List matches = new ArrayList<>(); + Scanner.Stream stream = scanner.openStream(db); + stream.scan("abc".getBytes(StandardCharsets.UTF_8), (expression, from, to) -> { + matches.add(new long[]{expression.getId(), from, to}); + return true; + }); + assertThat(matches).isEmpty(); + stream.close((expression, from, to) -> { + matches.add(new long[]{expression.getId(), from, to}); + return true; + }); + + assertThat(matches).hasSize(1); + assertThat(matches.get(0)).containsExactly(0L, 0L, 3L); + } finally { + db.close(); + } + } + + @Test + void streamScanAfterCloseThrows() { + Scanner.Stream stream = scanner.openStream(streamDb); + stream.close(); + assertThrows(IllegalStateException.class, + () -> stream.scan("x".getBytes(StandardCharsets.UTF_8), (expression, from, to) -> true)); + } + + @Test + void vectoredScanFindsMatchesAcrossSegments() { + List matches = new ArrayList<>(); + scanner.scanVector(vectorDb, + new byte[][]{"hello ".getBytes(StandardCharsets.UTF_8), "world!".getBytes(StandardCharsets.UTF_8)}, + (expression, from, to) -> { + matches.add(new long[]{expression.getId(), from, to}); + return true; + }); + + assertThat(matches).hasSize(2); + assertThat(matches.get(0)).containsExactly(0L, 0L, 5L); + assertThat(matches.get(1)).containsExactly(1L, 0L, 11L); + } + + @Test + void vectoredScanSupportsDirectAndHeapBuffers() { + ByteBuffer direct = ByteBuffer.allocateDirect(6); + direct.put("hello ".getBytes(StandardCharsets.UTF_8)); + ((java.nio.Buffer) direct).flip(); + ByteBuffer heap = ByteBuffer.wrap("world!".getBytes(StandardCharsets.UTF_8)); + + List matches = new ArrayList<>(); + scanner.scanVector(vectorDb, new ByteBuffer[]{direct, heap}, (expression, from, to) -> { + matches.add(new long[]{expression.getId(), from, to}); + return true; + }); + + assertThat(matches).hasSize(2); + assertThat(matches.get(0)).containsExactly(0L, 0L, 5L); + assertThat(matches.get(1)).containsExactly(1L, 0L, 11L); + assertThat(direct.position()).isEqualTo(0); + assertThat(direct.limit()).isEqualTo(6); + } + + @Test + void openStreamRejectsNonStreamingDatabase() throws CompileErrorException { + Database blockDb = Database.compile(new Expression("test", 0)); + try { + assertThrows(IllegalArgumentException.class, () -> scanner.openStream(blockDb)); + } finally { + blockDb.close(); + } + } + + @Test + void scanVectorRejectsNonVectoredDatabase() { + assertThrows(IllegalArgumentException.class, + () -> scanner.scanVector(streamDb, new byte[][]{"x".getBytes(StandardCharsets.UTF_8)}, + (expression, from, to) -> true)); + } +} diff --git a/src/test/java/com/gliwka/hyperscan/wrapper/Utf8EncoderTest.java b/src/test/java/com/gliwka/hyperscan/wrapper/Utf8EncoderTest.java index 1d32789..e2fe8c6 100644 --- a/src/test/java/com/gliwka/hyperscan/wrapper/Utf8EncoderTest.java +++ b/src/test/java/com/gliwka/hyperscan/wrapper/Utf8EncoderTest.java @@ -9,6 +9,37 @@ public class Utf8EncoderTest { + @Test + public void testAsciiEncodingUsesIdentityMapping() { + String input = "plain ascii text"; + ByteBuffer buffer = ByteBuffer.allocate(100); + + ByteCharMapping mapping = Utf8Encoder.encodeToBufferAndMap(buffer, input); + + assertEquals(0, mapping.getMappingSize()); + assertEquals(input.length(), buffer.limit()); + for (int i = 0; i < input.length(); i++) { + assertEquals(i, mapping.getCharIndex(i)); + } + } + + @Test + public void testMappingAllocatedOnFirstNonAscii() { + String input = "ab世cd"; + ByteBuffer buffer = ByteBuffer.allocate(100); + + ByteCharMapping mapping = Utf8Encoder.encodeToBufferAndMap(buffer, input); + + assertEquals(7, buffer.limit()); + assertEquals(0, mapping.getCharIndex(0)); + assertEquals(1, mapping.getCharIndex(1)); + assertEquals(2, mapping.getCharIndex(2)); + assertEquals(2, mapping.getCharIndex(3)); + assertEquals(2, mapping.getCharIndex(4)); + assertEquals(3, mapping.getCharIndex(5)); + assertEquals(4, mapping.getCharIndex(6)); + } + @Test public void testAsciiEncoding() { String input = "Hello, world!";