Fix catchup with limit, optionally support via config

Author: HaloFour

src/main/scala/com/comcast/xfinity/sirius/api/SiriusConfiguration.scala

@@ -298,6 +298,16 @@ object SiriusConfiguration {
    */
   final val CATCHUP_MAX_WINDOW_SIZE = "sirius.catchup.max-window-size"
 
+  /**
+   * Use log limit for congestion avoidance for catchup.  Default is false.
+   */
+  final val CATCHUP_USE_LIMIT = "sirius.catchup.use-limit"
+
+  /**
+   * Maximum catchup limit size, in number of events. Default is 1000.
+   */
+  final val CATCHUP_MAX_LIMIT_SIZE = "sirius.catchup.max-limit-size"
+
   /**
    * Starting ssthresh, which is the point where catchup transitions from Slow Start to
    * Congestion Avoidance. Default is 500.

src/main/scala/com/comcast/xfinity/sirius/api/impl/bridge/CatchupSupervisor.scala

@@ -38,7 +38,7 @@ object CatchupSupervisor {
 
   trait CatchupSupervisorInfoMBean {
     def getSSThresh: Int
-    def getWindow: Int
+    def getLimit: Int
   }
 
   /**
@@ -52,9 +52,14 @@ object CatchupSupervisor {
     val timeoutCoeff = config.getDouble(SiriusConfiguration.CATCHUP_TIMEOUT_INCREASE_PER_EVENT, .01)
     val timeoutConst = config.getDouble(SiriusConfiguration.CATCHUP_TIMEOUT_BASE, 1.0)
     val maxWindowSize = config.getInt(SiriusConfiguration.CATCHUP_MAX_WINDOW_SIZE, 1000)
+    val maxLimitSize = if (config.getProp(SiriusConfiguration.CATCHUP_USE_LIMIT, false)) {
+      config.getProp[Int](SiriusConfiguration.CATCHUP_MAX_LIMIT_SIZE)
+    } else {
+      None
+    }
     // must ensure ssthresh <= maxWindowSize
-    val startingSSThresh = Math.min(maxWindowSize, config.getInt(SiriusConfiguration.CATCHUP_DEFAULT_SSTHRESH, 500))
-    Props(new CatchupSupervisor(membershipHelper, timeoutCoeff, timeoutConst, maxWindowSize, startingSSThresh, config))
+    val startingSSThresh = Math.min(maxLimitSize.getOrElse(maxWindowSize), config.getInt(SiriusConfiguration.CATCHUP_DEFAULT_SSTHRESH, 500))
+    Props(new CatchupSupervisor(membershipHelper, timeoutCoeff, timeoutConst, maxWindowSize, maxLimitSize, startingSSThresh, config))
   }
 }
 
@@ -78,51 +83,56 @@ private[bridge] class CatchupSupervisor(membershipHelper: MembershipHelper,
                                         timeoutCoeff: Double,
                                         timeoutConst: Double,
                                         maxWindowSize: Int,
+                                        maxLimitSize: Option[Int],
                                         var ssthresh: Int,
                                         config: SiriusConfiguration) extends Actor with MonitoringHooks {
 
-  var window = 1
-  def timeout() = (timeoutConst + (window * timeoutCoeff)).seconds
+  var limit = 1
+  def timeout() = (timeoutConst + (limit * timeoutCoeff)).seconds
 
   implicit val executionContext = context.dispatcher
 
   def receive = {
     case InitiateCatchup(fromSeq) =>
       membershipHelper.getRandomMember.map(remote => {
-        requestSubrange(fromSeq, window, remote)
+        requestSubrange(fromSeq, limit, remote)
         context.become(catchup(remote))
       })
   }
 
   def catchup(source: ActorRef): Receive = {
     case CatchupRequestSucceeded(logSubrange: CompleteSubrange) =>
-      if (window >= ssthresh) { // we're in Congestion Avoidance phase
-        window = Math.min(window + 2, maxWindowSize)
+      if (limit >= ssthresh) { // we're in Congestion Avoidance phase
+        limit = Math.min(limit + 2, maxLimitSize.getOrElse(maxWindowSize))
       } else { // we're in Slow Start phase
-        window = Math.min(window * 2, ssthresh)
+        limit = Math.min(limit * 2, ssthresh)
       }
       context.parent ! logSubrange
 
     case CatchupRequestSucceeded(logSubrange) =>
       context.parent ! logSubrange
 
     case CatchupRequestFailed =>
-      if (window != 1) {
+      if (limit != 1) {
         // adjust ssthresh, revert to Slow Start phase
-        ssthresh = Math.max(window / 2, 1)
-        window = 1
+        ssthresh = Math.max(limit / 2, 1)
+        limit = 1
       }
       context.unbecome()
 
     case ContinueCatchup(fromSeq: Long) =>
-      requestSubrange(fromSeq, window, source)
+      requestSubrange(fromSeq, limit, source)
 
     case StopCatchup =>
       context.unbecome()
   }
 
-  def requestSubrange(fromSeq: Long, window: Int, source: ActorRef): Unit = {
-    source.ask(GetLogSubrange(fromSeq, fromSeq + window))(timeout()).onComplete {
+  def requestSubrange(fromSeq: Long, limit: Int, source: ActorRef): Unit = {
+    val message = maxLimitSize match {
+      case Some(_) => GetLogSubrangeWithLimit(fromSeq, fromSeq + maxWindowSize, limit)
+      case None => GetLogSubrange(fromSeq, fromSeq + limit)
+    }
+    source.ask(message)(timeout()).onComplete {
       case Success(logSubrange: LogSubrange) => self ! CatchupRequestSucceeded(logSubrange)
       case _ => self ! CatchupRequestFailed
     }
@@ -137,6 +147,6 @@ private[bridge] class CatchupSupervisor(membershipHelper: MembershipHelper,
 
   class CatchupSupervisorInfo extends CatchupSupervisorInfoMBean {
     def getSSThresh = ssthresh
-    def getWindow = window
+    def getLimit = limit
   }
 }

src/main/scala/com/comcast/xfinity/sirius/api/impl/state/SiriusPersistenceActor.scala

@@ -30,6 +30,10 @@ object SiriusPersistenceActor {
    * that is, the state of persisted data
    */
   sealed trait LogQuery
+  sealed trait LogQuerySubrange extends LogQuery {
+    def begin: Long
+    def end: Long
+  }
 
   case object GetLogSize extends LogQuery
   /**
@@ -42,7 +46,7 @@ object SiriusPersistenceActor {
    * @param begin first sequence number of the range
    * @param end last sequence number of the range, inclusive
    */
-  case class GetLogSubrange(begin: Long, end: Long) extends LogQuery
+  case class GetLogSubrange(begin: Long, end: Long) extends LogQuerySubrange
   /**
    * Message for directly requesting a chunk of the log from a node.
    *
@@ -54,7 +58,7 @@ object SiriusPersistenceActor {
    * @param end last sequence number of the range, inclusive
    * @param limit the maximum number of events
    */
-  case class GetLogSubrangeWithLimit(begin: Long, end: Option[Long], limit: Long) extends LogQuery
+  case class GetLogSubrangeWithLimit(begin: Long, end: Long, limit: Long) extends LogQuerySubrange
 
   trait LogSubrange
   trait PopulatedSubrange extends LogSubrange {
@@ -143,14 +147,11 @@ class SiriusPersistenceActor(stateActor: ActorRef,
 
       lastWriteTime = thisWriteTime
 
-    case GetLogSubrangeWithLimit(start, Some(end), limit) =>
-      sender ! queryLimitedSubrange(start, end, limit)
-
-    case GetLogSubrangeWithLimit(start, None, limit) =>
-      sender ! queryLimited(start, limit)
-
     case GetLogSubrange(start, end) =>
-      sender ! querySubrange(start, end)
+      sender ! querySubrange(start, end, Long.MaxValue)
+
+    case GetLogSubrangeWithLimit(start, end, limit) =>
+      sender ! querySubrange(start, end, limit)
 
     case GetNextLogSeq =>
       sender ! siriusLog.getNextSeq
@@ -163,99 +164,46 @@ class SiriusPersistenceActor(stateActor: ActorRef,
     case _: SiriusResult =>
   }
 
-  private def queryLimitedSubrange(rangeStart: Long, rangeEnd: Long, limit: Long): LogSubrange =
-    if (limit <= 0 || rangeEnd < rangeStart || rangeEnd <= 0) {
-      // invalid query subrange or limit, we can't send anything useful back
+  private def querySubrange(rangeStart: Long, rangeEnd: Long, limit: Long): LogSubrange = {
+    val nextSeq = siriusLog.getNextSeq
+    val lastSeq = nextSeq - 1
+    if (limit <= 0 || rangeEnd < rangeStart || rangeEnd <= 0 || rangeStart > lastSeq) {
+      // parameters are out of range, can't return anything useful
       EmptySubrange
-    } else if (limit > (rangeEnd - rangeStart)) {
-      // limit is larger than the subrange window, no need to enforce limit
-      querySubrange(rangeStart, rangeEnd)
     } else {
-      val nextSeq = siriusLog.getNextSeq
-      if (rangeStart >= nextSeq) {
-        // query is out of range, we can't send anything useful back
-        EmptySubrange
-      } else if (rangeEnd >= nextSeq) {
-        // we can only answer partially
-        val lastSeq = nextSeq - 1
-        val buffer = siriusLog.foldLeftRangeWhile(rangeStart, lastSeq)(ListBuffer.empty[OrderedEvent])(
-          // continue folding events as long as the buffer is smaller than the limit
-          buffer => buffer.size < limit
-        )(
-          (acc, event) => acc += event
-        )
-        if (buffer.size < limit) {
-          PartialSubrange(rangeStart, lastSeq, buffer.toList)
-        } else {
-          PartialSubrange(rangeStart, buffer.last.sequence, buffer.toList)
-        }
-      } else {
-        val buffer = siriusLog.foldLeftRangeWhile(rangeStart, rangeEnd)(ListBuffer.empty[OrderedEvent])(
-          // continue folding events as long as the buffer is smaller than the limit
-          buffer => buffer.size < limit
-        )(
-          (acc, event) => acc += event
-        )
-        if (buffer.size < limit) {
-          CompleteSubrange(rangeStart, rangeEnd, buffer.toList)
+      val endSeq = if (rangeEnd > lastSeq) lastSeq else rangeEnd
+      if (limit > (endSeq - rangeStart)) {
+        // the limit is larger than the subrange window, so do not enforce
+        val events = siriusLog.foldLeftRange(rangeStart, endSeq)(ListBuffer.empty[OrderedEvent])(
+          (acc, evt) => acc += evt
+        ).toList
+        if (endSeq < rangeEnd) {
+          // the end of the range extends beyond the end of the log, so can only partially answer
+          PartialSubrange(rangeStart, endSeq, events)
         } else {
-          PartialSubrange(rangeStart, buffer.last.sequence, buffer.toList)
+          // the range is entirely within the log, so can fully answer
+          CompleteSubrange(rangeStart, endSeq, events)
         }
-      }
-    }
-
-  private def queryLimited(rangeStart: Long, limit: Long): LogSubrange = {
-    if (limit <= 0) {
-      // invalid query subrange or limit, we can't send anything useful back
-      EmptySubrange
-    } else {
-      val nextSeq = siriusLog.getNextSeq
-      if (rangeStart >= nextSeq) {
-        // query is out of range, we can't send anything useful back
-        EmptySubrange
       } else {
-        val nextSeq = siriusLog.getNextSeq
-        val lastSeq = nextSeq - 1
-        val buffer = siriusLog.foldLeftRangeWhile(rangeStart, lastSeq)(ListBuffer.empty[OrderedEvent])(
-          // continue folding events as long as the buffer is smaller than the limit
+        // the limit is smaller than the subrange window
+        val buffer = siriusLog.foldLeftRangeWhile(rangeStart, endSeq)(ListBuffer.empty[OrderedEvent])(
           buffer => buffer.size < limit
         )(
-          (acc, event) => acc += event
+          (acc, evt) => acc += evt
         )
-        if (buffer.size < limit) {
-          CompleteSubrange(rangeStart, lastSeq, buffer.toList)
+        if (buffer.size < limit && endSeq < rangeEnd) {
+          // the end of the subrange extended part the end of the log
+          // and the buffer was not filled to the limit, so we can only partially respond
+          PartialSubrange(rangeStart, endSeq, buffer.toList)
         } else {
-          PartialSubrange(rangeStart, buffer.last.sequence, buffer.toList)
+          // the buffer was filled to the limit, so completely respond using the sequence of the
+          // last event as the end of the range
+          CompleteSubrange(rangeStart, buffer.last.sequence, buffer.toList)
         }
       }
     }
   }
 
-  private def querySubrange(rangeStart: Long, rangeEnd: Long): LogSubrange =
-    if (rangeEnd < rangeStart || rangeEnd <= 0) {
-      // invalid query subrange or limit, we can't send anything useful back
-      EmptySubrange
-    } else {
-      val nextSeq = siriusLog.getNextSeq
-      if (rangeStart >= nextSeq) {
-        // query is out of range, we can't send anything useful back
-        EmptySubrange
-      } else if (rangeEnd >= nextSeq) {
-        // we can answer partially
-        val lastSeq = nextSeq - 1
-        val buffer = siriusLog.foldLeftRange(rangeStart, lastSeq)(ListBuffer.empty[OrderedEvent])(
-          (acc, event) => acc += event
-        )
-        PartialSubrange(rangeStart, lastSeq, buffer.toList)
-      } else {
-        // we can answer fully
-        val buffer = siriusLog.foldLeftRange(rangeStart, rangeEnd)(ListBuffer.empty[OrderedEvent])(
-          (acc, event) => acc += event
-        )
-        CompleteSubrange(rangeStart, rangeEnd, buffer.toList)
-      }
-    }
-
   /**
    * Monitoring hooks
    */