Feature/migrate async summers from sb

algebird-util/src/main/scala/com/twitter/algebird/util/summer/AsyncListSum.scala

@@ -0,0 +1,103 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.algebird._
+import com.twitter.util.{Duration, Future, FuturePool}
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.JavaConverters._
+import scala.collection.mutable.{Set => MSet}
+/**
+ * @author Ian O Connell
+ */
+
+class AsyncListSum[Key, Value](bufferSize: BufferSize,
+                                          override val flushFrequency: FlushFrequency,
+                                          override val softMemoryFlush: MemoryFlushPercent,
+                                          workPool: FuturePool)
+                                         (implicit semigroup: Semigroup[Value])
+                                          extends AsyncSummer[(Key, Value), Map[Key, Value]]
+                                          with WithFlushConditions[(Key, Value), Map[Key, Value]] {
+
+  require(bufferSize.v > 0, "Use the Null summer for an empty async summer")
+
+  protected override val emptyResult = Map.empty[Key, Value]
+
+  private[this] final val queueMap = new ConcurrentHashMap[Key, IndexedSeq[Value]]()
+  private[this] final val elementsInCache = new AtomicInteger(0)
+
+  override def isFlushed: Boolean = elementsInCache.get == 0
+
+  override def flush: Future[Map[Key, Value]] =
+    workPool {
+      didFlush // bumps timeout on the flush conditions
+      // Take a copy of the keyset into a scala set (toSet forces the copy)
+      // We want this to be safe around uniqueness in the keys coming out of the keys.flatMap
+      val keys = queueMap.keySet.asScala.toSet
+      keys.flatMap { k =>
+        val retV = queueMap.remove(k)
+        if(retV != null) {
+          val newRemaining = elementsInCache.addAndGet(retV.size * -1)
+          Semigroup.sumOption(retV).map(v => (k, v))
+        }
+        else None
+      }.toMap
+    }
+
+  @annotation.tailrec
+  private[this] final def doInsert(key: Key, vals: IndexedSeq[Value]) {
+    require(key != null, "Key can not be null")
+    val success = if (queueMap.containsKey(key)) {
+      val oldValue = queueMap.get(key)
+      if(oldValue != null) {
+        val newValue = vals ++ oldValue
+        queueMap.replace(key, oldValue, newValue)
+      } else {
+        false // Removed between the check above and fetching
+      }
+    } else {
+      // Test if something else has raced into our spot.
+      queueMap.putIfAbsent(key, vals) == null
+    }
+
+    if(success) {
+      // Successful insert
+      elementsInCache.addAndGet(vals.size)
+    } else {
+      return doInsert(key, vals)
+    }
+  }
+
+  def addAll(vals: TraversableOnce[(Key, Value)]): Future[Map[Key, Value]] = {
+    val prepVals = vals.map { case (k, v) =>
+      require(k != null, "Inserting a null key?")
+      (k -> IndexedSeq(v))
+    } : TraversableOnce[(Key, IndexedSeq[Value])]
+
+    val curData = MapAlgebra.sumByKey(prepVals)
+
+    curData.foreach { case (k, v) =>
+      doInsert(k, v)
+    }
+
+    if(elementsInCache.get >= bufferSize.v) {
+      flush
+    } else {
+      Future.value(Map.empty)
+    }
+  }
+}

algebird-util/src/main/scala/com/twitter/algebird/util/summer/AsyncMapSum.scala

@@ -0,0 +1,63 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.algebird._
+import com.twitter.util.{Duration, Future, FuturePool}
+import java.util.concurrent.ArrayBlockingQueue
+import scala.collection.mutable.ListBuffer
+import scala.collection.JavaConverters._
+
+/**
+ * @author Ian O Connell
+ */
+
+class AsyncMapSum[Key, Value](bufferSize: BufferSize,
+                                          override val flushFrequency: FlushFrequency,
+                                          override val softMemoryFlush: MemoryFlushPercent,
+                                          workPool: FuturePool)
+                                         (implicit semigroup: Semigroup[Value])
+                                          extends AsyncSummer[(Key, Value), Map[Key, Value]]
+                                          with WithFlushConditions[(Key, Value), Map[Key, Value]] {
+
+  require(bufferSize.v > 0, "Use the Null summer for an empty async summer")
+
+  protected override val emptyResult = Map.empty[Key, Value]
+
+  private[this] final val queue = new ArrayBlockingQueue[Map[Key, Value]](bufferSize.v, true)
+  override def isFlushed: Boolean = queue.size == 0
+
+  override def flush: Future[Map[Key, Value]] = {
+    didFlush // bumps timeout on the flush conditions
+    val toSum = ListBuffer[Map[Key, Value]]()
+    queue.drainTo(toSum.asJava)
+    workPool {
+      Semigroup.sumOption(toSum).getOrElse(Map.empty)
+    }
+  }
+
+  def addAll(vals: TraversableOnce[(Key, Value)]): Future[Map[Key, Value]] = {
+    val curData = Semigroup.sumOption(vals.map(Map(_))).getOrElse(Map.empty)
+    if(!queue.offer(curData)) {
+      flush.map { flushRes =>
+        Semigroup.plus(flushRes, curData)
+      }
+    }
+    else {
+      Future.value(Map.empty)
+    }
+  }
+}

