Skip to content

Commit c9d69c4

Browse files
peterfreilingpeterfreiling
authored
Fixing Afa operator stale pointers (#141)
1 parent 7cf3f73 commit c9d69c4

20 files changed

Lines changed: 361 additions & 58 deletions

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
// ------------------------------------------------------------------------------
22
// <auto-generated>
33
// This code was generated by a tool.
4-
// Runtime Version: 15.0.0.0
4+
// Runtime Version: 16.0.0.0
55
//
66
// Changes to this file may cause incorrect behavior and will be lost if
77
// the code is regenerated.
@@ -15,7 +15,7 @@ namespace Microsoft.StreamProcessing
1515
/// <summary>
1616
/// Class to produce the template output
1717
/// </summary>
18-
[global::System.CodeDom.Compiler.GeneratedCodeAttribute("Microsoft.VisualStudio.TextTemplating", "15.0.0.0")]
18+
[global::System.CodeDom.Compiler.GeneratedCodeAttribute("Microsoft.VisualStudio.TextTemplating", "16.0.0.0")]
1919
internal partial class AfaMultiEventListTemplate : AfaTemplate
2020
{
2121
/// <summary>
@@ -564,6 +564,10 @@ private void ProcessCurrentTimestamp()
564564
if (src_vother[i] < 0)
565565
{
566566
OnPunctuation(synctime);
567+
this.dest_vsync = this.batch.vsync.col;
568+
this.dest_vother = this.batch.vother.col;
569+
this.destkey = this.batch.key.col;
570+
this.dest_hash = this.batch.hash.col;
567571
continue;
568572
}
569573
");

‎Sources/Core/Microsoft.StreamProcessing/Operators/Afa/AfaMultiEventListTemplate.tt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -493,6 +493,10 @@ internal sealed class <#= className #> : CompiledAfaPipeBase<<#= TKey #>, <#= TP
493493
if (src_vother[i] < 0)
494494
{
495495
OnPunctuation(synctime);
496+
this.dest_vsync = this.batch.vsync.col;
497+
this.dest_vother = this.batch.vother.col;
498+
this.destkey = this.batch.key.col;
499+
this.dest_hash = this.batch.hash.col;
496500
continue;
497501
}
498502
<# if (grouped) { #>

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,8 @@ protected virtual void IfFinalStateProduceOutput(int indent, int node, string in
4545

4646
this.Write(this.ToStringHelper.ToStringWithCulture(otherTime));
4747

48-
this.Write(" + MaxDuration, StreamEvent.InfinitySyncTime);;\r\ntentativeOutput.Values[ind].key " +
49-
"= ");
48+
this.Write(" + MaxDuration, StreamEvent.InfinitySyncTime);\r\ntentativeOutput.Values[ind].key =" +
49+
" ");
5050

5151
this.Write(this.ToStringHelper.ToStringWithCulture(keyValue));
5252

@@ -65,7 +65,7 @@ protected virtual void IfFinalStateProduceOutput(int indent, int node, string in
6565

6666
this.Write(this.ToStringHelper.ToStringWithCulture(otherTime));
6767

68-
this.Write(" + MaxDuration, StreamEvent.InfinitySyncTime);;\r\n");
68+
this.Write(" + MaxDuration, StreamEvent.InfinitySyncTime);\r\n");
6969

7070

7171
if (this.noPublicResultFields) {

‎Sources/Core/Microsoft.StreamProcessing/Operators/Afa/AfaTemplate.tt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ protected virtual void IfFinalStateProduceOutput(int indent, int node, string in
1414
{
1515
#>
1616
int ind = tentativeOutput.Insert(<#= insertArgument #>);
17-
tentativeOutput.Values[ind].other = Math.Min(<#= otherTime #> + MaxDuration, StreamEvent.InfinitySyncTime);;
17+
tentativeOutput.Values[ind].other = Math.Min(<#= otherTime #> + MaxDuration, StreamEvent.InfinitySyncTime);
1818
tentativeOutput.Values[ind].key = <#= keyValue #>;
1919
tentativeOutput.Values[ind].payload = newReg;
2020
<#+
@@ -23,7 +23,7 @@ tentativeOutput.Values[ind].payload = newReg;
2323
{
2424
#>
2525
dest_vsync[iter] = <#= syncTime #>;
26-
dest_vother[iter] = Math.Min(<#= otherTime #> + MaxDuration, StreamEvent.InfinitySyncTime);;
26+
dest_vother[iter] = Math.Min(<#= otherTime #> + MaxDuration, StreamEvent.InfinitySyncTime);
2727
<#+
2828
if (this.noPublicResultFields) {
2929
#>

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

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -162,9 +162,7 @@ private void CompileAfa(Afa<TPayload, TRegister, TAccumulator> afa)
162162
new MultiEventArcInfo<TPayload, TRegister, TAccumulator>
163163
{
164164
toState = to,
165-
Initialize = mearc.Initialize != null
166-
? mearc.Initialize.Compile()
167-
: (ts, reg) => this.defaultAccumulator,
165+
Initialize = mearc.Initialize?.Compile() ?? (Func<long, TRegister, TAccumulator>)((ts, reg) => this.defaultAccumulator),
168166
Accumulate = mearc.Accumulate != null ? mearc.Accumulate.Compile() : (ts, ev, reg, acc) => acc,
169167
SkipToEnd = mearc.SkipToEnd?.Compile(),
170168
Dispose = mearc.Dispose?.Compile(),

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -290,7 +290,6 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
290290
dest_vother = this.batch.vother.col;
291291
destkey = this.batch.key.col;
292292
dest_hash = this.batch.hash.col;
293-
294293
}
295294
}
296295
}
@@ -343,15 +342,29 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
343342
this.batch.hash.col[this.iter] = hash;
344343
this.iter++;
345344

346-
if (this.iter == Config.DataBatchSize) FlushContents();
345+
if (this.iter == Config.DataBatchSize)
346+
{
347+
FlushContents();
348+
dest_vsync = this.batch.vsync.col;
349+
dest_vother = this.batch.vother.col;
350+
destkey = this.batch.key.col;
351+
dest_hash = this.batch.hash.col;
352+
}
347353
}
348354

349355
this.tentativeOutput.Clear(); // Clear the tentative output list
350356
}
351357

352358
this.lastSyncTime = synctime;
353359
}
360+
361+
// Update dest_* on punctuation in case this event will hit the batch boundary and allocate a new batch
354362
OnPunctuation(synctime);
363+
364+
dest_vsync = this.batch.vsync.col;
365+
dest_vother = this.batch.vother.col;
366+
destkey = this.batch.key.col;
367+
dest_hash = this.batch.hash.col;
355368
}
356369
}
357370
}

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

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -300,7 +300,6 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
300300
dest_vother = this.batch.vother.col;
301301
destkey = this.batch.key.col;
302302
dest_hash = this.batch.hash.col;
303-
304303
}
305304
}
306305
}
@@ -387,7 +386,14 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
387386
this.batch.hash.col[this.iter] = hash;
388387
this.iter++;
389388

390-
if (this.iter == Config.DataBatchSize) FlushContents();
389+
if (this.iter == Config.DataBatchSize)
390+
{
391+
FlushContents();
392+
dest_vsync = this.batch.vsync.col;
393+
dest_vother = this.batch.vother.col;
394+
destkey = this.batch.key.col;
395+
dest_hash = this.batch.hash.col;
396+
}
391397
}
392398

393399
// Remove the partition
@@ -398,7 +404,13 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
398404
}
399405
}
400406

