Skip to content

Commit e955b7a

Browse files
DrawaesDrawaes
committed
Updates to mem tables, hoping we can get ethermind using it shortly
1 parent e8a227d commit e955b7a

19 files changed

Lines changed: 126 additions & 72 deletions

‎src/TrimDB.Core/InMemory/MemoryTable.cs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,20 @@ public void UpdateWalHighWatermark(long offset)
2727
public abstract bool Delete(ReadOnlySpan<byte> key);
2828
public abstract SearchResult TryGet(ReadOnlySpan<byte> key, out ReadOnlySpan<byte> value);
2929

30+
public virtual SearchResult TryGetMemory(ReadOnlySpan<byte> key, out ReadOnlyMemory<byte> value)
31+
{
32+
var result = TryGet(key, out var span);
33+
if (result == SearchResult.Found)
34+
{
35+
value = span.ToArray();
36+
}
37+
else
38+
{
39+
value = default;
40+
}
41+
return result;
42+
}
43+
3044
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
3145
}
3246
}

‎src/TrimDB.Core/InMemory/SkipList32/ArrayBasedAllocator32.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,13 @@ public unsafe ReadOnlySpan<byte> GetValue(int valueLocation)
5959
return new Span<byte>(_data, valueLocation + sizeof(int), length);
6060
}
6161

