Merge branch 'main' into luaMemoryLimits

.github/workflows/ci.yml

@@ -121,6 +121,10 @@ jobs:
     steps:
       - name: Check out code
         uses: actions/checkout@v4
+      - name: Set workaround for libaio on Ubuntu 24.04 (see https://askubuntu.com/questions/1512196/libaio1-on-noble/1512197#1512197)
+        run: |
+          sudo ln -s /usr/lib/x86_64-linux-gnu/libaio.so.1t64 /usr/lib/x86_64-linux-gnu/libaio.so.1
+        if: ${{ matrix.os == 'ubuntu-latest' }}
       - name: Set environment variable for Linux
         run: echo "RunAzureTests=yes" >> $GITHUB_ENV
         if: ${{ matrix.os == 'ubuntu-latest' }}

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- Versioning property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.49</VersionPrefix>
+		<VersionPrefix>1.0.51</VersionPrefix>
 	</PropertyGroup>
 </Project>

libs/cluster/Server/ClusterConfig.cs

@@ -943,6 +943,8 @@ private ClusterConfig MergeWorkerInfo(Worker worker)
 
         public ClusterConfig MergeSlotMap(ClusterConfig senderConfig, ILogger logger = null)
         {
+            // Track if update happened to avoid expensive merge and FlushConfig operation when possible
+            var updated = false;
             var senderSlotMap = senderConfig.slotMap;
             var senderWorkerId = GetWorkerIdFromNodeId(senderConfig.LocalNodeId);
 
@@ -979,12 +981,16 @@ public ClusterConfig MergeSlotMap(ClusterConfig senderConfig, ILogger logger = n
                 if (senderConfig.LocalNodeConfigEpoch != 0 && workers[currentOwnerId].ConfigEpoch >= senderConfig.LocalNodeConfigEpoch)
                     continue;
 
+                // Update happened only if workerId or state changed
+                // NOTE: this avoids message flooding when sender epoch equals zero
+                updated = newSlotMap[i]._workerId != senderWorkerId || newSlotMap[i]._state != SlotState.STABLE;
+
                 // Update ownership of node
                 newSlotMap[i]._workerId = senderWorkerId;
                 newSlotMap[i]._state = SlotState.STABLE;
             }
 
-            return new(newSlotMap, workers);
+            return updated ? new(newSlotMap, workers) : this;
         }
 
         /// <summary>

libs/common/RespWriteUtils.cs

@@ -8,6 +8,7 @@
 using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
 using System.Text;
+using Tsavorite.core;
 
 namespace Garnet.common
 {
@@ -697,6 +698,27 @@ public static bool WriteArrayWithNullElements(int len, ref byte* curr, byte* end
             return true;
         }
 
+        /// <summary>
+        /// Writes an array consisting of an ETag followed by a Bulk string value into the buffer.
+        /// NOTE: Caller should make sure there is enough space in the buffer for sending the etag, and value array.
+        /// </summary>
+        /// <param name="etag">etag value to write in the array</param>
+        /// <param name="value">value to write in the array</param>
+        /// <param name="curr">start of destination buffer</param>
+        /// <param name="end">end of destincation buffer</param>
+        /// <param name="writeDirect">Whether to write the value directly to buffer or transform it to a resp bulk string</param>
+        public static void WriteEtagValArray(long etag, ref ReadOnlySpan<byte> value, ref byte* curr, byte* end, bool writeDirect)
+        {
+            // Writes a Resp encoded Array of Integer for ETAG as first element, and bulk string for value as second element
+            RespWriteUtils.WriteArrayLength(2, ref curr, end);
+            RespWriteUtils.WriteInteger(etag, ref curr, end);
+
+            if (writeDirect)
+                RespWriteUtils.WriteDirect(value, ref curr, end);
+            else
+                RespWriteUtils.WriteBulkString(value, ref curr, end);
+        }
+
         /// <summary>
         /// Write newline (\r\n) to <paramref name="curr"/>
         /// </summary>