Skip to content

Commit 7cf3f73

Browse files
rodrigoaatmicrosoftrodrigoaatmicrosoft
authored
Fixing AFA Pipe (#134)
* Fixing AFA Pipe: 1- SyncTimeSimultaneityFree 2- Stale state 3- Event partition * Addressing comment * Undoing remove of fix in AFA for otherTIme
1 parent afd9926 commit 7cf3f73

2 files changed

Lines changed: 529 additions & 99 deletions

File tree

‎Sources/Core/Microsoft.StreamProcessing/Operators/Afa/CompiledPartitionedAfaPipe_SingleEvent.cs‎

Lines changed: 63 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,11 @@ internal sealed class CompiledPartitionedAfaPipe_SingleEvent<TKey, TPayload, TRe
2020
[DataMember]
2121
private FastDictionary2<TKey, byte> seenEvent;
2222

23-
// The follow three fields' keys need to be updated in lock step
23+
// The follow two fields' keys need to be updated in lock step
2424
[DataMember]
2525
private FastDictionary2<TPartitionKey, FastMap<OutputEvent<TKey, TRegister>>> tentativeOutput;
2626
[DataMember]
2727
private FastDictionary2<TPartitionKey, long> lastSyncTime = new FastDictionary2<TPartitionKey, long>();
28-
[DataMember]
29-
private FastDictionary2<TPartitionKey, List<TKey>> seenPartitions = new FastDictionary2<TPartitionKey, List<TKey>>();
3028

3129
[Obsolete("Used only by serialization. Do not call directly.")]
3230
public CompiledPartitionedAfaPipe_SingleEvent() { }
@@ -72,16 +70,16 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
7270
var key = srckey[i];
7371
var partitionKey = this.getPartitionKey(key);
7472
int partitionIndex = EnsurePartition(partitionKey);
75-
var tentativeVisibleTraverser = new FastMap<OutputEvent<TKey, TRegister>>.VisibleTraverser(this.tentativeOutput.entries[partitionIndex].value);
7673
long synctime = src_vsync[i];
7774

7875
if (!this.IsSyncTimeSimultaneityFree)
7976
{
8077
int index;
78+
var tentativeVisibleTraverser = new FastMap<OutputEvent<TKey, TRegister>>.VisibleTraverser(this.tentativeOutput.entries[partitionIndex].value);
8179

8280
if (synctime > this.lastSyncTime.entries[partitionIndex].value) // move time forward
8381
{
84-
foreach (var mapIndex in this.seenPartitions.entries[partitionIndex].value) this.seenEvent.Remove(mapIndex);
82+
this.seenEvent.Remove(key);
8583

8684
if (this.tentativeOutput.Count > 0)
8785
{
@@ -121,14 +119,17 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
121119
this.seenEvent.entries[index].value = 2;
122120

123121
// Delete tentative output for that key
124-
var tentativeFindTraverser = new FastMap<OutputEvent<TKey, TRegister>>.FindTraverser(this.tentativeOutput.entries[partitionIndex].value);
125-
if (tentativeFindTraverser.Find(src_hash[i]))
122+
if (!this.IsSyncTimeSimultaneityFree)
126123
{
127-
while (tentativeFindTraverser.Next(out index))
124+
var tentativeFindTraverser = new FastMap<OutputEvent<TKey, TRegister>>.FindTraverser(this.tentativeOutput.entries[partitionIndex].value);
125+
if (tentativeFindTraverser.Find(src_hash[i]))
128126
{
129-
if (this.keyEqualityComparer(this.tentativeOutput.entries[partitionIndex].value.Values[index].key, srckey[i]))
127+
while (tentativeFindTraverser.Next(out index))
130128
{
131-
tentativeFindTraverser.Remove();
129+
if (this.keyEqualityComparer(this.tentativeOutput.entries[partitionIndex].value.Values[index].key, srckey[i]))
130+
{
131+
tentativeFindTraverser.Remove();
132+
}
132133
}
133134
}
134135
}
@@ -187,7 +188,6 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
187188
if (this.isFinal[ns])
188189
{
189190
var otherTime = Math.Min(state.PatternStartTimestamp + this.MaxDuration, StreamEvent.InfinitySyncTime);
190-
191191
if (!this.IsSyncTimeSimultaneityFree)
192192
{
193193
var tentativeOutputEntry = this.tentativeOutput.entries[partitionIndex].value;
@@ -254,7 +254,7 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
254254
/* (2) Start new activations from the start state(s) */
255255
if (!this.AllowOverlappingInstances && !ended) continue;
256256

257-
for (int counter = 0; counter < this.numStartStates; counter++)
257+
for (int counter = 0; counter < this.numStartStates; counter++)
258258
{
259259
int startState = this.startStates[counter];
260260
var startStateMap = this.singleEventStateMap[startState];
@@ -275,7 +275,6 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
275275
if (this.isFinal[ns])
276276
{
277277
var otherTime = Math.Min(synctime + this.MaxDuration, StreamEvent.InfinitySyncTime);
278-
279278
if (!this.IsSyncTimeSimultaneityFree)
280279
{
281280
var tentativeOutputEntry = this.tentativeOutput.entries[partitionIndex].value;
@@ -339,37 +338,62 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
339338
long synctime = src_vsync[i];
340339
if (!this.IsSyncTimeSimultaneityFree)
341340
{
342-
var partitionIndex = FastDictionary2<TPartitionKey, long>.IteratorStart;
343-
while (this.tentativeOutput.Iterate(ref partitionIndex))
341+
// Clean active states for stale partitions
342+
int seenEventIndex;
343+
if (this.activeStates.Count > 0)
344344
{
345-
if (synctime > this.lastSyncTime.entries[partitionIndex].value) // move time forward
345+
var activeVisibleTraverser = new FastMap<GroupedActiveState<TKey, TRegister>>.VisibleTraverser(this.activeStates);
346+
while (activeVisibleTraverser.Next(out int activeStateIndex, out _))
346347
{
347-
var tentativeVisibleTraverser = new FastMap<OutputEvent<TKey, TRegister>>.VisibleTraverser(this.tentativeOutput.entries[partitionIndex].value);
348-
349-
foreach (var mapIndex in this.seenPartitions.entries[partitionIndex].value) this.seenEvent.Remove(mapIndex);
350-
351-
if (this.tentativeOutput.Count > 0)
348+
var activeState = this.activeStates.Values[activeStateIndex];
349+
if (synctime >= activeState.PatternStartTimestamp + this.MaxDuration)
352350
{
353-
tentativeVisibleTraverser.currIndex = 0;
351+
// Since we know this partition is stale, remove it from seenEvent as well.
352+
this.seenEvent.Remove(activeState.key);
353+
this.activeStates.Remove(activeStateIndex);
354+
}
355+
}
356+
}
354357

355-
while (tentativeVisibleTraverser.Next(out int index, out int hash))
356-
{
357-
var elem = this.tentativeOutput.entries[partitionIndex].value.Values[index];
358+
// Clean seen events from stale partitions. This enumeration is necessary for stale partitions without active state.
359+
seenEventIndex = FastDictionary2<TPartitionKey, long>.IteratorStart;
360+
int partitionIndex;
361+
while (this.seenEvent.Iterate(ref seenEventIndex))
362+
{
363+
var partitionKey = this.getPartitionKey(this.seenEvent.entries[seenEventIndex].key);
364+
if (this.lastSyncTime.Lookup(partitionKey, out partitionIndex) && synctime > this.lastSyncTime.entries[partitionIndex].value)
365+
{
366+
this.seenEvent.Remove(this.seenEvent.entries[seenEventIndex].key);
367+
}
368+
}
358369

359-
this.batch.vsync.col[this.iter] = this.lastSyncTime.entries[partitionIndex].value;
360-
this.batch.vother.col[this.iter] = elem.other;
361-
this.batch.payload.col[this.iter] = elem.payload;
362-
this.batch.key.col[this.iter] = elem.key;
363-
this.batch.hash.col[this.iter] = hash;
364-
this.iter++;
370+
// Clean last synctime and tentative output from stale partitions (these two need to be kept in sync)
371+
partitionIndex = FastDictionary2<TPartitionKey, long>.IteratorStart;
372+
while (this.lastSyncTime.Iterate(ref partitionIndex))
373+
{
374+
// Check to see if partition is stale
375+
if (synctime > this.lastSyncTime.entries[partitionIndex].value)
376+
{
377+
// Emit tentative output from stale partitions
378+
var tentativeVisibleTraverser = new FastMap<OutputEvent<TKey, TRegister>>.VisibleTraverser(this.tentativeOutput.entries[partitionIndex].value);
379+
while (tentativeVisibleTraverser.Next(out int index, out int hash))
380+
{
381+
var elem = this.tentativeOutput.entries[partitionIndex].value.Values[index];
365382

366-
if (this.iter == Config.DataBatchSize) FlushContents();
367-
}
383+
this.batch.vsync.col[this.iter] = this.lastSyncTime.entries[partitionIndex].value;
384+
this.batch.vother.col[this.iter] = elem.other;
385+
this.batch.payload.col[this.iter] = elem.payload;
386+
this.batch.key.col[this.iter] = elem.key;
387+
this.batch.hash.col[this.iter] = hash;
388+
this.iter++;
368389

369-
this.tentativeOutput.entries[partitionIndex].value.Clear(); // Clear the tentative output list
390+
if (this.iter == Config.DataBatchSize) FlushContents();
370391
}
371392

372-
this.lastSyncTime.entries[partitionIndex].value = synctime;
393+
// Remove the partition
394+
var partitionKey = this.lastSyncTime.entries[partitionIndex].key;
395+
this.tentativeOutput.Remove(partitionKey);
396+
this.lastSyncTime.Remove(partitionKey);
373397
}
374398
}
375399
}
@@ -388,10 +412,9 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
388412

389413
if (synctime > this.lastSyncTime.entries[partitionIndex].value) // move time forward
390414
{
415+
this.seenEvent.Remove(srckey[i]);
391416
var tentativeVisibleTraverser = new FastMap<OutputEvent<TKey, TRegister>>.VisibleTraverser(this.tentativeOutput.entries[partitionIndex].value);
392417

393-
foreach (var mapIndex in this.seenPartitions.entries[partitionIndex].value) this.seenEvent.Remove(mapIndex);
394-
395418
if (this.tentativeOutput.Count > 0)
396419
{
397420
tentativeVisibleTraverser.currIndex = 0;
@@ -438,8 +461,9 @@ private int EnsurePartition(TPartitionKey partitionKey)
438461
if (!this.lastSyncTime.Lookup(partitionKey, out int index))
439462
{
440463
index = this.lastSyncTime.Insert(partitionKey, long.MinValue);
441-
this.tentativeOutput.Insert(partitionKey, new FastMap<OutputEvent<TKey, TRegister>>());
442-
this.seenPartitions.Insert(partitionKey, new List<TKey>());
464+
465+
if (!this.IsSyncTimeSimultaneityFree)
466+
this.tentativeOutput.Insert(partitionKey, new FastMap<OutputEvent<TKey, TRegister>>());
443467
}
444468
return index;
445469
}

0 commit comments

Comments
 (0)