Compare commits

...

6 Commits

Author SHA1 Message Date
Andrea Cavalli
31243c269a Update migrate 2024-03-31 01:54:44 +01:00
Andrea Cavalli
caf3f6888e Improve logging 2024-03-31 01:54:34 +01:00
Andrea Cavalli
ecdb0b05b8 migrate 2024-03-30 23:14:05 +01:00
Andrea Cavalli
7a8e3d6158 Async migrate 2024-03-30 23:14:05 +01:00
Andrea Cavalli
9729704bb9 Restore columnId without getting it first 2024-03-30 23:14:05 +01:00
Andrea Cavalli
bb9e9ad33c Migration 2024-03-30 23:14:05 +01:00
19 changed files with 460 additions and 121 deletions

8
debug-logging.properties Normal file
View File

@ -0,0 +1,8 @@
handlers= java.util.logging.ConsoleHandler
.level= INFO
java.util.logging.ConsoleHandler.level = ALL
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter
java.util.logging.SimpleFormatter.format= [%1$tF %1$tT] [%4$-7s] %5$s %n
# your specific logger level
db.requests.level = ALL

View File

@ -34,6 +34,11 @@
</repositories>
<dependencies>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>bson</artifactId>
<version>5.0.0</version>
</dependency>
<dependency>
<groupId>org.rocksdb</groupId>
<artifactId>rocksdbjni</artifactId>

View File

@ -0,0 +1,148 @@
package it.cavallium.rockserver.core;
import it.cavallium.rockserver.core.common.api.ColumnSchema;
import it.cavallium.rockserver.core.common.api.RocksDB;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HexFormat;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.layered.TFramedTransport;
import org.bson.BSONObject;
import org.bson.BasicBSONDecoder;
import org.bson.BasicBSONEncoder;
import org.bson.BasicBSONObject;
import org.bson.Document;
public class Migrate {
public static void main(String[] args) throws TException, IOException, InterruptedException {
// Tunables
var columnName = args[0];
var password = args[1];
var hostname = args[2];
//
System.out.println("Column: " + columnName);
var temp = Files.createTempFile("temp-out-" + columnName + "-", ".json");
boolean peerMode = columnName.startsWith("peers_");
var columnSchema = peerMode ? new ColumnSchema(List.of(Long.BYTES), List.of(), true) : new ColumnSchema(List.of(Byte.BYTES), List.of(), true);
var result = Runtime
.getRuntime()
.exec(new String[]{"psql", "--host", "home.cavallium.it", "-d", "sessions", "-U", "postgres", "-c",
"SELECT json_agg(t) FROM (SELECT * FROM \"" + columnName + "\") t", "-qAtX",
"--output=" + temp}, new String[] {"PGPASSWORD=" + password});
result.waitFor();
var jo = Document.parse("{\"data\": " + Files.readString(temp) + "}");
Files.delete(temp);
List<Document> documents = (List<Document>) jo.get("data");
jo = null;
System.gc();
System.out.println("Read json file");
var it = documents.iterator();
List<Map.Entry<Object, BasicBSONObject>> documentMap = new ArrayList<>();
while (it.hasNext()) {
var document = it.next();
var obj = new BasicBSONObject();
if (peerMode) {
var id = ((Number) document.get("id")).longValue();
var accessHash = ((Number) document.get("access_hash")).longValue();
var peerType = ((String) document.get("type"));
var username = ((String) document.get("username"));
var phoneNumber = ((String) document.get("phone_number"));
var lastUpdateOn = ((Number) document.get("last_update_on")).longValue();
obj.put("access_hash", accessHash);
obj.put("peer_type", peerType);
obj.put("username", username);
obj.put("phone_number", phoneNumber);
obj.put("last_update_on", lastUpdateOn);
if (!peerType.equals("user")) {
documentMap.add(Map.entry(id, obj));
}
} else {
byte id = 0;
Long dcId = ((Number) document.get("dc_id")).longValue();
Long apiId = document.get("api_id") != null ? ((Number) document.get("api_id")).longValue() : null;
var testMode = ((Boolean) document.get("test_mode"));
var authKey = HexFormat.of().parseHex(((String) document.get("auth_key")).substring(2));
var date = ((Number) document.get("date")).longValue();
var userId = ((Number) document.get("user_id")).longValue();
var isBot = ((Boolean) document.get("is_bot"));
var phone = ((String) document.get("phone"));
obj.put("dc_id", dcId);
obj.put("api_id", apiId);
obj.put("test_mode", testMode);
obj.put("auth_key", authKey);
obj.put("date", date);
obj.put("user_id", userId);
obj.put("is_bot", isBot);
obj.put("phone", phone);
documentMap.add(Map.entry(id, obj));
}
}
documents = null;
System.gc();
System.out.println("parsed documents");
var protocol = new TBinaryProtocol.Factory();
var clients = ThreadLocal.withInitial(() -> {
try {
var socket = new TSocket(hostname, 5332);
var transport = new TFramedTransport(socket);
transport.open();
return new RocksDB.Client(new TBinaryProtocol(transport));
} catch (TTransportException e) {
throw new RuntimeException(e);
}
});
var columnId = clients.get().createColumn(columnName, columnSchema);
var encoder = ThreadLocal.withInitial(BasicBSONEncoder::new);
var keyBBLocal = ThreadLocal.withInitial(() -> peerMode ? ByteBuffer.allocate(Long.BYTES).order(ByteOrder.BIG_ENDIAN) : ByteBuffer.allocate(1).order(ByteOrder.BIG_ENDIAN));
AtomicLong next = new AtomicLong();
long total = documentMap.size();
var initTime = Instant.now();
documentMap.stream().parallel().forEach(longDocumentEntry -> {
ByteBuffer bb = keyBBLocal.get();
if (peerMode) {
bb.asLongBuffer().put((Long) longDocumentEntry.getKey());
} else {
bb.put((Byte) longDocumentEntry.getKey()).flip();
}
var valueArray = encoder.get().encode(longDocumentEntry.getValue());
var valueBuf = ByteBuffer.wrap(valueArray);
try {
clients.get().putFast(0, columnId, List.of(bb), valueBuf);
} catch (TException e) {
throw new RuntimeException(e);
}
var nn = next.incrementAndGet();
if (nn > 0 && nn % 10_000 == 0) {
var endTime = Instant.now();
var dur = Duration.between(initTime, endTime);
System.out.printf("Written %d/%d elements... %.2f, speed: %.2fHz%n", nn, total, ((nn * 100d) / total), nn / (dur.toMillis() / 1000d));
}
});
if (!peerMode) {
System.out.println("Schema: " + new BasicBSONDecoder().readObject(clients.get().get(0, columnId, List.of(ByteBuffer.wrap(new byte[] {0}))).getValue()));
}
var endTime = Instant.now();
var dur = Duration.between(initTime, endTime);
if (total > 0) System.out.printf("Took %s, speed: %.2fHz%n", dur, total / (dur.toMillis() / 1000d));
}
}

