Discuss new NetTrace V6 format (#2170)

Author: noahfalk

src/TraceEvent/DynamicTraceEventParser.cs

@@ -626,9 +626,27 @@ private object GetPayloadValueAt(ref PayloadFetch payloadFetch, int offset, int
                 case TypeCode.UInt32:
                     return (UInt32)GetInt32At(offset);
                 case TypeCode.Int64:
-                    return GetInt64At(offset);
+                    {
+                        if (payloadFetch.Size == VARINT)
+                        {
+                            return GetVarIntAt(offset);
+                        }
+                        else
+                        {
+                            return GetInt64At(offset);
+                        }
+                    }
                 case TypeCode.UInt64:
-                    return (UInt64)GetInt64At(offset);
+                    {
+                        if (payloadFetch.Size == VARINT)
+                        {
+                            return GetVarUIntAt(offset);
+                        }
+                        else
+                        {
+                            return (UInt64)GetInt64At(offset);
+                        }
+                    }
                 case TypeCode.Single:
                     return GetSingleAt(offset);
                 case TypeCode.Double:
@@ -1171,6 +1189,10 @@ internal int OffsetOfNextField(ref PayloadFetch payloadFetch, int offset, int pa
 
                     return offset + elemSize;
                 }
+                else if (size == VARINT)
+                {
+                    return SkipVarInt(offset);
+                }
                 else
                 {
                     return ushort.MaxValue;     // Something sure to fail 
@@ -1238,8 +1260,9 @@ internal static ushort SizeOfType(Type type)
         // Size 0xFFEF is NULL_TERMINATED | IS_ANSI
         internal const ushort NULL_TERMINATED = 0xFFEE; // value is a null terminated string.   
 
-        internal const ushort POINTER_SIZE = 0xFFED;        // It is the pointer size of the target machine. 
-        internal const ushort UNKNOWN_SIZE = 0xFFEC;        // Generic unknown.
+        internal const ushort POINTER_SIZE = 0xFFED;        // It is the pointer size of the target machine.
+        internal const ushort VARINT = 0xFFEC;
+        internal const ushort UNKNOWN_SIZE = 0xFFEB;        // Generic unknown.
         internal const ushort SPECIAL_SIZES = UNKNOWN_SIZE; // This is always the smallest size as an unsiged number.    
 
         internal struct PayloadFetch

src/TraceEvent/EventPipe/EventCache.cs

@@ -12,44 +12,46 @@ namespace Microsoft.Diagnostics.Tracing.EventPipe
 
     internal class EventCache
     {
+        public EventCache(EventPipeEventSource source, ThreadCache threads)
+        {
+            _source = source;
+            _threads = threads;
+        }
+
         public event ParseBufferItemFunction OnEvent;
         public event Action<int> OnEventsDropped;
 
-        public unsafe void ProcessEventBlock(byte[] eventBlockData)
+        public void ProcessEventBlock(byte[] eventBlockData, long streamOffset)
         {
+            PinnedBuffer buffer = new PinnedBuffer(eventBlockData);
+            SpanReader reader = new SpanReader(eventBlockData, streamOffset);
+
             // parse the header
-            if(eventBlockData.Length < 20)
+            if (eventBlockData.Length < 20)
             {
-                Debug.Assert(false, "Expected EventBlock of at least 20 bytes");
-                return;
+                throw new FormatException("Expected EventBlock of at least 20 bytes");
             }
-            ushort headerSize = BitConverter.ToUInt16(eventBlockData, 0);
+            ushort headerSize = reader.ReadUInt16();
             if(headerSize < 20 || headerSize > eventBlockData.Length)
             {
-                Debug.Assert(false, "Invalid EventBlock header size");
-                return;
+                throw new FormatException("Invalid EventBlock header size");
             }
-            ushort flags = BitConverter.ToUInt16(eventBlockData, 2);
+            ushort flags = reader.ReadUInt16();
             bool useHeaderCompression = (flags & (ushort)EventBlockFlags.HeaderCompression) != 0;
 
+            // skip the rest of the header
+            reader.ReadBytes(headerSize - 4);
+
             // parse the events
-            PinnedBuffer buffer = new PinnedBuffer(eventBlockData);
-            byte* cursor = (byte*)buffer.PinningHandle.AddrOfPinnedObject();
-            byte* end = cursor + eventBlockData.Length;
-            cursor += headerSize;
             EventMarker eventMarker = new EventMarker(buffer);
             long timestamp = 0;
-            EventPipeEventHeader.ReadFromFormatV4(cursor, useHeaderCompression, ref eventMarker.Header);
-            if (!_threads.TryGetValue(eventMarker.Header.CaptureThreadId, out EventCacheThread thread))
-            {
-                thread = new EventCacheThread();
-                thread.SequenceNumber = eventMarker.Header.SequenceNumber - 1;
-                AddThread(eventMarker.Header.CaptureThreadId, thread);
-            }
+            SpanReader tempReader = reader;
+            _source.ReadEventHeader(ref tempReader, useHeaderCompression, ref eventMarker.Header);
+            EventPipeThread thread = _threads.GetOrAddThread(eventMarker.Header.CaptureThreadIndexOrId, eventMarker.Header.SequenceNumber - 1);
             eventMarker = new EventMarker(buffer);
-            while (cursor < end)
+            while (reader.RemainingBytes.Length > 0)
             {
-                EventPipeEventHeader.ReadFromFormatV4(cursor, useHeaderCompression, ref eventMarker.Header);
+                _source.ReadEventHeader(ref reader, useHeaderCompression, ref eventMarker.Header);
                 bool isSortedEvent = eventMarker.Header.IsSorted;
                 timestamp = eventMarker.Header.TimeStamp;
                 int sequenceNumber = eventMarker.Header.SequenceNumber;
@@ -58,26 +60,10 @@ public unsafe void ProcessEventBlock(byte[] eventBlockData)
                     thread.LastCachedEventTimestamp = timestamp;
 
                     // sorted events are the only time the captureThreadId should change
-                    long captureThreadId = eventMarker.Header.CaptureThreadId;
-                    if (!_threads.TryGetValue(captureThreadId, out thread))
-                    {
-                        thread = new EventCacheThread();
-                        thread.SequenceNumber = sequenceNumber - 1;
-                        AddThread(captureThreadId, thread);
-                    }
-                }
-
-                int droppedEvents = (int)Math.Min(int.MaxValue, sequenceNumber - thread.SequenceNumber - 1);
-                if(droppedEvents > 0)
-                {
-                    OnEventsDropped?.Invoke(droppedEvents);
-                }
-                else
-                {
-                    // When a thread id is recycled the sequenceNumber can abruptly reset to 1 which
-                    // makes droppedEvents go negative
-                    Debug.Assert(droppedEvents == 0 || sequenceNumber == 1);
+                    long captureThreadId = eventMarker.Header.CaptureThreadIndexOrId;
+                    thread = _threads.GetOrAddThread(captureThreadId, sequenceNumber - 1);
                 }
+                NotifyDroppedEventsIfNeeded(thread.SequenceNumber, sequenceNumber - 1);
                 thread.SequenceNumber = sequenceNumber;
 
                 if(isSortedEvent)
@@ -91,81 +77,88 @@ public unsafe void ProcessEventBlock(byte[] eventBlockData)
 
                 }
 
-                cursor += eventMarker.Header.TotalNonHeaderSize + eventMarker.Header.HeaderSize;
+                reader.ReadBytes(eventMarker.Header.PayloadSize);
                 EventMarker lastEvent = eventMarker;
                 eventMarker = new EventMarker(buffer);
                 eventMarker.Header = lastEvent.Header;
             }
             thread.LastCachedEventTimestamp = timestamp;
         }
 
-        public unsafe void ProcessSequencePointBlock(byte[] sequencePointBytes)
+        public void ProcessSequencePointBlockV5OrLess(byte[] sequencePointBytes, long streamOffset)
         {
-            const int SizeOfTimestampAndThreadCount = 12;
-            const int SizeOfThreadIdAndSequenceNumber = 12;
-            if(sequencePointBytes.Length < SizeOfTimestampAndThreadCount)
+            SpanReader reader = new SpanReader(sequencePointBytes, streamOffset);
+            long timestamp = (long)reader.ReadUInt64();
+            int threadCount = (int)reader.ReadUInt32();
+            Flush(timestamp);
+            TrimEventsAfterSequencePoint();
+            for (int i = 0; i < threadCount; i++)
             {
-                Debug.Assert(false, "Bad sequence point block length");
-                return;
+                long captureThreadId = (long)reader.ReadUInt64();
+                int sequenceNumber = (int)reader.ReadUInt32();
+                CheckpointThread(captureThreadId, sequenceNumber);
             }
-            long timestamp = BitConverter.ToInt64(sequencePointBytes, 0);
-            int threadCount = BitConverter.ToInt32(sequencePointBytes, 8);
-            if(sequencePointBytes.Length < SizeOfTimestampAndThreadCount + threadCount*SizeOfThreadIdAndSequenceNumber)
+        }
+
+        enum SequencePointFlags : uint
+        {
+            FlushThreads = 1
+        }
+
+        public void ProcessSequencePointBlockV6OrGreater(byte[] sequencePointBytes, long streamOffset)
+        {
+            SpanReader reader = new SpanReader(sequencePointBytes, streamOffset);
+            long timestamp = (long)reader.ReadUInt64();
+            uint flags = reader.ReadUInt32();
+            int threadCount = (int)reader.ReadUInt32();
+            Flush(timestamp);
+            TrimEventsAfterSequencePoint();
+            for (int i = 0; i < threadCount; i++)
             {
-                Debug.Assert(false, "Bad sequence point block length");
-                return;
+                long captureThreadIndex = (long)reader.ReadVarUInt64();
+                int sequenceNumber = (int)reader.ReadVarUInt32();
+                CheckpointThread(captureThreadIndex, sequenceNumber);
             }
+
+            if((flags & (uint)SequencePointFlags.FlushThreads) != 0)
+            {
+                _threads.Flush();
+            }
+        }
+
+        /// <summary>
+        /// After all events have been parsed we could have some straglers that weren't
+        /// earlier than any sorted event. Sort and dispatch those now.
+        /// </summary>
+        public void Flush() => Flush(long.MaxValue);
+
+        private void Flush(long timestamp)
+        {
             SortAndDispatch(timestamp);
-            foreach(EventCacheThread thread in _threads.Values)
+            CheckForPendingThreadRemoval();
+        }
+
+        private void TrimEventsAfterSequencePoint()
+        {
+            foreach (EventPipeThread thread in _threads.Values)
             {
                 Debug.Assert(thread.Events.Count == 0, "There shouldn't be any pending events after a sequence point");
                 thread.Events.Clear();
                 thread.Events.TrimExcess();
             }
+        }
 
-            int cursor = SizeOfTimestampAndThreadCount;
-            for(int i = 0; i < threadCount; i++)
+        private void CheckForPendingThreadRemoval()
+        {
+            foreach (var thread in _threads.Values)
             {
-                long captureThreadId = BitConverter.ToInt64(sequencePointBytes, cursor);
-                int sequenceNumber = BitConverter.ToInt32(sequencePointBytes, cursor + 8);
-                if (!_threads.TryGetValue(captureThreadId, out EventCacheThread thread))
+                if (thread.RemovalPending && thread.Events.Count == 0)
                 {
-                    if(sequenceNumber > 0)
-                    {
-                        OnEventsDropped?.Invoke(sequenceNumber);
-                    }
-                    thread = new EventCacheThread();
-                    thread.SequenceNumber = sequenceNumber;
-                    AddThread(captureThreadId, thread);
-                }
-                else
-                {
-                    int droppedEvents = unchecked(sequenceNumber - thread.SequenceNumber);
-                    if (droppedEvents > 0)
-                    {
-                        OnEventsDropped?.Invoke(droppedEvents);
-                    }
-                    else
-                    {
-                        // When a thread id is recycled the sequenceNumber can abruptly reset to 1 which
-                        // makes droppedEvents go negative
-                        Debug.Assert(droppedEvents == 0 || sequenceNumber == 1);
-                    }
-                    thread.SequenceNumber = sequenceNumber;
+                    _threads.RemoveThread(thread.ThreadId);
                 }
-                cursor += SizeOfThreadIdAndSequenceNumber;
             }
         }
 
-        /// <summary>
-        /// After all events have been parsed we could have some straglers that weren't
-        /// earlier than any sorted event. Sort and dispatch those now.
-        /// </summary>
-        public void Flush()
-        {
-            SortAndDispatch(long.MaxValue);
-        }
-
         private unsafe void SortAndDispatch(long stopTimestamp)
         {
             // This sort could be made faster by using a min-heap but this is a simple place to start
@@ -203,7 +196,7 @@ private unsafe void SortAndDispatch(long stopTimestamp)
             // from the cache or memory usage will grow unbounded. AddThread handles the
             // the thread objects but the storage for the queue elements also does not shrink
             // below the high water mark unless we free it explicitly.
-            foreach(Queue<EventMarker> q in threadQueues)
+            foreach (Queue<EventMarker> q in threadQueues)
             {
                 if(q.Count == 0)
                 {
@@ -212,45 +205,46 @@ private unsafe void SortAndDispatch(long stopTimestamp)
             }
         }
 
-        private void AddThread(long captureThreadId, EventCacheThread thread)
+        public void CheckpointThreadAndPendRemoval(long threadIndex, int sequenceNumber)
+        {
+            EventPipeThread thread = _threads.GetThread(threadIndex);
+            CheckpointThread(thread, sequenceNumber);
+            thread.RemovalPending = true;
+        }
+
+        public void CheckpointThread(long threadIndex, int sequenceNumber)
+        {
+            EventPipeThread thread = _threads.GetOrAddThread(threadIndex, sequenceNumber);
+            CheckpointThread(thread, sequenceNumber);
+        }
+
+        private void CheckpointThread(EventPipeThread thread, int sequenceNumber)
+        {
+            NotifyDroppedEventsIfNeeded(thread.SequenceNumber, sequenceNumber);
+            thread.SequenceNumber = sequenceNumber;
+        }
+
+        private void NotifyDroppedEventsIfNeeded(int sequenceNumber, int expectedSequenceNumber)
         {
-            // To ensure we don't have unbounded growth we evict old threads to make room
-            // for new ones. Evicted threads can always be re-added later if they log again
-            // but there are two consequences:
-            // a) We won't detect lost events on that thread after eviction
-            // b) If the thread still had events pending dispatch they will be lost
-            // We pick the thread that has gone the longest since it last logged an event
-            // under the presumption that it is probably dead, has no events, and won't
-            // log again.
-            //
-            // In the future if we had explicit thread death notification events we could keep
-            // this cache leaner.
-            if(_threads.Count >= 5000)
+            // Either events were dropped or the sequence number was reset because the thread ID was recycled.
+            // V6 format never recycles thread indexes but prior formats do. We assume heuristically that if an event or sequence
+            // point implies the last sequenceNumber was zero then the thread was recycled.
+            if (_source.FileFormatVersionNumber >= 6 || sequenceNumber != 0)
             {
-                long oldestThreadCaptureId = -1;
-                long smallestTimestamp = long.MaxValue;
-                foreach(var kv in _threads)
+                int droppedEvents = unchecked(expectedSequenceNumber - sequenceNumber);
+                if (droppedEvents < 0)
                 {
-                    if(kv.Value.LastCachedEventTimestamp < smallestTimestamp)
-                    {
-                        smallestTimestamp = kv.Value.LastCachedEventTimestamp;
-                        oldestThreadCaptureId = kv.Key;
-                    }
+                    droppedEvents = int.MaxValue;
+                }
+                if (droppedEvents > 0)
+                {
+                    OnEventsDropped?.Invoke(droppedEvents);
                 }
-                Debug.Assert(oldestThreadCaptureId != -1);
-                _threads.Remove(oldestThreadCaptureId);
             }
-            _threads[captureThreadId] = thread;
         }
 
-        Dictionary<long, EventCacheThread> _threads = new Dictionary<long, EventCacheThread>();
-    }
-
-    internal class EventCacheThread
-    {
-        public Queue<EventMarker> Events = new Queue<EventMarker>();
-        public int SequenceNumber;
-        public long LastCachedEventTimestamp;
+        EventPipeEventSource _source;
+        ThreadCache _threads;
     }
 
     internal class EventMarker