util-core: Improve performance and simplify Var.collects

Problem

`Var.collectIndependent` is pretty slow as it has to copy an entire result (array) on every update of every
individual var.

Solution

Use similar technic from Future.collect and avoid intermediate copies of the result by wrapping a mutable
array with an immutable interface.

Result

This also drops higher-kinded API for both Activity's and Var's collect methods. It can't be a generic CC[X]
anymore and has to be Seq, just like in Future.collect.

JIRA Issues: CSL-11377

Differential Revision: https://phabricator.twitter.biz/D795123

Author: vkostyukov

CHANGELOG.rst

@@ -7,6 +7,13 @@ Note that ``PHAB_ID=#`` and ``RB_ID=#`` correspond to associated messages in com
 Unreleased
 ----------
 
+Breaking API Changes
+~~~~~~~~~~~~~~~~~~~~
+* util-core: `Activity.collect*` and `Var.collect*` are now implemented in terms of known collection
+  type `scala.collection.Seq` versus HKT `CC[X]` before. This allows for certain performance
+  enhancements as well as makes it more aligned with the `Future.collect` APIs.
+  ``PHAB_ID=D795123``
+
 21.11.0
 -------
 

util-core/src/main/scala/com/twitter/util/Activity.scala

@@ -1,14 +1,11 @@
 package com.twitter.util
 
 import java.util.{List => JList}
-
+import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.Buffer
 import scala.jdk.CollectionConverters._
-import scala.language.higherKinds
-import scala.reflect.ClassTag
 import scala.util.control.NonFatal
-import scala.Iterable
-import scala.collection.compat._
+import scala.collection.{Seq => AnySeq}
 
 /**
  * Activity, like [[com.twitter.util.Var Var]], is a value that varies over
@@ -195,11 +192,7 @@ object Activity {
    *
    * @example def collect[T](activities: Coll[Activity[T]]): Activity[Coll[T]]
    */