View File

@ -1,5 +1,6 @@
package it.cavallium.rockserver.core.client;
import it.cavallium.rockserver.core.common.Keys;
import it.cavallium.rockserver.core.common.RequestType;
import it.cavallium.rockserver.core.common.RequestType.RequestGet;
import it.cavallium.rockserver.core.common.RequestType.RequestPut;
@ -102,7 +103,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
public <T> T put(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
@NotNull Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return db.put(arena, transactionOrUpdateId, columnId, keys, value, requestType);
@ -112,7 +113,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
public <T> List<T> putMulti(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull List<@NotNull MemorySegment @NotNull []> keys,
@NotNull List<Keys> keys,
@NotNull List<@NotNull MemorySegment> values,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return db.putMulti(arena, transactionOrUpdateId, columnId, keys, values, requestType);
@ -122,7 +123,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
public <T> T get(Arena arena,
long transactionOrUpdateId,
long columnId,
MemorySegment @NotNull [] keys,
Keys keys,
RequestGet<? super MemorySegment, T> requestType) throws RocksDBException {
return db.get(arena, transactionOrUpdateId, columnId, keys, requestType);
}
@ -131,8 +132,8 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
public long openIterator(Arena arena,
long transactionId,
long columnId,
@NotNull MemorySegment @NotNull [] startKeysInclusive,
@Nullable MemorySegment[] endKeysExclusive,
@NotNull Keys startKeysInclusive,
@Nullable Keys endKeysExclusive,
boolean reverse,
long timeoutMs) throws RocksDBException {
return db.openIterator(arena, transactionId, columnId, startKeysInclusive, endKeysExclusive, reverse, timeoutMs);
@ -144,7 +145,7 @@ public class EmbeddedConnection extends BaseConnection implements RocksDBAPI {
}
@Override
public void seekTo(Arena arena, long iterationId, MemorySegment @NotNull [] keys) throws RocksDBException {
public void seekTo(Arena arena, long iterationId, Keys keys) throws RocksDBException {
db.seekTo(arena, iterationId, keys);
}

View File

@ -0,0 +1,88 @@
package it.cavallium.rockserver.core.client;
import it.cavallium.rockserver.core.common.RocksDBAPICommand;
import it.cavallium.rockserver.core.common.RocksDBAsyncAPI;
import it.cavallium.rockserver.core.common.RocksDBSyncAPI;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.CompletionStage;
import java.util.function.Supplier;
import java.util.logging.Level;
import java.util.logging.Logger;
public class LoggingClient implements RocksDBConnection {
private final RocksDBConnection client;
private final LoggingSyncApi syncApi;
private final LoggingAsyncApi asyncApi;
private final Logger logger;
public LoggingClient(RocksDBConnection client) {
this.client = client;
this.syncApi = new LoggingSyncApi(client.getSyncApi());
this.asyncApi = new LoggingAsyncApi(client.getAsyncApi());
this.logger = Logger.getLogger("db.requests");
}
@Override
public URI getUrl() {
return client.getUrl();
}
@Override
public RocksDBSyncAPI getSyncApi() {
return syncApi;
}
@Override
public RocksDBAsyncAPI getAsyncApi() {
return asyncApi;
}
@Override
public void close() throws IOException {
client.close();
}
private class LoggingSyncApi implements RocksDBSyncAPI {
private final RocksDBSyncAPI syncApi;
public LoggingSyncApi(RocksDBSyncAPI syncApi) {
this.syncApi = syncApi;
}
@Override
public <R> R requestSync(RocksDBAPICommand<R> req) {
R result;
try {
result = syncApi.requestSync(req);
} catch (Throwable e) {
logger.log(Level.FINEST, "Request failed: {0} Error: {1}", new Object[] {req, e.getMessage()});
throw e;
}
logger.log(Level.FINEST, "Request executed: {0} Result: {1}", new Object[] {req, result});
return result;
}
}
private class LoggingAsyncApi implements RocksDBAsyncAPI {
private final RocksDBAsyncAPI asyncApi;
public LoggingAsyncApi(RocksDBAsyncAPI asyncApi) {
this.asyncApi = asyncApi;
}
@Override
public <R> CompletionStage<R> requestAsync(RocksDBAPICommand<R> req) {
return asyncApi.requestAsync(req).whenComplete((result, e) -> {
if (e != null) {
logger.log(Level.FINEST, "Request failed: {0} Error: {1}", new Object[] {req, e.getMessage()});
} else {
logger.log(Level.FINEST, "Request executed: {0} Result: {1}", new Object[] {req, result});
}
});
}
}
}

View File

@ -0,0 +1,31 @@
package it.cavallium.rockserver.core.common;
import java.lang.foreign.MemorySegment;
import java.util.Arrays;
import java.util.stream.Collectors;
import org.jetbrains.annotations.NotNull;
public record Keys(@NotNull MemorySegment @NotNull [] keys) {
@Override
public String toString() {
return Arrays.stream(keys).map(Utils::toPrettyString).collect(Collectors.joining(";", "[", "]"));
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
Keys keys1 = (Keys) o;
return Arrays.equals(keys, keys1.keys);
}
@Override
public int hashCode() {
return Arrays.hashCode(keys);
}
}

View File

@ -5,6 +5,7 @@ import it.cavallium.rockserver.core.common.RequestType.RequestPut;
import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.util.List;
import java.util.StringJoiner;
import java.util.concurrent.CompletionStage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@ -138,7 +139,7 @@ public sealed interface RocksDBAPICommand<R> {
record Put<T>(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, T> requestType) implements RocksDBAPICommand<T> {
@ -152,6 +153,18 @@ public sealed interface RocksDBAPICommand<R> {
return api.putAsync(arena, transactionOrUpdateId, columnId, keys, value, requestType);
}
@Override
public String toString() {
var sb = new StringBuilder("PUT");
if (transactionOrUpdateId != 0) {
sb.append(" tx:").append(transactionOrUpdateId);
}
sb.append(" column:").append(columnId);
sb.append(" keys:").append(keys);
sb.append(" value:").append(Utils.toPrettyString(value));
sb.append(" expected:").append(requestType.getRequestTypeId());
return sb.toString();
}
}
/**
* Put multiple elements into the specified positions
@ -163,7 +176,7 @@ public sealed interface RocksDBAPICommand<R> {
* @param requestType the request type determines which type of data will be returned.
*/
record PutMulti<T>(Arena arena, long transactionOrUpdateId, long columnId,
@NotNull List<@NotNull MemorySegment @NotNull []> keys,
@NotNull List<Keys> keys,
@NotNull List<@NotNull MemorySegment> values,
RequestPut<? super MemorySegment, T> requestType) implements RocksDBAPICommand<List<T>> {
@ -177,6 +190,23 @@ public sealed interface RocksDBAPICommand<R> {
return api.putMultiAsync(arena, transactionOrUpdateId, columnId, keys, values, requestType);
}
@Override
public String toString() {
var sb = new StringBuilder("PUT_MULTI");
if (transactionOrUpdateId != 0) {
sb.append(" tx:").append(transactionOrUpdateId);
}
sb.append(" column:").append(columnId);
sb.append(" expected:").append(requestType.getRequestTypeId());
sb.append(" multi:[");
for (int i = 0; i < keys.size(); i++) {
if (i > 0) sb.append(",");
sb.append(" keys:").append(keys.get(i));
sb.append(" value:").append(Utils.toPrettyString(values.get(i)));
}
sb.append("]");
return sb.toString();
}
}
/**
* Get an element from the specified position
@ -189,7 +219,7 @@ public sealed interface RocksDBAPICommand<R> {
record Get<T>(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
Keys keys,
RequestGet<? super MemorySegment, T> requestType) implements RocksDBAPICommand<T> {
@Override
@ -202,6 +232,17 @@ public sealed interface RocksDBAPICommand<R> {
return api.getAsync(arena, transactionOrUpdateId, columnId, keys, requestType);
}
@Override
public String toString() {
var sb = new StringBuilder("GET");
if (transactionOrUpdateId != 0) {
sb.append(" tx:").append(transactionOrUpdateId);
}
sb.append(" column:").append(columnId);
sb.append(" keys:").append(keys);
sb.append(" expected:").append(requestType.getRequestTypeId());
return sb.toString();
}
}
/**
* Open an iterator
@ -217,8 +258,8 @@ public sealed interface RocksDBAPICommand<R> {
record OpenIterator(Arena arena,
long transactionId,
long columnId,
@NotNull MemorySegment @NotNull [] startKeysInclusive,
@NotNull MemorySegment @Nullable [] endKeysExclusive,
Keys startKeysInclusive,
@Nullable Keys endKeysExclusive,
boolean reverse,
long timeoutMs) implements RocksDBAPICommand<Long> {
@ -264,7 +305,7 @@ public sealed interface RocksDBAPICommand<R> {
* @param iterationId iteration id
* @param keys keys, inclusive. [] means "the beginning"
*/
record SeekTo(Arena arena, long iterationId, @NotNull MemorySegment @NotNull [] keys) implements
record SeekTo(Arena arena, long iterationId, Keys keys) implements
RocksDBAPICommand<Void> {
@Override

View File

@ -18,7 +18,6 @@ import it.cavallium.rockserver.core.common.RocksDBAPICommand.Subsequent;
import java.lang.foreign.Arena;
import java.lang.foreign.MemorySegment;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@ -59,7 +58,7 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
default <T> CompletionStage<T> putAsync(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
@NotNull Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return requestAsync(new Put<>(arena, transactionOrUpdateId, columnId, keys, value, requestType));
@ -69,7 +68,7 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
default <T> CompletionStage<List<T>> putMultiAsync(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull List<@NotNull MemorySegment @NotNull []> keys,
@NotNull List<@NotNull Keys> keys,
@NotNull List<@NotNull MemorySegment> values,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return requestAsync(new PutMulti<>(arena, transactionOrUpdateId, columnId, keys, values, requestType));
@ -79,7 +78,7 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
default <T> CompletionStage<T> getAsync(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
@NotNull Keys keys,
RequestGet<? super MemorySegment, T> requestType) throws RocksDBException {
return requestAsync(new Get<>(arena, transactionOrUpdateId, columnId, keys, requestType));
}
@ -88,8 +87,8 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
default CompletionStage<Long> openIteratorAsync(Arena arena,
long transactionId,
long columnId,
@NotNull MemorySegment @NotNull [] startKeysInclusive,
@NotNull MemorySegment @Nullable [] endKeysExclusive,
@NotNull Keys startKeysInclusive,
@Nullable Keys endKeysExclusive,
boolean reverse,
long timeoutMs) throws RocksDBException {
return requestAsync(new OpenIterator(arena,
@ -108,7 +107,7 @@ public interface RocksDBAsyncAPI extends RocksDBAsyncAPIRequestHandler {
}
/** See: {@link SeekTo}. */
default CompletionStage<Void> seekToAsync(Arena arena, long iterationId, @NotNull MemorySegment @NotNull [] keys) throws RocksDBException {
default CompletionStage<Void> seekToAsync(Arena arena, long iterationId, @NotNull Keys keys) throws RocksDBException {
return requestAsync(new SeekTo(arena, iterationId, keys));
}

View File

@ -57,7 +57,7 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
default <T> T put(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return requestSync(new Put<>(arena, transactionOrUpdateId, columnId, keys, value, requestType));
@ -67,7 +67,7 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
default <T> List<T> putMulti(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull List<@NotNull MemorySegment @NotNull []> keys,
@NotNull List<Keys> keys,
@NotNull List<@NotNull MemorySegment> values,
RequestPut<? super MemorySegment, T> requestType) throws RocksDBException {
return requestSync(new PutMulti<>(arena, transactionOrUpdateId, columnId, keys, values, requestType));
@ -77,7 +77,7 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
default <T> T get(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
Keys keys,
RequestGet<? super MemorySegment, T> requestType) throws RocksDBException {
return requestSync(new Get<>(arena, transactionOrUpdateId, columnId, keys, requestType));
}
@ -86,8 +86,8 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
default long openIterator(Arena arena,
long transactionId,
long columnId,
@NotNull MemorySegment @NotNull [] startKeysInclusive,
@NotNull MemorySegment @Nullable [] endKeysExclusive,
Keys startKeysInclusive,
@Nullable Keys endKeysExclusive,
boolean reverse,
long timeoutMs) throws RocksDBException {
return requestSync(new OpenIterator(arena, transactionId, columnId, startKeysInclusive, endKeysExclusive, reverse, timeoutMs));
@ -99,7 +99,7 @@ public interface RocksDBSyncAPI extends RocksDBSyncAPIRequestHandler {
}
/** See: {@link SeekTo}. */
default void seekTo(Arena arena, long iterationId, @NotNull MemorySegment @NotNull [] keys) throws RocksDBException {
default void seekTo(Arena arena, long iterationId, Keys keys) throws RocksDBException {
requestSync(new SeekTo(arena, iterationId, keys));
}

View File

@ -15,6 +15,7 @@ import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HexFormat;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Stream;
@ -160,4 +161,9 @@ public class Utils {
return port;
}
}
public static String toPrettyString(MemorySegment s) {
var b = s.toArray(BIG_ENDIAN_BYTES);
return HexFormat.of().formatHex(b);
}
}

View File

@ -6,9 +6,9 @@ import static org.rocksdb.KeyMayExist.KeyMayExistEnum.kExistsWithValue;
import static org.rocksdb.KeyMayExist.KeyMayExistEnum.kExistsWithoutValue;
import it.cavallium.rockserver.core.common.ColumnHashType;
import it.cavallium.rockserver.core.common.Keys;
import it.cavallium.rockserver.core.common.RequestType;
import it.cavallium.rockserver.core.common.RequestType.RequestGet;
import it.cavallium.rockserver.core.common.RequestType.RequestNothing;
import it.cavallium.rockserver.core.common.RequestType.RequestPut;
import it.cavallium.rockserver.core.common.ColumnSchema;
import it.cavallium.rockserver.core.common.Delta;
@ -392,7 +392,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
public <T> T put(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull MemorySegment @NotNull [] keys,
@NotNull Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, T> requestType) throws it.cavallium.rockserver.core.common.RocksDBException {
ops.beginOp();
@ -420,7 +420,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
public <T> List<T> putMulti(Arena arena,
long transactionOrUpdateId,
long columnId,
@NotNull List<@NotNull MemorySegment @NotNull []> keys,
@NotNull List<Keys> keys,
@NotNull List<@NotNull MemorySegment> values,
RequestPut<? super MemorySegment, T> requestType) throws it.cavallium.rockserver.core.common.RocksDBException {
if (keys.size() != values.size()) {
@ -480,7 +480,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
@Nullable Tx optionalTxOrUpdate,
ColumnInstance col,
long updateId,
@NotNull MemorySegment @NotNull[] keys,
@NotNull Keys keys,
@NotNull MemorySegment value,
RequestPut<? super MemorySegment, U> callback) throws it.cavallium.rockserver.core.common.RocksDBException {
// Check for null value
@ -501,14 +501,14 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
}
return wrapWithTransactionIfNeeded(optionalTxOrUpdate, needsTx, tx -> {
MemorySegment previousValue;
MemorySegment calculatedKey = col.calculateKey(arena, keys);
MemorySegment calculatedKey = col.calculateKey(arena, keys.keys());
if (updateId != 0L) {
assert tx != null;
tx.val().setSavePoint();
}
if (col.hasBuckets()) {
assert tx != null;
var bucketElementKeys = col.getBucketElementKeys(keys);
var bucketElementKeys = col.getBucketElementKeys(keys.keys());
try (var readOptions = new ReadOptions()) {
var previousRawBucketByteArray = tx.val().getForUpdate(readOptions, col.cfh(), calculatedKey.toArray(BIG_ENDIAN_BYTES), true);
MemorySegment previousRawBucket = toMemorySegment(arena, previousRawBucketByteArray);
@ -589,7 +589,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
public <T> T get(Arena arena,
long transactionOrUpdateId,
long columnId,
MemorySegment @NotNull [] keys,
Keys keys,
RequestGet<? super MemorySegment, T> requestType) throws it.cavallium.rockserver.core.common.RocksDBException {
// Column id
var col = getColumn(columnId);
@ -620,7 +620,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
Tx tx,
long updateId,
ColumnInstance col,
MemorySegment @NotNull [] keys,
Keys keys,
RequestGet<? super MemorySegment, T> callback) throws it.cavallium.rockserver.core.common.RocksDBException {
ops.beginOp();
try {
@ -631,9 +631,9 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
MemorySegment foundValue;
boolean existsValue;
MemorySegment calculatedKey = col.calculateKey(arena, keys);
MemorySegment calculatedKey = col.calculateKey(arena, keys.keys());
if (col.hasBuckets()) {
var bucketElementKeys = col.getBucketElementKeys(keys);
var bucketElementKeys = col.getBucketElementKeys(keys.keys());
try (var readOptions = new ReadOptions()) {
MemorySegment previousRawBucket = dbGet(tx, col, arena, readOptions, calculatedKey);
if (previousRawBucket != null) {
@ -689,8 +689,8 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
public long openIterator(Arena arena,
long transactionId,
long columnId,
MemorySegment @NotNull [] startKeysInclusive,
@Nullable MemorySegment[] endKeysExclusive,
Keys startKeysInclusive,
@Nullable Keys endKeysExclusive,
boolean reverse,
long timeoutMs) throws it.cavallium.rockserver.core.common.RocksDBException {
// Open an operation that ends when the iterator is closed
@ -725,7 +725,7 @@ public class EmbeddedDB implements RocksDBSyncAPI, Closeable {
}
@Override
public void seekTo(Arena arena, long iterationId, @NotNull MemorySegment @NotNull [] keys)
public void seekTo(Arena arena, long iterationId, @NotNull Keys keys)
throws it.cavallium.rockserver.core.common.RocksDBException {
ops.beginOp();
try {

View File

@ -1,5 +1,6 @@
package it.cavallium.rockserver.core.server;
import it.cavallium.rockserver.core.client.LoggingClient;
import it.cavallium.rockserver.core.client.RocksDBConnection;
import java.io.Closeable;
import java.io.IOException;
@ -9,7 +10,11 @@ public class Server implements Closeable {
private final RocksDBConnection client;
public Server(RocksDBConnection client) {
this.client = client;
this.client = new LoggingClient(client);
}
public RocksDBConnection getClient() {
return client;
}
@Override

View File

@ -1,6 +1,7 @@
package it.cavallium.rockserver.core.server;
import it.cavallium.rockserver.core.client.RocksDBConnection;
import it.cavallium.rockserver.core.common.Keys;
import it.cavallium.rockserver.core.common.RequestType;
import it.cavallium.rockserver.core.common.UpdateContext;
import it.cavallium.rockserver.core.common.api.ColumnHashType;
@ -32,7 +33,7 @@ public class ThriftServer extends Server {
public ThriftServer(RocksDBConnection client, String http2Host, int http2Port) throws IOException {
super(client);
var handler = new ThriftHandler(client);
var handler = new ThriftHandler(this.getClient());
try {
var serverTransport = new TNonblockingServerSocket(new InetSocketAddress(http2Host, http2Port));
@ -46,11 +47,11 @@ public class ThriftServer extends Server {
}
}
private static @NotNull List<@NotNull MemorySegment[]> keysToRecords(Arena arena, @NotNull List<@NotNull List< @NotNull ByteBuffer>> keysMulti) {
private static @NotNull List<@NotNull Keys> keysToRecords(Arena arena, @NotNull List<@NotNull List< @NotNull ByteBuffer>> keysMulti) {
return keysMulti.stream().map(keys -> keysToRecord(arena, keys)).toList();
}
private static MemorySegment[] keysToRecord(Arena arena, List<@NotNull ByteBuffer> keys) {
private static Keys keysToRecord(Arena arena, List<@NotNull ByteBuffer> keys) {
if (keys == null) {
return null;
}
@ -60,7 +61,7 @@ public class ThriftServer extends Server {
result[i] = keyToRecord(arena, key);
i++;
}
return result;
return new Keys(result);
}
private static @NotNull List<@NotNull MemorySegment> keyToRecords(Arena arena, @NotNull List<@NotNull ByteBuffer> keyMulti) {

View File

@ -9,6 +9,7 @@ module rockserver.core {
requires it.unimi.dsi.fastutil;
requires org.apache.thrift;
requires org.slf4j;
requires org.mongodb.bson;
exports it.cavallium.rockserver.core.client;
exports it.cavallium.rockserver.core.common;

View File

@ -3,54 +3,55 @@ package it.cavallium.rockserver.core.impl.test;
import static it.cavallium.rockserver.core.common.Utils.toMemorySegmentSimple;
import it.cavallium.rockserver.core.common.ColumnHashType;
import it.cavallium.rockserver.core.common.Keys;
import it.unimi.dsi.fastutil.objects.ObjectList;
import java.lang.foreign.MemorySegment;
public class EmbeddedDBFixedWithValueTest extends EmbeddedDBTest {
@Override
protected MemorySegment[] getKeyI(int i) {
return new MemorySegment[] {
protected Keys getKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 7),
toMemorySegmentSimple(arena, i)
};
});
}
@Override
protected MemorySegment[] getNotFoundKeyI(int i) {
return new MemorySegment[] {
protected Keys getNotFoundKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 5, 6),
toMemorySegmentSimple(arena, i)
};
});
}
@Override
protected MemorySegment[] getKey1() {
return new MemorySegment[] {
protected Keys getKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3)
};
});
}
@Override
protected MemorySegment[] getKey2() {
return new MemorySegment[] {
protected Keys getKey2() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 4)
};
});
}
@Override
protected MemorySegment[] getCollidingKey1() {
return new MemorySegment[] {
protected Keys getCollidingKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 5)
};
});
}
@Override

View File

@ -3,6 +3,7 @@ package it.cavallium.rockserver.core.impl.test;
import static it.cavallium.rockserver.core.common.Utils.toMemorySegmentSimple;
import it.cavallium.rockserver.core.common.ColumnHashType;
import it.cavallium.rockserver.core.common.Keys;
import it.unimi.dsi.fastutil.objects.ObjectList;
import java.lang.foreign.MemorySegment;
@ -14,48 +15,48 @@ public class EmbeddedDBFixedWithoutValueTest extends EmbeddedDBTest {
}
@Override
protected MemorySegment[] getKeyI(int i) {
return new MemorySegment[] {
protected Keys getKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 7),
toMemorySegmentSimple(arena, i)
};
});
}
@Override
protected MemorySegment[] getNotFoundKeyI(int i) {
return new MemorySegment[] {
protected Keys getNotFoundKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 5, 6),
toMemorySegmentSimple(arena, i)
};
});
}
@Override
protected MemorySegment[] getKey1() {
return new MemorySegment[] {
protected Keys getKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3)
};
});
}
@Override
protected MemorySegment[] getKey2() {
return new MemorySegment[] {
protected Keys getKey2() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 4)
};
});
}
@Override
protected MemorySegment[] getCollidingKey1() {
return new MemorySegment[] {
protected Keys getCollidingKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 5)
};
});
}
@Override

View File

@ -3,6 +3,7 @@ package it.cavallium.rockserver.core.impl.test;
import static it.cavallium.rockserver.core.common.Utils.toMemorySegmentSimple;
import it.cavallium.rockserver.core.client.EmbeddedConnection;
import it.cavallium.rockserver.core.common.Keys;
import it.cavallium.rockserver.core.common.RequestType;
import it.cavallium.rockserver.core.common.ColumnHashType;
import it.cavallium.rockserver.core.common.ColumnSchema;
@ -26,9 +27,9 @@ abstract class EmbeddedDBTest {
protected long colId = 0L;
protected Arena arena;
protected MemorySegment bigValue;
protected MemorySegment[] key1;
protected MemorySegment[] collidingKey1;
protected MemorySegment[] key2;
protected Keys key1;
protected Keys collidingKey1;
protected Keys key2;
protected MemorySegment value1;
protected MemorySegment value2;
@ -56,34 +57,34 @@ abstract class EmbeddedDBTest {
return Utils.toMemorySegment(arena, bigValueArray);
}
protected MemorySegment[] getKey2() {
return new MemorySegment[] {
protected Keys getKey2() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 7)
};
});
}
protected MemorySegment[] getCollidingKey1() {
return new MemorySegment[] {
protected Keys getCollidingKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, -48)
};
});
}
protected MemorySegment[] getKey1() {
return new MemorySegment[] {
protected Keys getKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 8)
};
});
}
protected boolean getHasValues() {
@ -120,24 +121,24 @@ abstract class EmbeddedDBTest {
}
}
protected MemorySegment[] getKeyI(int i) {
return new MemorySegment[] {
protected Keys getKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 7, i)
};
});
}
protected MemorySegment[] getNotFoundKeyI(int i) {
return new MemorySegment[] {
protected Keys getNotFoundKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 4, 6),
toMemorySegmentSimple(arena, 3),
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 0, i)
};
});
}
protected MemorySegment getValueI(int i) {

View File

@ -2,44 +2,45 @@ package it.cavallium.rockserver.core.impl.test;
import static it.cavallium.rockserver.core.common.Utils.toMemorySegmentSimple;
import it.cavallium.rockserver.core.common.Keys;
import it.unimi.dsi.fastutil.ints.IntList;
import java.lang.foreign.MemorySegment;
public class EmbeddedDBVarKeysWithValueTest extends EmbeddedDBTest {
protected MemorySegment[] getKeyI(int i) {
return new MemorySegment[] {
protected Keys getKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 7, i)
};
});
}
protected MemorySegment[] getNotFoundKeyI(int i) {
return new MemorySegment[] {
protected Keys getNotFoundKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 0, i)
};
});
}
protected MemorySegment[] getKey2() {
return new MemorySegment[] {
protected Keys getKey2() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 7)
};
});
}
protected MemorySegment[] getCollidingKey1() {
return new MemorySegment[] {
protected Keys getCollidingKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, -48)
};
});
}
protected MemorySegment[] getKey1() {
return new MemorySegment[] {
protected Keys getKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 8)
};
});
}
@Override

