You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by jg...@apache.org on 2014/02/13 20:29:17 UTC
git commit: SAMZA-146: Throughput degrades unreasonably as the number
of topic-partitions increases.
Updated Branches:
refs/heads/master 30a606370 -> 5cdbd02b8
SAMZA-146: Throughput degrades unreasonably as the number of topic-partitions increases.
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/5cdbd02b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/5cdbd02b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/5cdbd02b
Branch: refs/heads/master
Commit: 5cdbd02b8df1b10195e8b10e786a69c310d12433
Parents: 30a6063
Author: Jakob Glen Homan <jg...@apache.org>
Authored: Thu Feb 13 11:28:34 2014 -0800
Committer: Jakob Glen Homan <jg...@apache.org>
Committed: Thu Feb 13 11:28:34 2014 -0800
----------------------------------------------------------------------
.../apache/samza/system/SystemConsumers.scala | 65 +++++++++++++-------
.../org/apache/samza/util/DoublingBackOff.scala | 62 +++++++++++++++++++
.../apache/samza/util/TestDoublingBackOff.scala | 61 ++++++++++++++++++
.../samza/system/mock/MockSystemConsumer.java | 2 +-
.../test/integration/TestStatefulTask.scala | 2 +-
.../TestSamzaContainerPerformance.scala | 2 +-
6 files changed, 170 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala b/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
index cdba7fe..d70c91b 100644
--- a/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
+++ b/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
@@ -24,6 +24,7 @@ import scala.collection.mutable.Queue
import org.apache.samza.serializers.SerdeManager
import grizzled.slf4j.Logging
import org.apache.samza.system.chooser.MessageChooser
+import org.apache.samza.util.DoublingBackOff
/**
* The SystemConsumers class coordinates between all SystemConsumers, the
@@ -135,6 +136,13 @@ class SystemConsumers(
*/
val depletedQueueSizeThreshold = (maxMsgsPerStreamPartition * (1 - fetchThresholdPct)).toInt
+ /**
+ * Make the maximum backoff proportional to the number of streams we're consuming.
+ * For a few streams, make the max back off 1, but for hundreds make it up to 1k,
+ * which experimentally has shown to be the most performant.
+ */
+ var maxBackOff = 0
+
debug("Got stream consumers: %s" format consumers)
debug("Got max messages per stream: %s" format maxMsgsPerStreamPartition)
debug("Got no new message timeout: %s" format noNewMessagesTimeout)
@@ -148,6 +156,10 @@ class SystemConsumers(
def start {
debug("Starting consumers.")
+ maxBackOff = scala.math.pow(10, scala.math.log10(fetchMap.size).toInt).toInt
+
+ debug("Got maxBackOff: " + maxBackOff)
+
consumers
.keySet
.foreach(metrics.registerSystem)
@@ -176,7 +188,31 @@ class SystemConsumers(
chooser.register(systemStreamPartition, lastReadOffset)
}
- def choose = {
+ /**
+ * Needs to be be lazy so that we are sure to get the value of maxBackOff assigned
+ * in start(), rather than its initial value.
+ */
+ lazy val refresh = new DoublingBackOff(maxBackOff) {
+ def call(): Boolean = {
+ debug("Refreshing chooser with new messages.")
+
+ // Poll every system for new messages.
+ val receivedNewMessages = consumers.keys.map(poll(_)).contains(true)
+
+ // Update the chooser.
+ neededByChooser.foreach(systemStreamPartition =>
+ // If we have messages for a stream that the chooser needs, then update.
+ if (fetchMap(systemStreamPartition).intValue < maxMsgsPerStreamPartition) {
+ chooser.update(unprocessedMessages(systemStreamPartition).dequeue)
+ updateFetchMap(systemStreamPartition)
+ neededByChooser -= systemStreamPartition
+ })
+
+ receivedNewMessages
+ }
+ }
+
+ def choose: IncomingMessageEnvelope = {
val envelopeFromChooser = chooser.choose
if (envelopeFromChooser == null) {
@@ -200,31 +236,16 @@ class SystemConsumers(
metrics.systemStreamMessagesChosen(envelopeFromChooser.getSystemStreamPartition.getSystemStream).inc
}
- refresh
+ refresh.maybeCall()
envelopeFromChooser
}
-
- private def refresh {
- debug("Refreshing chooser with new messages.")
-
- // Poll every system for new messages.
- consumers.keys.foreach(poll(_))
-
- // Update the chooser.
- neededByChooser.foreach(systemStreamPartition =>
- // If we have messages for a stream that the chooser needs, then update.
- if (fetchMap(systemStreamPartition).intValue < maxMsgsPerStreamPartition) {
- chooser.update(unprocessedMessages(systemStreamPartition).dequeue)
- updateFetchMap(systemStreamPartition)
- neededByChooser -= systemStreamPartition
- })
- }
-
+
/**
* Poll a system for new messages from SystemStreamPartitions that have
- * dipped below the depletedQueueSizeThreshold threshold.
+ * dipped below the depletedQueueSizeThreshold threshold. Return true if
+ * any envelopes were found, false if none.
*/
- private def poll(systemName: String) = {
+ private def poll(systemName: String): Boolean = {
debug("Polling system consumer: %s" format systemName)
metrics.systemPolls(systemName).inc
@@ -259,6 +280,8 @@ class SystemConsumers(
debug("Updated unprocessed messages for: %s, %s" format (systemStreamPartition, unprocessedMessages))
})
+
+ !incomingEnvelopes.isEmpty
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-core/src/main/scala/org/apache/samza/util/DoublingBackOff.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/util/DoublingBackOff.scala b/samza-core/src/main/scala/org/apache/samza/util/DoublingBackOff.scala
new file mode 100644
index 0000000..e1d6d4c
--- /dev/null
+++ b/samza-core/src/main/scala/org/apache/samza/util/DoublingBackOff.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.samza.util
+
+/**
+ * Perform the provided action (via the call method) and, if it returns true,
+ * perform it again, the next time. If, however, call returns false, do not
+ * perform call the next time, instead wait 2*n calls before actually calling,
+ * with n increasing to the maximum specified in the constructor.
+
+ * @param maxBackOff Absolute maximum number of calls to call before actually performing call.
+ */
+abstract class DoublingBackOff(maxBackOff:Int = 64) {
+ var invocationsBeforeCall = 0
+ var currentBackOff = 0
+
+ /**
+ * Method to invoke and whose return value will determine the next time
+ * it is called again.
+ */
+ def call():Boolean
+
+ /**
+ * Possibly execute the call method, based on the result of the previous run.
+ */
+ def maybeCall():Unit = {
+ if(invocationsBeforeCall == 0) {
+ if (call()) {
+ // call succeeded so reset backoff
+ currentBackOff = 0
+ } else {
+ // Call failed, so start backing off
+ currentBackOff = scala.math.min(maxBackOff, nextBackOff(currentBackOff))
+ invocationsBeforeCall = currentBackOff
+ }
+ } else {
+ invocationsBeforeCall -= 1
+ }
+
+ }
+
+ // 2 * 0 == 0, making getting started a wee bit hard, so we need a little help with that first back off
+ private def nextBackOff(i:Int) = if(i == 0) 1 else 2 * i
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-core/src/test/scala/org/apache/samza/util/TestDoublingBackOff.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/util/TestDoublingBackOff.scala b/samza-core/src/test/scala/org/apache/samza/util/TestDoublingBackOff.scala
new file mode 100644
index 0000000..eaeb005
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/util/TestDoublingBackOff.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.samza.util
+
+import org.junit.Test
+import org.junit.Assert._
+
+class TestDoublingBackOff {
+ @Test def zeroBackOffWorks() {
+ var counter = 0
+ val zeroBackOff = new DoublingBackOff(0) {
+ def call(): Boolean = {
+ counter += 1
+ true
+ }
+ }
+
+ for(i <- 0 to 1000) {
+ assertEquals(i, counter)
+ zeroBackOff.maybeCall()
+ }
+ }
+
+ @Test def backOffWorks() {
+ val toReturn = List(true, false, true, true)
+ var counter = 0
+ val ebo = new DoublingBackOff() {
+ def call(): Boolean = {
+ counter += 1
+ toReturn(counter - 1)
+ }
+ }
+
+ ebo.maybeCall() // will get back true
+ assertEquals(1, counter)
+ ebo.maybeCall() // will get back false
+ assertEquals(2, counter)
+ ebo.maybeCall() // last false means we won't actually call, will hold off for one iteration
+ assertEquals(2, counter)
+ ebo.maybeCall() // and call on this one, which gives back true
+ assertEquals(3, counter)
+ ebo.maybeCall() // so we immediately call again and increment
+ assertEquals(4, counter)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java b/samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java
index 1e3457b..7ff3111 100644
--- a/samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java
+++ b/samza-test/src/main/java/org/apache/samza/system/mock/MockSystemConsumer.java
@@ -98,7 +98,7 @@ public class MockSystemConsumer extends BlockingEnvelopeMap {
}
// Start thread.
- Thread thread = new Thread(new MockSystemConsumerRunnable(threadSsps));
+ Thread thread = new Thread(new MockSystemConsumerRunnable(threadSsps), "MockSystemConsumer-" + i);
thread.setDaemon(true);
threads.add(thread);
thread.start();
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
index ae3f663..387cf44 100644
--- a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
+++ b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
@@ -433,7 +433,7 @@ class TestTask extends StreamTask with InitableTask {
}
def awaitMessage {
- gotMessage.await(60, TimeUnit.SECONDS)
+ assertTrue("Timed out of waiting for message rather than received one.", gotMessage.await(60, TimeUnit.SECONDS))
assertEquals(0, gotMessage.getCount)
gotMessage = new CountDownLatch(1)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/5cdbd02b/samza-test/src/test/scala/org/apache/samza/test/performance/TestSamzaContainerPerformance.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/test/scala/org/apache/samza/test/performance/TestSamzaContainerPerformance.scala b/samza-test/src/test/scala/org/apache/samza/test/performance/TestSamzaContainerPerformance.scala
index e5a676e..3dc2630 100644
--- a/samza-test/src/test/scala/org/apache/samza/test/performance/TestSamzaContainerPerformance.scala
+++ b/samza-test/src/test/scala/org/apache/samza/test/performance/TestSamzaContainerPerformance.scala
@@ -84,7 +84,7 @@ class TestSamzaContainerPerformance extends Logging{
"task.inputs" -> (0 until streamCount).map(i => "mock.stream" + i).mkString(","),
"task.log.interval" -> logInterval.toString,
"task.max.messages" -> maxMessages.toString,
- "systems.mock.samza.factory" -> "org.apache.samza.system.mock.MockSystemFactory",
+ "systems.mock.samza.factory" -> classOf[org.apache.samza.system.mock.MockSystemFactory].getCanonicalName,
"systems.mock.partitions.per.stream" -> partitionsPerStreamCount.toString,
"systems.mock.messages.per.batch" -> messagesPerBatch.toString,
"systems.mock.consumer.thread.count" -> consumerThreadCount.toString,