Eliminate heap allocations in BinaryDecoder/BinaryEncoder hot paths

BinaryDecoder:
- DecodeFloat, DecodeDouble, DecodeGuid, ReadIntFixed: replace new byte[]
  with stackalloc Span<byte>; add Span<byte> overload of ReadAllRequiredBytes
- DecodeString: bypass DecodeByteArray, rent from ArrayPool<byte>, decode
  string, return buffer — avoids a heap allocation per string field

BinaryEncoder:
- Encode(float): replace BitConverter.GetBytes (allocates) with stackalloc +
  BitConverter.TryWriteBytes
- Encode(Guid): replace ToByteArray() (allocates) with stackalloc +
  TryWriteBytes
- Encode(string): replace Encoding.UTF8.GetBytes(string) (allocates) with
  ArrayPool<byte> rent, encode into span, delegate to Encode(ReadOnlySpan<byte>)

Author: amirburbea

Sources/Core/Microsoft.StreamProcessing/Serializer/Encoders/BinaryDecoder.cs

@@ -3,6 +3,7 @@
 // Licensed under the MIT License
 // *********************************************************************
 using System;
+using System.Buffers;
 using System.IO;
 using System.Runtime.Serialization;
 using System.Text;
@@ -110,18 +111,15 @@ public ulong DecodeULong()
 
         public float DecodeFloat()
         {
-            var value = new byte[4];
+            Span<byte> value = stackalloc byte[4];
             ReadAllRequiredBytes(value);
-            if (!BitConverter.IsLittleEndian)
-            {
-                Array.Reverse(value);
-            }
-            return BitConverter.ToSingle(value, 0);
+            if (!BitConverter.IsLittleEndian) value.Reverse();
+            return BitConverter.ToSingle(value);
         }
 
         public double DecodeDouble()
         {
-            var value = new byte[8];
+            Span<byte> value = stackalloc byte[8];
             ReadAllRequiredBytes(value);
             long longValue = value[0]
                 | (long)value[1] << 0x8
@@ -142,7 +140,21 @@ public byte[] DecodeByteArray()
             return array;
         }
 
-        public string DecodeString() => Encoding.UTF8.GetString(DecodeByteArray());
+        public string DecodeString()
+        {
+            int byteCount = DecodeInt();
+            if (byteCount == 0) return string.Empty;
+            var rented = ArrayPool<byte>.Shared.Rent(byteCount);
+            try
+            {
+                ReadAllRequiredBytes(rented.AsSpan(0, byteCount));
+                return Encoding.UTF8.GetString(rented, 0, byteCount);
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(rented);
+            }
+        }
 
         public int DecodeArrayChunk()
         {
@@ -157,9 +169,9 @@ public int DecodeArrayChunk()
 
         public Guid DecodeGuid()
         {
-            var array = new byte[16];
-            ReadAllRequiredBytes(array);
-            return new Guid(array);
+            Span<byte> value = stackalloc byte[16];
+            ReadAllRequiredBytes(value);
+            return new Guid(value);
         }
 
         private void ReadAllRequiredBytes(byte[] array)
@@ -172,15 +184,26 @@ private void ReadAllRequiredBytes(byte[] array)
             }
         }
 
+        private void ReadAllRequiredBytes(Span<byte> span)
+        {
+            int totalRead = 0;
+            while (totalRead < span.Length)
+            {
+                int read = this.stream.Read(span.Slice(totalRead));
+                if (read == 0)
+                    throw new SerializationException($"Unexpected end of stream: '{span.Length - totalRead}' bytes missing.");
+                totalRead += read;
+            }
+        }
+
         private int ReadIntFixed()
         {
-            var value = new byte[4];
-            this.stream.ReadAllRequiredBytes(value, 0, value.Length);
-            int intValue = value[0]
+            Span<byte> value = stackalloc byte[4];
+            ReadAllRequiredBytes(value);
+            return value[0]
                 | value[1] << 0x8
                 | value[2] << 0x10
                 | value[3] << 0x18;
-            return intValue;
         }
 
         public unsafe T[] DecodeArray<T>() where T : struct

Sources/Core/Microsoft.StreamProcessing/Serializer/Encoders/BinaryEncoder.cs

@@ -3,6 +3,7 @@
 // Licensed under the MIT License
 // *********************************************************************
 using System;
+using System.Buffers;
 using System.IO;
 using System.Text;
 using Microsoft.StreamProcessing.Internal;
@@ -83,13 +84,10 @@ public void Encode(ulong value)
 
         public void Encode(float value)
         {
-            byte[] bytes = BitConverter.GetBytes(value);
-            if (!BitConverter.IsLittleEndian)
-            {
-                Array.Reverse(bytes);
-            }
-
-            this.stream.Write(bytes, 0, bytes.Length);
+            Span<byte> bytes = stackalloc byte[4];
+            BitConverter.TryWriteBytes(bytes, value);
+            if (!BitConverter.IsLittleEndian) bytes.Reverse();
+            this.stream.Write(bytes);
         }
 
         public void Encode(double value)
@@ -113,12 +111,40 @@ public void Encode(byte[] value)
             if (value.Length > 0) this.stream.Write(value, 0, value.Length);
         }
 
+        public void Encode(ReadOnlySpan<byte> span)
+        {
+            int count = span.Length;
+            Encode(count);
+            if (count > 0)
+            {
+                this.stream.Write(span);
+            }
+        }
+
         public void Encode(string value)
-            => Encode(Encoding.UTF8.GetBytes(value ?? throw new ArgumentNullException(nameof(value))));
+        {
+            if (value == null) throw new ArgumentNullException(nameof(value));
+            int byteCount = Encoding.UTF8.GetByteCount(value);
+            var rented = ArrayPool<byte>.Shared.Rent(byteCount);
+            try
+            {
+                int written = Encoding.UTF8.GetBytes(value, rented);
+                Encode(rented.AsSpan(0, written));
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(rented);
+            }
+        }
 
         public void EncodeArrayChunk(int size) => Encode(size);
 
-        public void Encode(Guid value) => this.stream.Write(value.ToByteArray(), 0, 16);
+        public void Encode(Guid value)
+        {
+            Span<byte> bytes = stackalloc byte[16];
+            value.TryWriteBytes(bytes);
+            this.stream.Write(bytes);
+        }
 
         private void WriteIntFixed(int encodedValue)
         {