62+
public ReadOnlyMemory<byte> GetValueMemory(int valueLocation)
63+
{
64+
ref var sizeOffset = ref _data[valueLocation];
65+
var length = Unsafe.ReadUnaligned<int>(ref sizeOffset);
66+
return new ReadOnlyMemory<byte>(_data, valueLocation + sizeof(int), length);
67+
}
68+
6269
[MethodImpl(MethodImplOptions.AggressiveInlining)]
6370
public static int AlignLength(int length)
6471
{

‎src/TrimDB.Core/InMemory/SkipList32/SkipList32.cs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,20 @@ public override SearchResult TryGet(ReadOnlySpan<byte> key, out ReadOnlySpan<byt
190190
return result;
191191
}
192192

193+
public override SearchResult TryGetMemory(ReadOnlySpan<byte> key, out ReadOnlyMemory<byte> value)
194+
{
195+
var result = Search(key, out var nextNode);
196+
if (result == SearchResult.Found)
197+
{
198+
value = _allocator.GetValueMemory(nextNode.ValueLocation);
199+
}
200+
else
201+
{
202+
value = default;
203+
}
204+
return result;
205+
}
206+
193207
private SearchResult Search(ReadOnlySpan<byte> key, out SkipListNode32 node)
194208
{
195209
var currentNode = _allocator.HeadNode;

‎src/TrimDB.Core/KVLog/KVLogManager.cs‎

Lines changed: 23 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,7 @@ public class FileBasedKVLogManager : KVLogManager
7272
private readonly string _fileName;
7373
private readonly string? _metadataFileName;
7474

75-
public FileBasedKVLogManager(string fileName, string? metadataFileName = null, bool waitForFlush = true)
75+
public FileBasedKVLogManager(string fileName, string? metadataFileName = null, bool waitForFlush = true, int channelCapacity = 4096)
7676
{
7777
_metadataFileName = metadataFileName;
7878
_fileName = fileName;
@@ -86,7 +86,11 @@ public FileBasedKVLogManager(string fileName, string? metadataFileName = null, b
8686
_kvLogStream.Seek(0, SeekOrigin.End);
8787
_kvLogWriter = PipeWriter.Create(_kvLogStream);
8888

89-
_channel = Channel.CreateUnbounded<PutOperation>();
89+
_channel = Channel.CreateBounded<PutOperation>(new BoundedChannelOptions(channelCapacity)
90+
{
91+
FullMode = BoundedChannelFullMode.Wait,
92+
SingleReader = true,
93+
});
9094
_consumerTask = Task.Run(() => ConsumeOperationsFromChannel());
9195
}
9296

@@ -259,20 +263,20 @@ public override async Task<Memory<byte>> ReadValueAtLocation(long offset)
259263
var intBuffer = new byte[sizeof(int)];
260264

261265
// Skip past key
262-
await fs.ReadAsync(intBuffer.AsMemory());
266+
await fs.ReadExactlyAsync(intBuffer.AsMemory());
263267
var keyLen = BinaryPrimitives.ReadInt32LittleEndian(intBuffer);
264268
fs.Seek(keyLen, SeekOrigin.Current);
265269

266270
// Skip deleted flag
267271
fs.Seek(1, SeekOrigin.Current);
268272

269273
// Read value
270-
await fs.ReadAsync(intBuffer.AsMemory());
274+
await fs.ReadExactlyAsync(intBuffer.AsMemory());
271275
var valSize = BinaryPrimitives.ReadInt32LittleEndian(intBuffer);
272276
var valBuffer = new byte[valSize];
273277
if (valSize > 0)
274278
{
275-
await fs.ReadAsync(valBuffer.AsMemory());
279+
await fs.ReadExactlyAsync(valBuffer.AsMemory());
276280
}
277281
return new Memory<byte>(valBuffer, 0, valSize);
278282
}
@@ -332,7 +336,7 @@ public async Task<long> ReadOffset()
332336
if (metaFs.Length >= sizeof(long))
333337
{
334338
var offsetBuffer = new byte[sizeof(long)];
335-
await metaFs.ReadAsync(offsetBuffer.AsMemory());
339+
await metaFs.ReadExactlyAsync(offsetBuffer.AsMemory());
336340
lastCommitted = BinaryPrimitives.ReadInt64LittleEndian(offsetBuffer);
337341
}
338342
}
@@ -369,34 +373,33 @@ public override async IAsyncEnumerable<PutOperation> GetUncommittedOperations()
369373

370374
if (keySize <= 0) break; // corrupt or unknown
371375

372-
var putOperation = new PutOperation();
373-
374376
// Read key
375377
var keyBuffer = new byte[keySize];
376-
await fs.ReadAsync(keyBuffer.AsMemory());
378+
await fs.ReadExactlyAsync(keyBuffer.AsMemory());
377379

378-
// Read deleted flag
379-
var deletedByte = new byte[1];
380-
await fs.ReadAsync(deletedByte.AsMemory());
381-
var isDeleted = deletedByte[0] != 0;
380+
// Read deleted flag — single byte, no allocation
381+
var deletedRaw = fs.ReadByte();
382+
if (deletedRaw < 0) break;
383+
var isDeleted = deletedRaw != 0;
382384

383385
// Read value length + value
384-
await fs.ReadAsync(intBuffer.AsMemory());
386+
await fs.ReadExactlyAsync(intBuffer.AsMemory());
385387
var valSize = BinaryPrimitives.ReadInt32LittleEndian(intBuffer);
386388
var valBuffer = new byte[valSize];
387389
if (valSize > 0)
388390
{
389-
await fs.ReadAsync(valBuffer.AsMemory());
391+
await fs.ReadExactlyAsync(valBuffer.AsMemory());
390392
}
391393

392394
// Advance position: keyLen(4) + key(N) + deleted(1) + valLen(4) + val(N)
393395
currentPos += sizeof(int) + keySize + 1 + sizeof(int) + valSize;
394396

395-
putOperation.Key = keyBuffer;
396-
putOperation.Value = valBuffer;
397-
putOperation.Deleted = isDeleted;
398-
399-
yield return putOperation;
397+
yield return new PutOperation
398+
{
399+
Key = keyBuffer,
400+
Value = valBuffer,
401+
Deleted = isDeleted,
402+
};
400403
}
401404
}
402405

‎src/TrimDB.Core/Storage/Blocks/AsyncCache/AsyncBlockCacheFile.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -90,8 +90,9 @@ internal unsafe ValueTask<IMemoryOwner<byte>> GetBlockAsync(int blockId)
9090
overlapped.hEvent = IntPtr.Zero;
9191
overlapped.Internal = IntPtr.Zero;
9292
overlapped.InternalHigh = IntPtr.Zero;
93-
overlapped.Offset = (uint)blockId * FileConsts.PageSize;
94-
overlapped.OffsetHigh = 0;
93+
long fileOffset = (long)blockId * FileConsts.PageSize;
94+
overlapped.Offset = (uint)(fileOffset & 0xFFFF_FFFF);
95+
overlapped.OffsetHigh = (uint)(fileOffset >> 32);
9596
overlapped.Pointer = IntPtr.Zero;
9697
overlapped.FileId = _id.FileId;
9798
overlapped.BlockId = blockId;