View File

@ -2,44 +2,45 @@ package it.cavallium.rockserver.core.impl.test;
import static it.cavallium.rockserver.core.common.Utils.toMemorySegmentSimple;
import it.cavallium.rockserver.core.common.Keys;
import it.unimi.dsi.fastutil.ints.IntList;
import java.lang.foreign.MemorySegment;
public class EmbeddedDBVarKeysWithoutValueTest extends EmbeddedDBTest {
protected MemorySegment[] getKeyI(int i) {
return new MemorySegment[] {
protected Keys getKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 7, i)
};
});
}
protected MemorySegment[] getNotFoundKeyI(int i) {
return new MemorySegment[] {
protected Keys getNotFoundKeyI(int i) {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 8, 2, 5, 1, 0, i)
};
});
}
protected MemorySegment[] getKey2() {
return new MemorySegment[] {
protected Keys getKey2() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 7)
};
});
}
protected MemorySegment[] getCollidingKey1() {
return new MemorySegment[] {
protected Keys getCollidingKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, -48)
};
});
}
protected MemorySegment[] getKey1() {
return new MemorySegment[] {
protected Keys getKey1() {
return new Keys(new MemorySegment[] {
toMemorySegmentSimple(arena, 1, 2, 3),
toMemorySegmentSimple(arena, 6, 7, 8)
};
});
}
@Override