algebird-util/src/main/scala/com/twitter/algebird/util/summer/AsyncSummer.scala

@@ -0,0 +1,80 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.algebird._
+import com.twitter.util.{Duration, Future}
+
+/**
+ * @author Ian O Connell
+ */
+
+
+trait AsyncSummer[T, M <: Iterable[T]] { self =>
+  def flush: Future[M]
+  def tick: Future[M]
+  def add(t: T) = addAll(Iterator(t))
+  def addAll(vals: TraversableOnce[T]): Future[M]
+
+  def isFlushed: Boolean
+  def cleanup: Future[Unit] = Future.Unit
+  def withCleanup(cleanup: () => Future[Unit]): AsyncSummer[T, M] = {
+    val oldSelf = self
+    new AsyncSummerProxy[T, M] {
+      override val self = oldSelf
+      override def cleanup = {
+        oldSelf.cleanup.flatMap { _ => cleanup }
+      }
+    }
+  }
+}
+
+trait AsyncSummerProxy[T, M <: Iterable[T]] extends AsyncSummer[T, M] {
+  def self: AsyncSummer[T, M]
+  def flush = self.flush
+  def tick = self.tick
+  override def add(t: T) = self.add(t)
+  def addAll(vals: TraversableOnce[T]) = self.addAll(vals)
+  def isFlushed = self.isFlushed
+  override def cleanup: Future[Unit] = self.cleanup
+}
+
+
+private[summer] trait WithFlushConditions[T, M <: Iterable[T]] extends AsyncSummer[T, M] {
+  protected var lastDump:Long = System.currentTimeMillis
+  protected def softMemoryFlush: MemoryFlushPercent
+  protected def flushFrequency: FlushFrequency
+  protected def emptyResult: M
+
+  protected def timedOut = (System.currentTimeMillis - lastDump) >= flushFrequency.v.inMilliseconds
+  protected lazy val runtime  = Runtime.getRuntime
+
+  protected def didFlush {lastDump = System.currentTimeMillis}
+
+  protected def memoryWaterMark = {
+    val used = ((runtime.totalMemory - runtime.freeMemory).toDouble * 100) / runtime.maxMemory
+    used > softMemoryFlush.v
+  }
+
+  def tick: Future[M] = {
+    if (timedOut || memoryWaterMark) {
+          flush
+      }
+    else {
+      Future.value(emptyResult)
+    }
+  }
+}

algebird-util/src/main/scala/com/twitter/algebird/util/summer/BufferSize.scala

@@ -0,0 +1,24 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.util.Duration
+
+/**
+ * @author Ian O Connell
+ */
+
+case class BufferSize(v: Int)

algebird-util/src/main/scala/com/twitter/algebird/util/summer/FlushFrequency.scala

@@ -0,0 +1,24 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.util.Duration
+
+/**
+ * @author Ian O Connell
+ */
+
+case class FlushFrequency(v: Duration)

algebird-util/src/main/scala/com/twitter/algebird/util/summer/MemoryFlushPercent.scala

@@ -0,0 +1,23 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+
+/**
+ * @author Ian O Connell
+ */
+
+case class MemoryFlushPercent(v: Float)

algebird-util/src/main/scala/com/twitter/algebird/util/summer/NullSummer.scala

@@ -0,0 +1,33 @@
+/*
+Copyright 2012 Twitter, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package com.twitter.algebird.util.summer
+
+import com.twitter.algebird._
+import com.twitter.util.Future
+
+/**
+ * @author Ian O Connell
+ */
+
+class NullSummer[Key, Value](implicit semigroup: Semigroup[Value])
+                                          extends AsyncSummer[(Key, Value), Map[Key, Value]] {
+  def flush: Future[Map[Key, Value]] = Future.value(Map.empty)
+  def tick: Future[Map[Key, Value]] = Future.value(Map.empty)
+  def addAll(vals: TraversableOnce[(Key, Value)]): Future[Map[Key, Value]] =
+    Future.value(Semigroup.sumOption(vals.map(Map(_))).getOrElse(Map.empty))
+  override val isFlushed: Boolean = true
+}
+