@@ -69,23 +69,28 @@ public AofAddress PreviousAddress
6969 }
7070
7171 /// <summary>
72- /// Return previous address for a specific sublog without copying the full AofAddress struct
72+ /// Return previous address for a specific sublog without copying the full AofAddress struct.
73+ /// This is the address actually shipped (Consume'd) to the replica, used to bound AOF
74+ /// truncation conservatively.
7375 /// </summary>
7476 /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
7577 /// <returns>The previous address of the specified sublog's sync task.</returns>
7678 public long GetPreviousAddress ( int physicalSublogIdx ) => aofSyncTasks [ physicalSublogIdx ] . PreviousAddress ;
7779
7880 /// <summary>
79- /// Return start address for a specific sublog without copying the full AofAddress struct
81+ /// Return the shipped watermark address for a specific sublog (the monotonic high-water this
82+ /// replica's sync task has shipped/scanned past), used to drive the backpressure gate.
8083 /// </summary>
8184 /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
82- /// <returns>The start address of the specified sublog's sync task.</returns>
83- public long GetStartAddress ( int physicalSublogIdx ) => aofSyncTasks [ physicalSublogIdx ] . StartAddress ;
85+ /// <returns>The shipped watermark address of the specified sublog's sync task.</returns>
86+ public long GetShippedWatermarkAddress ( int physicalSublogIdx ) => aofSyncTasks [ physicalSublogIdx ] . ShippedWatermarkAddress ;
8487
8588 /// <summary>
86- /// Replica endpoint
89+ /// Return start address for a specific sublog without copying the full AofAddress struct
8790 /// </summary>
88- readonly IPEndPoint endPoint ;
91+ /// <param name="physicalSublogIdx">Index of the physical sublog.</param>
92+ /// <returns>The start address of the specified sublog's sync task.</returns>
93+ public long GetStartAddress ( int physicalSublogIdx ) => aofSyncTasks [ physicalSublogIdx ] . StartAddress ;
8994
9095 public AofSyncDriver (
9196 ClusterProvider clusterProvider ,
@@ -100,13 +105,12 @@ public AofSyncDriver(
100105 this . aofSyncDriverStore = aofSyncDriverStore ;
101106 this . localNodeId = localNodeId ;
102107 this . remoteNodeId = remoteNodeId ;
103- this . endPoint = endPoint ;
104108 cts = new ( ) ;
105109 this . logger = logger ;
106110
107111 aofSyncTasks = new AofSyncTask [ clusterProvider . serverOptions . AofPhysicalSublogCount ] ;
108112 for ( var physicalSublogIdx = 0 ; physicalSublogIdx < aofSyncTasks . Length ; physicalSublogIdx ++ )
109- aofSyncTasks [ physicalSublogIdx ] = new AofSyncTask ( clusterProvider , physicalSublogIdx , endPoint , startAddress [ physicalSublogIdx ] , localNodeId , remoteNodeId , cts , logger ) ;
113+ aofSyncTasks [ physicalSublogIdx ] = new AofSyncTask ( clusterProvider , aofSyncDriverStore , physicalSublogIdx , endPoint , startAddress [ physicalSublogIdx ] , localNodeId , remoteNodeId , cts , logger ) ;
110114 }
111115
112116 /// <summary>
@@ -152,10 +156,9 @@ public async Task RunAsync()
152156 }
153157 else
154158 {
155- var tasks = new Task [ aofSyncTasks . Length + 1 ] ;
156- tasks [ 0 ] = AdvancePhysicalSublogTimeAsync ( ) ;
159+ var tasks = new Task [ aofSyncTasks . Length ] ;
157160 for ( var i = 0 ; i < aofSyncTasks . Length ; i ++ )
158- tasks [ i + 1 ] = aofSyncTasks [ i ] . RunAofSyncTaskAsync ( this ) ;
161+ tasks [ i ] = aofSyncTasks [ i ] . RunAofSyncTaskAsync ( this ) ;
159162
160163 _ = await Task . WhenAny ( tasks ) . ConfigureAwait ( false ) ;
161164 }
@@ -174,60 +177,6 @@ public async Task RunAsync()
174177 }
175178 }
176179
177- /// <summary>
178- /// Advance physical sublog time background task.
179- /// </summary>
180- /// <returns></returns>
181- /// <exception cref="GarnetException"></exception>
182- /// <seealso cref="T:Garnet.cluster.ReplicationManager.AdvanceTime"/>
183- async Task AdvancePhysicalSublogTimeAsync ( )
184- {
185- var enteredMonitor = false ;
186- var client = new GarnetClientSession (
187- endPoint ,
188- clusterProvider . replicationManager . GetAofSyncNetworkBufferSettings ,
189- clusterProvider . replicationManager . GetNetworkPool ,
190- tlsOptions : this . clusterProvider . serverOptions . TlsOptions ? . TlsClientOptions ,
191- authUsername : this . clusterProvider . ClusterUsername ,
192- authPassword : this . clusterProvider . ClusterPassword ,
193- logger : logger ) ;
194-
195- try
196- {
197- enteredMonitor = activeWorkerMonitor . TryEnter ( ) ;
198- if ( ! enteredMonitor )
199- throw new GarnetException ( $ "Failed to acquire read lock at { nameof ( AdvancePhysicalSublogTimeAsync ) } ") ;
200-
201- // Connect to replica
202- await client . ConnectAsync ( ( int ) clusterProvider . serverOptions . ReplicaSyncTimeout . TotalMilliseconds , cts . Token ) . ConfigureAwait ( false ) ;
203-
204- var appendOnlyFile = clusterProvider . storeWrapper . appendOnlyFile ;
205- var previousTailAddress = AofAddress . Create ( appendOnlyFile . Log . Size , 0 ) ;
206-
207- while ( ! cts . IsCancellationRequested )
208- {
209- await Task . Delay ( clusterProvider . storeWrapper . runtimeConfig . GetInt ( ServerConfigType . AOF_TAIL_WITNESS_FREQ ) , cts . Token ) . ConfigureAwait ( false ) ;
210- var currentTailAddress = appendOnlyFile . Log . TailAddress ;
211- var newWrites = previousTailAddress . AnyLesser ( currentTailAddress ) ;
212-
213- if ( newWrites )
214- {
215- var sequenceNumber = appendOnlyFile . GetLargerThanMaximumSequenceNumber ( ) ;
216- _ = await client . ExecuteClusterAdvanceTime ( sequenceNumber , currentTailAddress . Span ) .
217- WaitAsync ( clusterProvider . serverOptions . ReplicaSyncTimeout , cts . Token ) .
218- ConfigureAwait ( false ) ;
219- previousTailAddress . MonotonicUpdate ( ref currentTailAddress ) ;
220- }
221- }
222- }
223- finally
224- {
225- if ( enteredMonitor )
226- _ = activeWorkerMonitor . Exit ( ) ;
227- client ? . Dispose ( ) ;
228- }
229- }
230-
231180 #region DisklesSyncInterface
232181 public async Task ConnectClientsAsync ( )
233182 {
0 commit comments