1 Commits

Author SHA1 Message Date
4bd7266c89 don't try to read objects we know don't exist when committing
such as new chunks with random ids
2025-04-13 11:57:22 +02:00
8 changed files with 57 additions and 10 deletions

View File

@@ -42,13 +42,16 @@ public class CachingObjectPersistentStore {
if (old != null)
newSize -= old.size();
TreePMap<JObjectKey, CacheEntry> newCache = map().plus(key, entry);
TreePMap<JObjectKey, CacheEntry> newCache = map();
while (newSize > sizeLimit) {
var del = newCache.firstEntry();
newCache = newCache.minusFirstEntry();
newSize -= del.getValue().size();
}
newCache = newCache.plus(key, entry);
return new Cache(
newCache,
newSize,

View File

@@ -43,4 +43,9 @@ public class CurrentTransaction implements Transaction {
public <T extends JData> void put(JData obj) {
transactionManager.current().put(obj);
}
@Override
public <T extends JData> void putNew(JData obj) {
transactionManager.current().putNew(obj);
}
}

View File

@@ -1,11 +1,11 @@
package com.usatiuk.objects.transaction;
import com.usatiuk.dhfs.utils.AutoCloseableNoThrow;
import com.usatiuk.objects.JData;
import com.usatiuk.objects.JDataVersionedWrapper;
import com.usatiuk.objects.JObjectKey;
import com.usatiuk.objects.snapshot.Snapshot;
import com.usatiuk.objects.snapshot.SnapshotManager;
import com.usatiuk.dhfs.utils.AutoCloseableNoThrow;
import io.quarkus.logging.Log;
import io.quarkus.runtime.StartupEvent;
import jakarta.annotation.Priority;
@@ -16,7 +16,6 @@ import jakarta.inject.Inject;
import org.apache.commons.lang3.tuple.Pair;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -106,10 +105,17 @@ public class JObjectManager {
for (var entry : curIteration.entrySet()) {
// Log.trace("Running pre-commit hook " + hook.getClass() + " for" + entry.getKey());
if (entry.getValue() instanceof TxRecord.TxObjectRecordNewWrite<?> newWrite) {
lastCurHookSeen.put(entry.getKey(), entry.getValue());
hook.onCreate(newWrite.key(), newWrite.data());
continue;
}
var oldObj = getPrev.apply(entry.getKey());
lastCurHookSeen.put(entry.getKey(), entry.getValue());
switch (entry.getValue()) {
case TxRecord.TxObjectRecordWrite<?> write -> {
case TxRecord.TxObjectRecordWriteChecked<?> write -> {
if (oldObj == null) {
hook.onCreate(write.key(), write.data());
} else {
@@ -221,8 +227,10 @@ public class JObjectManager {
writes.values().stream()
.filter(r -> {
if (!same)
if (r instanceof TxRecord.TxObjectRecordWrite<?>(JData data)) {
if (r instanceof TxRecord.TxObjectRecordWriteChecked<?>(JData data)) {
var dep = dependenciesLocked.get(data.key());
if (dep == null)
return true;
if (dep.isPresent() && dep.get().version() > snapshotId) {
Log.trace("Skipping write " + data.key() + " - dependency " + dep.get().version() + " vs " + snapshotId);
return false;

View File

@@ -14,6 +14,7 @@ public interface Transaction extends TransactionHandle {
<T extends JData> Optional<T> get(Class<T> type, JObjectKey key, LockingStrategy strategy);
<T extends JData> void put(JData obj);
<T extends JData> void putNew(JData obj);
void delete(JObjectKey key);

View File

@@ -57,6 +57,7 @@ public class TransactionFactoryImpl implements TransactionFactory {
private class TransactionImpl implements TransactionPrivate {
private final Map<JObjectKey, TransactionObject<?>> _readSet = new HashMap<>();
private final NavigableMap<JObjectKey, TxRecord.TxObjectRecord<?>> _writes = new TreeMap<>();
private final HashSet<JObjectKey> _totallyNew = new HashSet<>();
private final List<Runnable> _onCommit = new ArrayList<>();
private final List<Runnable> _onFlush = new ArrayList<>();
private final Snapshot<JObjectKey, JDataVersionedWrapper> _snapshot;
@@ -97,8 +98,13 @@ public class TransactionFactoryImpl implements TransactionFactory {
var got = _readSet.get(key);
if (got == null) {
if (_totallyNew.contains(key)) {
return Optional.empty();
}
var read = _snapshot.readObject(key);
_readSet.put(key, new TransactionObjectNoLock<>(read));
// Log.infov("Read object {0} from source, type {1}", key, type);
return read.map(JDataVersionedWrapper::data).map(type::cast);
}
@@ -175,8 +181,15 @@ public class TransactionFactoryImpl implements TransactionFactory {
@Override
public void put(JData obj) {
_writes.put(obj.key(), new TxRecord.TxObjectRecordWrite<>(obj));
_newWrites.put(obj.key(), new TxRecord.TxObjectRecordWrite<>(obj));
_writes.put(obj.key(), new TxRecord.TxObjectRecordWriteChecked<>(obj));
_newWrites.put(obj.key(), new TxRecord.TxObjectRecordWriteChecked<>(obj));
}
@Override
public <T extends JData> void putNew(JData obj) {
_writes.put(obj.key(), new TxRecord.TxObjectRecordNewWrite<>(obj));
_newWrites.put(obj.key(), new TxRecord.TxObjectRecordNewWrite<>(obj));
_totallyNew.add(obj.key());
}
@Override

View File

@@ -8,7 +8,18 @@ public class TxRecord {
JObjectKey key();
}
public record TxObjectRecordWrite<T extends JData>(JData data) implements TxObjectRecord<T> {
public interface TxObjectRecordWrite<T> extends TxObjectRecord<T> {
JData data();
}
public record TxObjectRecordWriteChecked<T extends JData>(JData data) implements TxObjectRecordWrite<T> {
@Override
public JObjectKey key() {
return data.key();
}
}
public record TxObjectRecordNewWrite<T extends JData>(JData data) implements TxObjectRecordWrite<T> {
@Override
public JObjectKey key() {
return data.key();
@@ -17,4 +28,5 @@ public class TxRecord {
public record TxObjectRecordDeleted(JObjectKey key) implements TxObjectRecord<JData> {
}
}

View File

@@ -1,9 +1,9 @@
package com.usatiuk.dhfs;
import com.usatiuk.objects.JObjectKey;
import com.usatiuk.dhfs.repository.PersistentPeerDataService;
import com.usatiuk.dhfs.repository.RemoteObjectServiceClient;
import com.usatiuk.dhfs.repository.SyncHandler;
import com.usatiuk.objects.JObjectKey;
import com.usatiuk.objects.transaction.LockingStrategy;
import com.usatiuk.objects.transaction.Transaction;
import io.quarkus.logging.Log;
@@ -123,6 +123,11 @@ public class RemoteTransaction {
curTx.put(newData);
}
public <T extends JDataRemote> void putDataNew(T obj) {
curTx.putNew(new RemoteObjectMeta(obj, persistentPeerDataService.getSelfUuid()));
curTx.putNew(new RemoteObjectDataWrapper<>(obj));
}
public Optional<RemoteObjectMeta> getMeta(JObjectKey key) {
return getMeta(key, LockingStrategy.OPTIMISTIC);
}

View File

@@ -81,7 +81,7 @@ public class DhfsFileServiceImpl implements DhfsFileService {
private ChunkData createChunk(ByteString bytes) {
var newChunk = new ChunkData(JObjectKey.of(UUID.randomUUID().toString()), bytes);
remoteTx.putData(newChunk);
remoteTx.putDataNew(newChunk);
return newChunk;
}