-  def collect[T: ClassTag, CC[X] <: Iterable[X]](
-    acts: CC[Activity[T]]
-  )(
-    implicit factory: Factory[T, CC[T]]
-  ): Activity[CC[T]] = {
+  def collect[T](acts: AnySeq[Activity[T]]): Activity[Seq[T]] = {
     collect(acts, false)
   }
 
@@ -213,32 +206,26 @@ object Activity {
    *
    * @example def collectIndependent[T](activities: Coll[Activity[T]]): Activity[Coll[T]]
    */
-  def collectIndependent[T: ClassTag, CC[X] <: Iterable[X]](
-    acts: CC[Activity[T]]
-  )(
-    implicit factory: Factory[T, CC[T]]
-  ): Activity[CC[T]] = {
+  def collectIndependent[T](acts: AnySeq[Activity[T]]): Activity[Seq[T]] = {
     collect(acts, true)
   }
 
-  private[this] def collect[T: ClassTag, CC[X] <: Iterable[X]](
-    acts: CC[Activity[T]],
+  private[this] def collect[T](
+    acts: AnySeq[Activity[T]],
     collectIndependent: Boolean
-  )(
-    implicit factory: Factory[T, CC[T]]
-  ): Activity[CC[T]] = {
+  ): Activity[Seq[T]] = {
     if (acts.isEmpty)
-      return Activity.value(factory.fromSpecific(Nil))
+      return Activity.value(Seq.empty[T])
 
-    val states: Iterable[Var[State[T]]] = acts.map(_.run)
-    val stateVar: Var[Iterable[State[T]]] = if (collectIndependent) {
+    val states: AnySeq[Var[State[T]]] = acts.map(_.run)
+    val stateVar: Var[Seq[State[T]]] = if (collectIndependent) {
       Var.collectIndependent(states)
     } else {
       Var.collect(states)
     }
 
-    def flip(states: Iterable[State[T]]): State[CC[T]] = {
-      val notOk = states find {
+    def flip(states: AnySeq[State[T]]): State[Seq[T]] = {
+      val notOk = states.find {
         case Pending | Failed(_) => true
         case Ok(_) => false
       }
@@ -250,16 +237,16 @@ object Activity {
         case Some(_) => assert(false)
       }
 
-      val ts = factory.newBuilder
-      states foreach {
+      val ts = new ArrayBuffer[T](states.size)
+      states.foreach {
         case Ok(t) => ts += t
         case _ => assert(false)
       }
 
-      Ok(ts.result())
+      Ok(ts.toSeq)
     }
 
-    Activity(stateVar map flip)
+    Activity(stateVar.map(flip))
   }
 
   /**

util-core/src/main/scala/com/twitter/util/Var.scala

@@ -4,14 +4,13 @@ import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
 import java.util.{List => JList}
 import scala.annotation.tailrec
-import scala.collection.compat._
+import scala.collection.{Seq => AnySeq}
+import scala.collection.compat.immutable.ArraySeq
 import scala.collection.immutable
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.Buffer
 import scala.jdk.CollectionConverters._
 import scala.language.higherKinds
-import scala.reflect.ClassTag
-import scala.Iterable
 
 /**
  * Vars are values that vary over time. To create one, you must give it an
@@ -311,12 +310,8 @@ object Var {
    *  // refCollectIndependent == Vector((1,2), (2,2), (2,4))
    * }}}
    */
-  def collect[T: ClassTag, CC[X] <: Iterable[X]](
-    vars: CC[Var[T]]
-  )(
-    implicit factory: Factory[T, CC[T]]
-  ): Var[CC[T]] = {
-    val vs = vars.toArray
+  def collect[T](vars: AnySeq[Var[T]]): Var[Seq[T]] = {
+    val vs = vars.toIndexedSeq
 
     def tree(begin: Int, end: Int): Var[Seq[T]] =
       if (begin == end) Var(Seq.empty)
@@ -330,7 +325,7 @@ object Var {
         } yield left ++ right
       }
 
-    tree(0, vs.length).map(_.to(factory))
+    tree(0, vs.length)
   }
 
   /**
@@ -354,12 +349,8 @@ object Var {
    *  // refCollectIndependent == Vector((1,2), (2,2), (2,4))
    * }}}
    */
-  def collectIndependent[T: ClassTag, CC[X] <: Iterable[X]](
-    vars: CC[Var[T]]
-  )(
-    implicit factory: Factory[T, CC[T]]
-  ): Var[CC[T]] =
-    async(factory.newBuilder.result()) { u =>
+  def collectIndependent[T](vars: AnySeq[Var[T]]): Var[Seq[T]] =
+    async(Seq.empty[T]) { u =>
       val N = vars.size
 
       // `filling` represents whether or not we have gone through our collection
@@ -377,19 +368,25 @@ object Var {
       // @note "filling" only works with the guarantee that the initial `observe` is
       // synchronous. This should be the case with Vars since they have an initial value.
       @volatile var filling = true
-      val cur = new Array[T](N)
+      val cur = new Array[Any](N)
 
-      def publish(i: Int, newValue: T): Unit = cur.synchronized {
-        cur(i) = newValue
-        if (!filling) u() = cur.to(factory)
+      def publishAt(i: Int): T => Unit = { newValue =>
+        cur.synchronized {
+          cur(i) = newValue
+          if (!filling) {
+            u() = ArraySeq.unsafeWrapArray(cur).asInstanceOf[ArraySeq[T]]
+          }
+        }
       }
 
       val closables = new ArrayBuffer[Closable](N)
-      val iter = vars.iterator.zipWithIndex
+      var i = 0
+      val iter = vars.iterator
       while (iter.hasNext) {
-        val (v, i) = iter.next()
+        val v = iter.next()
         if (i == N - 1) filling = false
-        closables += v.observe(newValue => publish(i, newValue))
+        closables += v.observe(publishAt(i))
+        i += 1
       }
 
       Closable.all(closables.toSeq: _*)

util-core/src/test/scala/com/twitter/util/VarTest.scala

@@ -243,23 +243,6 @@ class VarTest extends AnyFunSuite with ScalaCheckDrivenPropertyChecks {
     }
   }
 
-  // This is either very neat or very horrendous,
-  // depending on your point of view.
-  test("Var.collect[Set]") {
-    val vars = Seq(Var(1), Var(2), Var(3))
-
-    val coll = Var.collect(vars.map { v => v: Var[Int] }.toSet)
-    val ref = new AtomicReference[Set[Int]]
-    coll.changes.register(Witness(ref))
-    assert(ref.get == Set(1, 2, 3))
-
-    vars(1).update(1)
-    assert(ref.get == Set(1, 3))
-
-    vars(1).update(999)
-    assert(ref.get == Set(1, 999, 3))
-  }
-
   test("Var.collect: ordering") {
     val v1 = Var(1)
     val v2 = v1.map(_ * 2)