core: avoid per block allocation in aggregation (#1973)

The inner loop of MemoryDatabase.executeImpl runs once per block per
matched series, so for a broad query it can execute billions of times.
Three objects were being allocated on every iteration:

* The `newBuffer` argument was a local def. Passing a def where a
  function is expected eta-expands at the call site, which is inside the
  loop, so the captured cfStep/bufStart/bufEnd produced a new function
  object per block. Bind it to a val once per query instead.

* The inner foreach body ended with the Int returned by add(), so the
  loop body was typed as AnyVal and boxed a new Integer per block. The
  value is buffer.values.length, well above the Integer cache limit, so
  none of these were shared. Keep the body Unit typed.

* AggregateCollector.add iterated its blocks with foreach, allocating a
  closure that captured the buffer, op, and the aggr/cf/multiple params.
  Use an explicit loop over the list in both implementations.

After this the only remaining allocation in the per block frame is the
`List(b)` wrapper. Removing that means adding a single block overload to
the public AggregateCollector trait, left for a separate change.

Note that the four `Math.addNaN _` style eta-expansions in add() are NOT
a source of allocation, despite looking like one in a profile: Math is a
static module, so they compile to non-capturing invokedynamic call sites
that are bound to a singleton and are already specialized.

Author: brharrington

atlas-core/src/main/scala/com/netflix/atlas/core/db/AggregateCollector.scala

@@ -121,7 +121,12 @@ abstract class SimpleAggregateCollector extends AggregateCollector {
       case Block.Max   => Math.maxNaN _
     }
 
-    blocks.foreach { b =>
+    // Explicit loop rather than `foreach`: the closure would capture the buffer, op,
+    // and the aggr/cf/multiple params, allocating a new object on every call. This is
+    // called once per block, so it shows up as a top allocation leaf.
+    var bs = blocks
+    while (bs.nonEmpty) {
+      val b = bs.head
       if (valueMask != null) {
         val v = buffer.aggrBlock(b, aggr, ConsolidationFunction.Sum, multiple, op)
         buffer.valueMask(valueMask, b, multiple)
@@ -130,6 +135,7 @@ abstract class SimpleAggregateCollector extends AggregateCollector {
         val v = buffer.aggrBlock(b, aggr, cf, multiple, op)
         valueCount += v
       }
+      bs = bs.tail
     }
     buffer.values.length
   }
@@ -290,9 +296,12 @@ class AllAggregateCollector extends LimitedAggregateCollector {
       case Block.Max   => Math.maxNaN _
     }
 
-    blocks.foreach { b =>
-      val v = buffer.aggrBlock(b, aggr, cf, multiple, op)
-      valueCount += v
+    // Explicit loop to avoid allocating a capturing closure per call, see the comment
+    // on SimpleAggregateCollector.add.
+    var bs = blocks
+    while (bs.nonEmpty) {
+      valueCount += buffer.aggrBlock(bs.head, aggr, cf, multiple, op)
+      bs = bs.tail
     }
 
     if (valueCount > 0) add(buffer)

atlas-core/src/main/scala/com/netflix/atlas/core/db/MemoryDatabase.scala

@@ -229,7 +229,10 @@ class MemoryDatabase(registry: Registry, config: Config) extends Database {
     val bufStart = context.start
     val bufEnd = context.end - cfStep
 
-    def newBuffer(tags: Map[String, String]): TimeSeriesBuffer = {
+    // Bound to a val rather than a local def: passing a def where a function is expected
+    // eta-expands at the call site, and since this captures cfStep/bufStart/bufEnd that
+    // would allocate a new function object for every block. Bind it once per query.
+    val newBuffer: Map[String, String] => TimeSeriesBuffer = { tags =>
       TimeSeriesBuffer(tags, cfStep, bufStart, bufEnd)
     }
 
@@ -249,6 +252,9 @@ class MemoryDatabase(registry: Registry, config: Config) extends Database {
         if (b.start <= bufEnd && blockEnd > bufStart - cfStep) {
           numAggrBlocks += 1
           collector.add(item.tags, List(b), aggr, expr.cf, multiple, newBuffer)
+          // Keep the closure Unit typed. Otherwise the Int result of add() becomes the
+          // value of the loop body and gets boxed into an Integer for every block.
+          ()
         }
       }
     }