407+
// Update dest_* on low watermark in case this event will hit the batch boundary and allocate a new batch
401408
OnLowWatermark(synctime);
409+
410+
dest_vsync = this.batch.vsync.col;
411+
dest_vother = this.batch.vother.col;
412+
destkey = this.batch.key.col;
413+
dest_hash = this.batch.hash.col;
402414
}
403415
else if (src_vother[i] == PartitionedStreamEvent.PunctuationOtherTime)
404416
{
@@ -430,7 +442,14 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
430442
this.batch.hash.col[this.iter] = hash;
431443
this.iter++;
432444

433-
if (this.iter == Config.DataBatchSize) FlushContents();
445+
if (this.iter == Config.DataBatchSize)
446+
{
447+
FlushContents();
448+
dest_vsync = this.batch.vsync.col;
449+
dest_vother = this.batch.vother.col;
450+
destkey = this.batch.key.col;
451+
dest_hash = this.batch.hash.col;
452+
}
434453
}
435454

436455
this.tentativeOutput.entries[partitionIndex].value.Clear(); // Clear the tentative output list
@@ -448,7 +467,14 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
448467
this.batch.bitvector.col[this.iter >> 6] |= (1L << (this.iter & 0x3f));
449468
this.iter++;
450469

451-
if (this.iter == Config.DataBatchSize) FlushContents();
470+
if (this.iter == Config.DataBatchSize)
471+
{
472+
FlushContents();
473+
dest_vsync = this.batch.vsync.col;
474+
dest_vother = this.batch.vother.col;
475+
destkey = this.batch.key.col;
476+
dest_hash = this.batch.hash.col;
477+
}
452478
}
453479
}
454480
}

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

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -314,7 +314,14 @@ public override unsafe void OnNext(StreamMessage<Empty, TPayload> batch)
314314
this.batch.hash.col[this.iter] = 0;
315315
this.iter++;
316316

