perf(saves): drop the 9-byte per-entity placement state; write snapshots from worker segment logs
Every ISerializable carried SerializedThread/SerializedPosition/SerializedLength so WriteSnapshot could gather each entity's bytes from the worker heaps in dictionary order. The idx records absolute positions, so bin order is free - the snapshot can be written in worker-heap order instead, and the join inverts: - Chunks are persistence-homogeneous: SerializeAll declares the owner at each boundary, publishing the partial chunk on change. - Workers log segments (owner, slot range, heap start) plus one length per record as they serialize. Positions are implicit because a worker's writes are contiguous; identity comes from re-walking the same snapshot slots in the same order (guaranteed stable - mutations divert to the pending queues until PostWorldSave), or from an entities log on the fallback path. - WriteSnapshot routes segments by owner, emits idx entries during the re-walk, and writes each segment's heap bytes as a single span instead of one copy per entity, which also speeds up the background write phase. - Persistence self-payloads keep placement as three private fields on the ~dozens of persistence instances; PushSingle is now typed accordingly. Net effect: 9 bytes (plus padding) of resident state removed from every item, mobile, guild, and account on every shard; three interface-property stores per entity leave the drain hot path (stamping dirtied one cache line per entity mid-freeze - the lengths log is a single sequential stream); and the vestigial loader-side length stamp is gone. The transient cost is ~4 bytes per entity in pooled per-worker logs that are released after each write. The save format is unchanged (idx v3, same loader); only the write-side mechanics moved. Adds an end-to-end round-trip test that drives real workers through the chunk source, snapshots from the segment logs, and reloads. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kamron Batman
parent
9780fc6634
commit
3b75b96008
19 changed files with 625 additions and 189 deletions
|
|
@ -80,8 +80,8 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
|
||||
var threads = World._threadWorkers;
|
||||
|
||||
// 1MB buffer: entity spans are written individually, so the default 4KB buffer
|
||||
// costs a syscall every few entities on the snapshot thread.
|
||||
// 1MB buffer: segments are written as large spans, but idx entries and skip splits
|
||||
// still benefit on the snapshot thread.
|
||||
using var binFs = new FileStream(
|
||||
Path.Combine(dir, $"{Name}.bin"), FileMode.Create, FileAccess.Write, FileShare.None, 1024 * 1024
|
||||
);
|
||||
|
|
@ -91,24 +91,24 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
var binPosition = 0L;
|
||||
|
||||
// Support for non-entity generic serialization.
|
||||
if (SerializedLength > 0)
|
||||
if (_selfLength > 0)
|
||||
{
|
||||
try
|
||||
{
|
||||
binFs.Write(threads[SerializedThread].GetHeap(SerializedPosition, SerializedLength));
|
||||
binFs.Write(threads[_selfThread].GetHeap(_selfPosition, _selfLength));
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
logger.Error(
|
||||
error,
|
||||
"Error writing entity: (Thread: {Thread} - {Start} {Length})",
|
||||
SerializedThread,
|
||||
SerializedPosition,
|
||||
SerializedLength
|
||||
"Error writing self-payload: (Thread: {Thread} - {Start} {Length})",
|
||||
_selfThread,
|
||||
_selfPosition,
|
||||
_selfLength
|
||||
);
|
||||
}
|
||||
|
||||
binPosition += SerializedLength;
|
||||
binPosition += _selfLength;
|
||||
}
|
||||
|
||||
idx.Write(3); // Version
|
||||
|
|
@ -116,42 +116,40 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
var countPosition = idx.Position;
|
||||
idx.Write(0);
|
||||
|
||||
var entityCount = EntitiesBySerial.Count;
|
||||
foreach (var e in EntitiesBySerial.Values)
|
||||
// The bin is written in worker-heap order, not dictionary order: each worker logged
|
||||
// (segment, record lengths) as it serialized, so records pair with their bytes by
|
||||
// re-walking the same slots in the same order. The idx stores absolute positions,
|
||||
// so the loader never cares about record order.
|
||||
var entityCount = 0;
|
||||
|
||||
for (var t = 0; t < threads.Length; t++)
|
||||
{
|
||||
if (e is Item { SkipSerialization: true } or Mobile { SkipSerialization: true })
|
||||
var worker = threads[t];
|
||||
var segments = worker.Segments;
|
||||
|
||||
for (var s = 0; s < segments.Count; s++)
|
||||
{
|
||||
entityCount--;
|
||||
continue;
|
||||
var segment = segments[s];
|
||||
if (!ReferenceEquals(segment.Owner, this))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
binPosition = WriteSegmentRecords(worker, in segment, idx, binFs, binPosition, ref entityCount);
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
logger.Error(
|
||||
error,
|
||||
"Error writing segment: (Thread: {Thread} - {Start}, {Records} records)",
|
||||
t,
|
||||
segment.HeapStart,
|
||||
segment.RecordCount
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
var thread = e.SerializedThread;
|
||||
var heapStart = e.SerializedPosition;
|
||||
var heapLength = e.SerializedLength;
|
||||
|
||||
idx.Write(e.GetType());
|
||||
idx.Write(e.Serial);
|
||||
idx.Write(e.Created.Ticks);
|
||||
idx.Write(binPosition);
|
||||
idx.Write(heapLength);
|
||||
|
||||
try
|
||||
{
|
||||
binFs.Write(threads[thread].GetHeap(heapStart, heapLength));
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
logger.Error(
|
||||
error,
|
||||
"Error writing entity: {Entity} (Thread: {Thread} - {Start} {Length})",
|
||||
e,
|
||||
thread,
|
||||
heapStart,
|
||||
heapLength
|
||||
);
|
||||
}
|
||||
|
||||
binPosition += heapLength;
|
||||
}
|
||||
|
||||
var currentPosition = idx.Position;
|
||||
|
|
@ -160,6 +158,99 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
idx.Seek(currentPosition, SeekOrigin.Begin);
|
||||
}
|
||||
|
||||
private long WriteSegmentRecords(
|
||||
SerializationThreadWorker worker, in SerializedSegment segment, MemoryMapFileWriter idx, FileStream binFs,
|
||||
long binPosition, ref int entityCount
|
||||
)
|
||||
{
|
||||
var lengths = worker.Lengths;
|
||||
var lengthIndex = segment.LengthsStart;
|
||||
|
||||
var heapPos = (int)segment.HeapStart;
|
||||
var spanStart = heapPos;
|
||||
|
||||
if (segment.SlotOffset >= 0)
|
||||
{
|
||||
// Re-walk the same slots the worker serialized; occupancy cannot have changed
|
||||
// because dictionary mutations divert to the pending queues until PostWorldSave.
|
||||
var entries = Unsafe.As<ShadowEntry<T>[]>(_entriesSnapshot);
|
||||
ref var entry = ref Unsafe.Add(ref MemoryMarshal.GetArrayDataReference(entries), segment.SlotOffset);
|
||||
|
||||
for (var i = 0; i < segment.SlotCount; i++, entry = ref Unsafe.Add(ref entry, 1))
|
||||
{
|
||||
var entity = entry.Value;
|
||||
if (entity == null)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var length = lengths[lengthIndex++];
|
||||
|
||||
if (entity is Item { SkipSerialization: true } or Mobile { SkipSerialization: true })
|
||||
{
|
||||
// The bytes exist in the heap but are not part of the save: split the span.
|
||||
if (heapPos > spanStart)
|
||||
{
|
||||
binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart));
|
||||
}
|
||||
|
||||
heapPos += length;
|
||||
spanStart = heapPos;
|
||||
continue;
|
||||
}
|
||||
|
||||
idx.Write(entity.GetType());
|
||||
idx.Write(entity.Serial);
|
||||
idx.Write(entity.Created.Ticks);
|
||||
idx.Write(binPosition);
|
||||
idx.Write(length);
|
||||
|
||||
binPosition += length;
|
||||
heapPos += length;
|
||||
entityCount++;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
var bufferEntities = worker.BufferEntities;
|
||||
|
||||
for (var i = 0; i < segment.RecordCount; i++)
|
||||
{
|
||||
var entity = (T)bufferEntities[segment.EntitiesStart + i];
|
||||
var length = lengths[lengthIndex++];
|
||||
|
||||
if (entity is Item { SkipSerialization: true } or Mobile { SkipSerialization: true })
|
||||
{
|
||||
if (heapPos > spanStart)
|
||||
{
|
||||
binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart));
|
||||
}
|
||||
|
||||
heapPos += length;
|
||||
spanStart = heapPos;
|
||||
continue;
|
||||
}
|
||||
|
||||
idx.Write(entity.GetType());
|
||||
idx.Write(entity.Serial);
|
||||
idx.Write(entity.Created.Ticks);
|
||||
idx.Write(binPosition);
|
||||
idx.Write(length);
|
||||
|
||||
binPosition += length;
|
||||
heapPos += length;
|
||||
entityCount++;
|
||||
}
|
||||
}
|
||||
|
||||
if (heapPos > spanStart)
|
||||
{
|
||||
binFs.Write(worker.GetHeap(spanStart, heapPos - spanStart));
|
||||
}
|
||||
|
||||
return binPosition;
|
||||
}
|
||||
|
||||
public override void Serialize()
|
||||
{
|
||||
// Self-payload first so a large one overlaps the entity stream instead of ending it.
|
||||
|
|
@ -196,7 +287,7 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
return false;
|
||||
}
|
||||
|
||||
int ISlotRangeSource.SerializeRange(BufferWriter writer, byte threadIndex, int offset, int count)
|
||||
int ISlotRangeSource.SerializeRange(BufferWriter writer, List<int> lengths, int offset, int count)
|
||||
{
|
||||
// Layout proven at startup by ShadowDictionaryEntries.Supported; ranges are produced
|
||||
// from the same array's length, so every read is in-bounds.
|
||||
|
|
@ -212,7 +303,9 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
var entity = entry.Value;
|
||||
if (entity != null)
|
||||
{
|
||||
SerializationThreadWorker.Serialize(entity, writer, threadIndex);
|
||||
var start = writer.Position;
|
||||
entity.Serialize(writer);
|
||||
lengths.Add((int)(writer.Position - start));
|
||||
serialized++;
|
||||
}
|
||||
}
|
||||
|
|
@ -393,8 +486,6 @@ public class GenericEntityPersistence<T> : GenericPersistence, IGenericEntityPer
|
|||
if (ctor.Invoke(ctorArgs) is T entity)
|
||||
{
|
||||
entity.Created = created;
|
||||
// Cost estimate for the first save's scheduling; overwritten by every save.
|
||||
entity.SerializedLength = length;
|
||||
entities.Add(new EntitySpan<T>(entity, pos, length));
|
||||
EntitiesBySerial[serial] = entity;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,14 +25,24 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable
|
|||
public string Name { get; }
|
||||
public string SaveFilePath { get; protected set; } // "<Folder>/<System>.bin"
|
||||
|
||||
public byte SerializedThread { get; set; }
|
||||
public int SerializedPosition { get; set; }
|
||||
public int SerializedLength { get; set; }
|
||||
// Placement of the self-payload in the worker heaps for the most recent save. Only
|
||||
// persistences carry placement state — entities are located through the per-worker
|
||||
// segment logs instead.
|
||||
private protected byte _selfThread;
|
||||
private protected int _selfPosition;
|
||||
private protected int _selfLength;
|
||||
|
||||
internal void SetSelfPlacement(byte thread, int position, int length)
|
||||
{
|
||||
_selfThread = thread;
|
||||
_selfPosition = position;
|
||||
_selfLength = length;
|
||||
}
|
||||
|
||||
private long _loadedFileLength;
|
||||
|
||||
// Scheduling estimate only: previous save's payload size, or the loaded file size before the first save.
|
||||
internal long EstimatedSize => SerializedLength > 0 ? SerializedLength : _loadedFileLength;
|
||||
internal long EstimatedSize => _selfLength > 0 ? _selfLength : _loadedFileLength;
|
||||
|
||||
public GenericPersistence(string name, int priority) : base(priority)
|
||||
{
|
||||
|
|
@ -49,7 +59,7 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable
|
|||
|
||||
public override void WriteSnapshot(string savePath, HashSet<Type> typeSet)
|
||||
{
|
||||
if (SerializedLength == 0)
|
||||
if (_selfLength == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
|
@ -62,11 +72,7 @@ public abstract class GenericPersistence : Persistence, IGenericSerializable
|
|||
|
||||
using var binFs = new FileStream(file, FileMode.Create, FileAccess.Write, FileShare.None);
|
||||
|
||||
var thread = SerializedThread;
|
||||
var heapStart = SerializedPosition;
|
||||
var heapLength = SerializedLength;
|
||||
|
||||
binFs.Write(threads[thread].GetHeap(heapStart, heapLength));
|
||||
binFs.Write(threads[_selfThread].GetHeap(_selfPosition, _selfLength));
|
||||
}
|
||||
|
||||
public override unsafe void Deserialize(string savePath, Dictionary<ulong, string> typesDb)
|
||||
|
|
|
|||
|
|
@ -17,9 +17,5 @@ namespace Server;
|
|||
|
||||
public interface IGenericSerializable
|
||||
{
|
||||
byte SerializedThread { get; set; }
|
||||
int SerializedPosition { get; set; }
|
||||
int SerializedLength { get; set; }
|
||||
|
||||
void Serialize(IGenericWriter writer);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -122,6 +122,10 @@ public abstract class Persistence
|
|||
|
||||
foreach (var p in ordered)
|
||||
{
|
||||
// Chunks are persistence-homogeneous: publishing the partial chunk at each
|
||||
// boundary lets workers attribute buffer-chunk records to their owner without
|
||||
// any per-entity state.
|
||||
World.SetChunkSourceOwner(p);
|
||||
p.Serialize();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Runtime.InteropServices;
|
||||
|
||||
|
|
@ -29,10 +30,10 @@ namespace Server;
|
|||
public interface ISlotRangeSource
|
||||
{
|
||||
/// <summary>
|
||||
/// Serializes every occupied slot in [offset, offset + count) into the writer,
|
||||
/// stamping each entity's thread/position/length. Returns the number serialized.
|
||||
/// Serializes every occupied slot in [offset, offset + count) into the writer, appending
|
||||
/// each record's byte length to <paramref name="lengths"/>. Returns the number serialized.
|
||||
/// </summary>
|
||||
int SerializeRange(BufferWriter writer, byte threadIndex, int offset, int count);
|
||||
int SerializeRange(BufferWriter writer, List<int> lengths, int offset, int count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -54,22 +55,24 @@ public sealed class SerializationChunkSource
|
|||
|
||||
internal readonly struct Chunk
|
||||
{
|
||||
public readonly IGenericSerializable Single;
|
||||
public readonly GenericPersistence Single;
|
||||
public readonly IGenericSerializable[] Buffer;
|
||||
public readonly ISlotRangeSource Source;
|
||||
public readonly Persistence Owner; // buffer chunks only; ranges use Source, singles record their own placement
|
||||
public readonly int Offset;
|
||||
public readonly int Count;
|
||||
|
||||
public Chunk(IGenericSerializable single)
|
||||
public Chunk(GenericPersistence single)
|
||||
{
|
||||
Single = single;
|
||||
Count = 1;
|
||||
}
|
||||
|
||||
public Chunk(IGenericSerializable[] buffer, int count)
|
||||
public Chunk(IGenericSerializable[] buffer, int count, Persistence owner)
|
||||
{
|
||||
Buffer = buffer;
|
||||
Count = count;
|
||||
Owner = owner;
|
||||
}
|
||||
|
||||
public Chunk(ISlotRangeSource source, int offset, int count)
|
||||
|
|
@ -86,6 +89,21 @@ public sealed class SerializationChunkSource
|
|||
// Producer state - written only by the game loop thread.
|
||||
private IGenericSerializable[] _current;
|
||||
private int _count;
|
||||
private Persistence _currentOwner;
|
||||
|
||||
/// <summary>
|
||||
/// Declares the owner of subsequently pushed entities. Publishes the partial chunk when
|
||||
/// the owner changes, keeping buffer chunks persistence-homogeneous so workers can
|
||||
/// attribute their serialized records to a persistence without any per-entity state.
|
||||
/// </summary>
|
||||
public void SetOwner(Persistence owner)
|
||||
{
|
||||
if (!ReferenceEquals(_currentOwner, owner))
|
||||
{
|
||||
Flush();
|
||||
_currentOwner = owner;
|
||||
}
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public void Push(IGenericSerializable entity)
|
||||
|
|
@ -99,18 +117,18 @@ public sealed class SerializationChunkSource
|
|||
|
||||
if (++_count == ChunkCapacity)
|
||||
{
|
||||
_chunks.Enqueue(new Chunk(current, ChunkCapacity));
|
||||
_chunks.Enqueue(new Chunk(current, ChunkCapacity, _currentOwner));
|
||||
_current = null;
|
||||
_count = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Publishes the entity as a dedicated chunk regardless of its estimated size.
|
||||
/// Used for persistence self-payloads, which can be large on the first save
|
||||
/// before a <see cref="IGenericSerializable.SerializedLength"/> estimate exists.
|
||||
/// Publishes a persistence self-payload as a dedicated chunk regardless of its
|
||||
/// estimated size — it can be large on the first save before an estimate exists.
|
||||
/// The worker records the payload's placement on the persistence itself.
|
||||
/// </summary>
|
||||
public void PushSingle(IGenericSerializable entity) => _chunks.Enqueue(new Chunk(entity));
|
||||
public void PushSingle(GenericPersistence persistence) => _chunks.Enqueue(new Chunk(persistence));
|
||||
|
||||
/// <summary>
|
||||
/// Publishes slot ranges covering [0, slotCount) of a directly-iterable persistence.
|
||||
|
|
@ -133,7 +151,7 @@ public sealed class SerializationChunkSource
|
|||
{
|
||||
if (_count > 0)
|
||||
{
|
||||
_chunks.Enqueue(new Chunk(_current, _count));
|
||||
_chunks.Enqueue(new Chunk(_current, _count, _currentOwner));
|
||||
_current = null;
|
||||
_count = 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,11 +14,44 @@
|
|||
*************************************************************************/
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Threading;
|
||||
|
||||
namespace Server;
|
||||
|
||||
/// <summary>
|
||||
/// One contiguous run of records a worker serialized into its heap from a single chunk.
|
||||
/// Together with the worker's lengths log this replaces per-entity placement state:
|
||||
/// positions are implicit (a worker's writes are contiguous), and identity comes from
|
||||
/// re-walking the same slots for range segments, or from the entities log for buffer
|
||||
/// segments. The snapshot writer routes segments to files by <see cref="Owner"/>.
|
||||
/// </summary>
|
||||
internal readonly struct SerializedSegment
|
||||
{
|
||||
public readonly object Owner; // ISlotRangeSource for range segments, Persistence for buffer segments
|
||||
public readonly int SlotOffset; // -1 when the segment came from a buffer chunk
|
||||
public readonly int SlotCount;
|
||||
public readonly long HeapStart;
|
||||
public readonly int LengthsStart;
|
||||
public readonly int RecordCount;
|
||||
public readonly int EntitiesStart; // buffer segments only
|
||||
|
||||
public SerializedSegment(
|
||||
object owner, int slotOffset, int slotCount, long heapStart, int lengthsStart, int recordCount,
|
||||
int entitiesStart
|
||||
)
|
||||
{
|
||||
Owner = owner;
|
||||
SlotOffset = slotOffset;
|
||||
SlotCount = slotCount;
|
||||
HeapStart = heapStart;
|
||||
LengthsStart = lengthsStart;
|
||||
RecordCount = recordCount;
|
||||
EntitiesStart = entitiesStart;
|
||||
}
|
||||
}
|
||||
|
||||
public class SerializationThreadWorker
|
||||
{
|
||||
private const int MinHeapSize = 1024 * 1024; // 1MB
|
||||
|
|
@ -35,6 +68,28 @@ public class SerializationThreadWorker
|
|||
private long _entitiesSerialized;
|
||||
private long _bytesSerialized;
|
||||
|
||||
// What this worker serialized where, logged during the drain and consumed by
|
||||
// WriteSnapshot on the background writer thread. Cleared once the snapshot is on disk.
|
||||
private readonly List<SerializedSegment> _segments = [];
|
||||
private readonly List<int> _lengths = [];
|
||||
private readonly List<IGenericSerializable> _bufferEntities = [];
|
||||
|
||||
internal List<SerializedSegment> Segments => _segments;
|
||||
internal List<int> Lengths => _lengths;
|
||||
internal List<IGenericSerializable> BufferEntities => _bufferEntities;
|
||||
|
||||
/// <summary>
|
||||
/// Releases the write logs after the snapshot is written so serialized entity
|
||||
/// references don't linger between saves. Capacity is retained: the logs regrow to
|
||||
/// the same size every save.
|
||||
/// </summary>
|
||||
internal void ReleaseWriteLogs()
|
||||
{
|
||||
_segments.Clear();
|
||||
_lengths.Clear();
|
||||
_bufferEntities.Clear();
|
||||
}
|
||||
|
||||
public SerializationThreadWorker(int index, SerializationChunkSource chunkSource, int heapSizeHint = 0)
|
||||
: this(index, chunkSource, heapSizeHint, inline: false)
|
||||
{
|
||||
|
|
@ -98,39 +153,55 @@ public class SerializationThreadWorker
|
|||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public ReadOnlySpan<byte> GetHeap(int start, int length) => _heap.AsSpan(start, length);
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
internal static void Serialize(IGenericSerializable e, BufferWriter writer, byte threadIndex)
|
||||
{
|
||||
e.SerializedThread = threadIndex;
|
||||
var start = e.SerializedPosition = (int)writer.Position;
|
||||
e.Serialize(writer);
|
||||
e.SerializedLength = (int)(writer.Position - start);
|
||||
}
|
||||
|
||||
private static long ProcessChunk(
|
||||
in SerializationChunkSource.Chunk chunk, SerializationChunkSource chunkSource,
|
||||
BufferWriter writer, byte threadIndex
|
||||
)
|
||||
private long ProcessChunk(in SerializationChunkSource.Chunk chunk, BufferWriter writer)
|
||||
{
|
||||
if (chunk.Single != null)
|
||||
{
|
||||
Serialize(chunk.Single, writer, threadIndex);
|
||||
// Self-payloads are written to their own file, so they record placement on the
|
||||
// persistence itself instead of the segment logs.
|
||||
var start = writer.Position;
|
||||
chunk.Single.Serialize(writer);
|
||||
chunk.Single.SetSelfPlacement((byte)_index, (int)start, (int)(writer.Position - start));
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (chunk.Source != null)
|
||||
{
|
||||
return chunk.Source.SerializeRange(writer, threadIndex, chunk.Offset, chunk.Count);
|
||||
var heapStart = writer.Position;
|
||||
var lengthsStart = _lengths.Count;
|
||||
var serialized = chunk.Source.SerializeRange(writer, _lengths, chunk.Offset, chunk.Count);
|
||||
|
||||
if (serialized > 0)
|
||||
{
|
||||
_segments.Add(
|
||||
new SerializedSegment(chunk.Source, chunk.Offset, chunk.Count, heapStart, lengthsStart, serialized, -1)
|
||||
);
|
||||
}
|
||||
|
||||
return serialized;
|
||||
}
|
||||
|
||||
var buffer = chunk.Buffer;
|
||||
var count = chunk.Count;
|
||||
|
||||
var bufferHeapStart = writer.Position;
|
||||
var bufferLengthsStart = _lengths.Count;
|
||||
var entitiesStart = _bufferEntities.Count;
|
||||
|
||||
for (var i = 0; i < count; i++)
|
||||
{
|
||||
Serialize(buffer[i], writer, threadIndex);
|
||||
var e = buffer[i];
|
||||
var start = writer.Position;
|
||||
e.Serialize(writer);
|
||||
_lengths.Add((int)(writer.Position - start));
|
||||
_bufferEntities.Add(e);
|
||||
}
|
||||
|
||||
chunkSource.Return(buffer, count);
|
||||
_segments.Add(
|
||||
new SerializedSegment(chunk.Owner, -1, 0, bufferHeapStart, bufferLengthsStart, count, entitiesStart)
|
||||
);
|
||||
|
||||
_chunkSource.Return(buffer, count);
|
||||
return count;
|
||||
}
|
||||
|
||||
|
|
@ -141,13 +212,14 @@ public class SerializationThreadWorker
|
|||
/// </summary>
|
||||
public void DrainInline()
|
||||
{
|
||||
ReleaseWriteLogs();
|
||||
|
||||
var writer = new BufferWriter(_heap, true, World.SerializedTypes);
|
||||
var threadIndex = (byte)_index;
|
||||
var entities = 0L;
|
||||
|
||||
while (_chunkSource.TryTake(out var chunk))
|
||||
{
|
||||
entities += ProcessChunk(in chunk, _chunkSource, writer, threadIndex);
|
||||
entities += ProcessChunk(in chunk, writer);
|
||||
}
|
||||
|
||||
_heap = writer.Buffer;
|
||||
|
|
@ -160,13 +232,14 @@ public class SerializationThreadWorker
|
|||
private static void Execute(object obj)
|
||||
{
|
||||
var worker = (SerializationThreadWorker)obj;
|
||||
var threadIndex = (byte)worker._index;
|
||||
|
||||
var chunkSource = worker._chunkSource;
|
||||
var serializedTypes = World.SerializedTypes;
|
||||
|
||||
while (worker._startEvent.WaitOne())
|
||||
{
|
||||
worker.ReleaseWriteLogs();
|
||||
|
||||
var writer = new BufferWriter(worker._heap, true, serializedTypes);
|
||||
var entities = 0L;
|
||||
var spinner = new SpinWait();
|
||||
|
|
@ -177,7 +250,7 @@ public class SerializationThreadWorker
|
|||
if (chunkSource.TryTake(out var chunk))
|
||||
{
|
||||
spinner.Reset();
|
||||
entities += ProcessChunk(in chunk, chunkSource, writer, threadIndex);
|
||||
entities += worker.ProcessChunk(in chunk, writer);
|
||||
}
|
||||
else if (pauseRequested) // Break when finished
|
||||
{
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue