remove BoundedQueue

summingbird-online/src/main/scala/com/twitter/summingbird/online/Channel.scala

@@ -0,0 +1,117 @@
+/*
+Copyright 2013 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.summingbird.online
+
+import com.twitter.util.{ Await, Duration, Future, Try }
+
+import java.util.Queue
+import java.util.concurrent.{ArrayBlockingQueue, BlockingQueue, LinkedBlockingQueue, TimeUnit}
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicInteger
+/**
+ *
+ * @author Oscar Boykin
+ */
+
+object Channel {
+  /**
+   * By default, don't block on put
+   */
+  def apply[T]() = linkedNonBlocking[T]
+
+  def arrayBlocking[T](size: Int): Channel[T] =
+    new Channel[T](new ArrayBlockingQueue(size))
+
+  def linkedBlocking[T]: Channel[T] =
+    new Channel[T](new LinkedBlockingQueue())
+
+  def linkedNonBlocking[T]: Channel[T] =
+    new Channel[T](new ConcurrentLinkedQueue())
+}
+
+/**
+ * Use this class with a thread-safe queue to receive
+ * results from futures in one thread.
+ * Storm needs us to touch it's code in one event path (via
+ * the execute method in bolts)
+ */
+class Channel[T] private (queue: Queue[T]) {
+
+  private val count = new AtomicInteger(0)
+
+  def put(item: T): Int = {
+    queue.add(item)
+    count.incrementAndGet
+  }
+
+  /** Returns the size immediately after the put */
+  def putAll(items: TraversableOnce[T]): Int = {
+    val added = items.foldLeft(0) { (cnt, item) =>
+      queue.add(item)
+      cnt + 1
+    }
+    count.addAndGet(added)
+  }
+
+  /**
+   * check if something is ready now
+   */
+  def poll: Option[T] = Option(queue.poll())
+
+  /**
+   * Obviously, this might not be the same by the time you
+   * call spill
+   */
+  def size: Int = count.get
+
+  // Do something on all the elements ready:
+  @annotation.tailrec
+  final def foreach(fn: T => Unit): Unit =
+    queue.poll() match {
+      case null => ()
+      case itt => fn(itt); foreach(fn)
+    }
+
+  // fold on all the elements ready:
+  @annotation.tailrec
+  final def foldLeft[V](init: V)(fn: (V, T) => V): V = {
+    queue.poll() match {
+      case null => init
+      case itt => foldLeft(fn(init, itt))(fn)
+    }
+  }
+
+  /**
+   * Take enough elements to get the queue under the maxLength
+   */
+  def trimTo(maxLength: Int): Seq[T] = {
+    require(maxLength >= 0, "maxLength must be >= 0.")
+
+    @annotation.tailrec
+    def loop(size: Int, acc: List[T] = Nil): List[T] = {
+      if(size > maxLength) {
+        queue.poll match {
+          case null => acc.reverse // someone else cleared us out
+          case item =>
+            loop(count.decrementAndGet, item::acc)
+        }
+      }
+      else acc.reverse
+    }
+    loop(count.get)
+  }
+}

summingbird-online/src/test/scala/com/twitter/summingbird/online/QueueChannelLaws.scala

@@ -21,53 +21,53 @@ import Gen._
 import Arbitrary._
 import org.scalacheck.Prop._
 
-import com.twitter.util.{Return, Throw, Future}
+import com.twitter.util.{Return, Throw, Future, Try}
 
-object QueueChannelLaws extends Properties("Queue and Channel") {
+object QueueChannelLaws extends Properties("Channel") {
 
   property("Putting into a BoundedQueue gets size right") = forAll { (items: List[String]) =>
-    val q = new BoundedQueue[String](10)
+    val q = Channel[String]()
     q.putAll(items)
     q.size == items.size
   }
   property("not spill if capacity is enough") = forAll { (items: List[Int]) =>
-    val q = new BoundedQueue[Int](items.size)
+    val q = Channel[Int]()
     q.putAll(items)
-    q.spill.size == 0
+    q.trimTo(items.size).size == 0
   }
   property("Work with indepent additions") = forAll { (items: List[Int]) =>
-    val q = new BoundedQueue[Int](items.size)
+    val q = Channel[Int]()
     items.map(q.put(_)) == (1 to items.size).toList
   }
   property("spill all with zero capacity") = forAll { (items: List[Int]) =>
-    val q = new BoundedQueue[Int](0)
+    val q = Channel[Int]()
     q.putAll(items)
-    q.spill == items
+    q.trimTo(0) == items
   }
-  property("FutureChannel works with finished futures") = forAll { (items: List[Int]) =>
-    val q = FutureChannel.linkedBlocking[Int,Int]
-    items.foreach { i => q.put(i, Future(i*i)) }
+  property("Channel works with finished futures") = forAll { (items: List[Int]) =>
+    val q = Channel.linkedBlocking[(Int,Try[Int])]
+    items.foreach { i => q.put((i, Try(i*i))) }
     q.foldLeft((0, true)) { case ((cnt, good), (i, ti)) =>
       ti match {
         case Return(ii) => (cnt + 1, good)
         case Throw(e) => (cnt + 1, false)
       }
     } == (items.size, true)
   }
-  property("FutureChannel.linkedNonBlocking works with finished futures") = forAll { (items: List[Int]) =>
-    val q = FutureChannel.linkedNonBlocking[Int,Int]
-    items.foreach { i => q.put(i, Future(i*i)) }
+  property("Channel.linkedNonBlocking works") = forAll { (items: List[Int]) =>
+    val q = Channel.linkedNonBlocking[(Int,Try[Int])]
+    items.foreach { i => q.put((i, Try(i*i))) }
     q.foldLeft((0, true)) { case ((cnt, good), (i, ti)) =>
       ti match {
         case Return(ii) => (cnt + 1, good)
         case Throw(e) => (cnt + 1, false)
       }
     } == (items.size, true)
   }
-  property("FutureChannel foreach works") = forAll { (items: List[Int]) =>
+  property("Channel foreach works") = forAll { (items: List[Int]) =>
     // Make sure we can fit everything
-    val q = FutureChannel.arrayBlocking[Int,Int](items.size + 1)
-    items.foreach { q.call(_) { i => Future(i*i) } }
+    val q = Channel.arrayBlocking[(Int,Try[Int])](items.size + 1)
+    items.foreach { i => q.put((i,Try(i*i))) }
     var works = true
     q.foreach { case (i, Return(ii)) =>
       works = works && (ii == i*i)

summingbird-storm/src/main/scala/com/twitter/summingbird/storm/BaseBolt.scala

@@ -25,9 +25,9 @@ import java.util.{ Map => JMap, Arrays => JArrays, List => JList, ArrayList => J
 import com.twitter.summingbird.batch.Timestamp
 import com.twitter.summingbird.storm.option.{AnchorTuples, MaxWaitingFutures}
 
-import com.twitter.summingbird.online.{BoundedQueue, FutureChannel}
+import com.twitter.summingbird.online.Channel
 
-import com.twitter.util.{Await, Future, Return, Throw}
+import com.twitter.util.{Await, Future, Return, Throw, Try}
 import scala.collection.JavaConverters._
 
 import org.slf4j.{LoggerFactory, Logger}
@@ -109,8 +109,8 @@ abstract class AsyncBaseBolt[I, O](metrics: () => TraversableOnce[StormMetric[_]
    */
   def apply(tup: Tuple, in: (Timestamp, I)): Future[Iterable[(JList[Tuple], Future[TraversableOnce[(Timestamp, O)]])]]
 
-  private lazy val futureQueue = new BoundedQueue[Future[Unit]](maxWaitingFutures.get)
-  private lazy val futureChannel = FutureChannel[JList[Tuple], TraversableOnce[(Timestamp, O)]]()
+  private lazy val futureQueue = Channel[Future[Unit]]()
+  private lazy val channel = Channel[(JList[Tuple], Try[TraversableOnce[(Timestamp, O)]])]()
 
   override def execute(tuple: Tuple) {
     /**
@@ -124,7 +124,7 @@ abstract class AsyncBaseBolt[I, O](metrics: () => TraversableOnce[StormMetric[_]
         .onSuccess { iter =>
           // Collect the result onto our channel
           val (puts, maxSize) = iter.foldLeft((0, 0)) { case ((p, ms), (tups, res)) =>
-            futureChannel.put(tups, res)
+            res.respond { t => channel.put((tups, t)) }
             // Make sure there are not too many outstanding:
             val count = futureQueue.put(res.unit)
             (p + 1, ms max count)
@@ -150,15 +150,15 @@ abstract class AsyncBaseBolt[I, O](metrics: () => TraversableOnce[StormMetric[_]
   }
 
   protected def forceExtraFutures {
-    val toForce = futureQueue.spill
+    val toForce = futureQueue.trimTo(maxWaitingFutures.get)
     if(!toForce.isEmpty) Await.result(Future.collect(toForce))
   }
 
   protected def emptyChannel = {
     // don't let too many futures build up
     forceExtraFutures
     // Handle all ready results now:
-    futureChannel.foreach { case (tups, res) =>
+    channel.foreach { case (tups, res) =>
       res match {
         case Return(outs) => finish(tups, outs)
         case Throw(t) => fail(tups, t)