317-
if (this.iter == Config.DataBatchSize) FlushContents();
317+
if (this.iter == Config.DataBatchSize)
318+
{
319+
FlushContents();
320+
dest_vsync = this.batch.vsync.col;
321+
dest_vother = this.batch.vother.col;
322+
destkey = this.batch.key.col;
323+
dest_hash = this.batch.hash.col;
324+
}
318325
}
319326

320327
this.tentativeOutput.Clear(); // Clear the tentative output list
@@ -323,7 +330,14 @@ public override unsafe void OnNext(StreamMessage<Empty, TPayload> batch)
323330
this.lastSyncTime = synctime;
324331
}
325332
}
333+
334+
// Update dest_* on punctuation in case this event will hit the batch boundary and allocate a new batch
326335
OnPunctuation(synctime);
336+
337+
dest_vsync = this.batch.vsync.col;
338+
dest_vother = this.batch.vother.col;
339+
destkey = this.batch.key.col;
340+
dest_hash = this.batch.hash.col;
327341
}
328342
}
329343
}

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

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -279,7 +279,14 @@ public override unsafe void OnNext(StreamMessage<Empty, TPayload> batch)
279279
this.batch.hash.col[this.iter] = 0;
280280
this.iter++;
281281

282-
if (this.iter == Config.DataBatchSize) FlushContents();
282+
if (this.iter == Config.DataBatchSize)
283+
{
284+
FlushContents();
285+
dest_vsync = this.batch.vsync.col;
286+
dest_vother = this.batch.vother.col;
287+
destkey = this.batch.key.col;
288+
dest_hash = this.batch.hash.col;
289+
}
283290
}
284291

285292
this.tentativeOutput.Clear(); // Clear the tentative output list
@@ -288,7 +295,14 @@ public override unsafe void OnNext(StreamMessage<Empty, TPayload> batch)
288295
this.lastSyncTime = synctime;
289296
}
290297
}
298+
299+
// Update dest_* on punctuation in case this event will hit the batch boundary and allocate a new batch
291300
OnPunctuation(synctime);
301+
302+
dest_vsync = this.batch.vsync.col;
303+
dest_vother = this.batch.vother.col;
304+
destkey = this.batch.key.col;
305+
dest_hash = this.batch.hash.col;
292306
}
293307
}
294308
}

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