‎src/TrimDB.Core/Storage/Blocks/AsyncCache/AsyncBlockManager.cs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,11 +12,9 @@ internal class AsyncBlockManager
1212
{
1313
private TaskCompletionSource<bool> _taskSource = new TaskCompletionSource<bool>();
1414
private int _refCount;
15-
private FileIdentifier _fileId;
16-
private int _blockId;
1715
private AsyncBlockCacheFile _cacheFile;
1816

19-
public IMemoryOwner<byte> BlockMemory { get; set; }
17+
public IMemoryOwner<byte> BlockMemory { get; set; } = null!;
2018
public Task<bool> Task => _taskSource.Task;
2119

2220
public AsyncBlockManager(AsyncBlockCacheFile cacheFile) => _cacheFile = cacheFile;

‎src/TrimDB.Core/Storage/Blocks/CachePrototype/ProtoEventSource.cs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,6 @@ public class ProtoEventSource : EventSource
1212
{
1313
private IncrementingEventCounter _cacheHits;
1414
private IncrementingEventCounter _cacheMisses;
15-
private int _cacheHitsCount;
16-
private int _cacheMissCount;
1715
public static readonly ProtoEventSource Log = new ProtoEventSource();
1816

1917
public ProtoEventSource()

‎src/TrimDB.Core/Storage/Blocks/CachePrototype/ProtoFile.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,9 @@ public unsafe void ReadBlock(IntPtr buffer, BlockIdentifier bid)
5050
overlapped.hEvent = IntPtr.Zero;
5151
overlapped.Internal = IntPtr.Zero;
5252
overlapped.InternalHigh = IntPtr.Zero;
53-
overlapped.Offset = bid.BlockId * FileConsts.PageSize;
54-
overlapped.OffsetHigh = 0;
53+
long fileOffset = (long)bid.BlockId * FileConsts.PageSize;
54+
overlapped.Offset = (uint)(fileOffset & 0xFFFF_FFFF);
55+
overlapped.OffsetHigh = (uint)(fileOffset >> 32);
5556
overlapped.Pointer = IntPtr.Zero;
5657
overlapped.FileId = bid.FileId;
5758
overlapped.BlockId = (int)bid.BlockId;

‎src/TrimDB.Core/Storage/Blocks/CachePrototype/ProtoLRUCache.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,8 @@ public ProtoLRUCache(int maxBlocks, ConcurrentDictionary<FileIdentifier, ProtoFi
3737
public async Task<IMemoryOwner<byte>> GetMemory(FileIdentifier fileId, int blockId)
3838
{
3939
var bid = new BlockIdentifier() { BlockId = (uint)blockId, FileId = (ushort)fileId.FileId, LevelId = (ushort)fileId.Level };
40-
LinkedListNode<ProtoBlock> node;
41-
ProtoBlock block = null;
40+
LinkedListNode<ProtoBlock>? node;
41+
ProtoBlock? block = null;
4242
var lookup = _lookup;
4343

4444
lock (_lock)

‎src/TrimDB.Core/Storage/Blocks/CachePrototype/ProtoSharded.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,8 @@ public ProtoSharded(int maxBlocks)
4747

4848
private ProtoLRUCache GetCache(FileIdentifier id, int blockId)
4949
{
50-
var bits = 0b0111_1111_1111_1111 & HashCode.Combine(id.GetHashCode(), blockId);
51-
var index = System.Numerics.BitOperations.PopCount((uint)bits);
50+
var hash = (uint)HashCode.Combine(id.GetHashCode(), blockId);
51+
var index = (int)(hash & 0xF); // uniform across 16 shards
5252
return _lruCache[index];
5353
}
5454

0 commit comments

Comments
 (0)