Lines changed: 23 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -251,26 +251,29 @@ private void ProcessCurrentTimestamp()
251251
this.Write(this.ToStringHelper.ToStringWithCulture(TKey));
252252
this.Write(", ");
253253
this.Write(this.ToStringHelper.ToStringWithCulture(TPayload));
254-
this.Write("> batch)\r\n {\r\n dest_vsync = this.batch.vsync.col;\r\n dest_vother " +
255-
"= this.batch.vother.col;\r\n destkey = this.batch.key.col;\r\n dest_ha" +
256-
"sh = this.batch.hash.col;\r\n\r\n var count = batch.Count;\r\n\r\n var src" +
257-
"key = batch.key.col;\r\n\r\n fixed (long* src_bv = batch.bitvector.col, src_v" +
258-
"sync = batch.vsync.col, src_vother = batch.vother.col)\r\n {\r\n f" +
259-
"ixed (int* src_hash = batch.hash.col)\r\n {\r\n for (int i" +
260-
" = 0; i < count; i++)\r\n {\r\n if ((src_bv[i >> 6" +
261-
"] & (1L << (i & 0x3f))) == 0 || src_vother[i] < 0)\r\n {\r\n " +
262-
" long synctime = src_vsync[i];\r\n\r\n if (" +
263-
"synctime > lastSyncTime) // move time forward\r\n {\r\n " +
264-
" ProcessCurrentTimestamp();\r\n la" +
265-
"stSyncTime = synctime;\r\n }\r\n\r\n if " +
266-
"(src_vother[i] < 0)\r\n {\r\n OnPu" +
267-
"nctuation(synctime);\r\n continue;\r\n " +
268-
" }\r\n\r\n bool done = false;\r\n in" +
269-
"t index, hash;\r\n\r\n if (eventListTraverser.Find(src_hash[i" +
270-
"]))\r\n {\r\n while (eventListTrav" +
271-
"erser.Next(out index))\r\n {\r\n " +
272-
" var state = currentTimestampEventList.Values[index];\r\n\r\n " +
273-
" if (");
254+
this.Write("> batch)\r\n {\r\n this.dest_vsync = this.batch.vsync.col;\r\n this.de" +
255+
"st_vother = this.batch.vother.col;\r\n this.destkey = this.batch.key.col;\r\n" +
256+
" this.dest_hash = this.batch.hash.col;\r\n\r\n var count = batch.Count" +
257+
";\r\n\r\n var srckey = batch.key.col;\r\n\r\n fixed (long* src_bv = batch." +
258+
"bitvector.col, src_vsync = batch.vsync.col, src_vother = batch.vother.col)\r\n " +
259+
" {\r\n fixed (int* src_hash = batch.hash.col)\r\n {\r\n " +
260+
" for (int i = 0; i < count; i++)\r\n {\r\n " +
261+
" if ((src_bv[i >> 6] & (1L << (i & 0x3f))) == 0 || src_vother[i] < 0)\r\n " +
262+
" {\r\n long synctime = src_vsync[i];\r\n\r\n " +
263+
" if (synctime > lastSyncTime) // move time forward\r\n " +
264+
" {\r\n ProcessCurrentTimestamp();\r\n " +
265+
" lastSyncTime = synctime;\r\n }\r\n\r\n " +
266+
" if (src_vother[i] < 0)\r\n {\r\n " +
267+
" OnPunctuation(synctime);\r\n this.dest_" +
268+
"vsync = this.batch.vsync.col;\r\n this.dest_vother = th" +
269+
"is.batch.vother.col;\r\n this.destkey = this.batch.key." +
270+
"col;\r\n this.dest_hash = this.batch.hash.col;\r\n " +
271+
" continue;\r\n }\r\n\r\n " +
272+
" bool done = false;\r\n int index, hash;\r\n\r\n " +
273+
" if (eventListTraverser.Find(src_hash[i]))\r\n " +
274+
"{\r\n while (eventListTraverser.Next(out index))\r\n " +
275+
" {\r\n var state = currentTim" +
276+
"estampEventList.Values[index];\r\n\r\n if (");
274277
this.Write(this.ToStringHelper.ToStringWithCulture(keyEqualityComparer("state.key", "srckey[i]")));
275278
this.Write(@")
276279
{

0 commit comments

Comments
 (0)