Add CachingCostBalancerStrategy #61

dgolitsyn · 2017-05-30T09:54:33Z

CachingCostBalancerStrategy is fast implementation of CostBalancerStrategy. Computation algorithm slightly differs from the original version and allows pre-compute and cache cost function values to reduce complexity of the algorithm.

Motivation:

Current implementation is very complex computational task and calculation time grows linearly with number of segments in the cluster. For rather big cluster one druid balancer iteration could take tens of minutes
If large number of segments need to be rebalanced, cost function calculation could take hours or even days (if number of historical nodes changed for instance)
Total server cost could not be calculated due to high computational complexity and thus could not be included in balancing algorithm

Implementation details:

Result numbers are not equal for CachingCostBalancerStrategy and CostBalancerStrategy, but decisions made are the same, meaning that the same server has been chosen for particular segment by two strategies (has been tested for ~100k segments from our cluster)
CachingCostBalancerStrategy is less memory efficient and requires ~100mb heap for each million of segments
Computation time is ~20000x faster (depends on segment distribution and total number of servers and segments)

Benchmark

                           Mode  Cnt     Score     Error  Units
cachingCostStrategy       avgt   20     0.238 ±  0.003  ms/op
costStrategy              avgt   20  4549.247 ± 26.430  ms/op

…gyFactory

leventov · 2017-05-30T16:47:29Z

benchmarks/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyBenchmark.java

+    Random random = new Random(seed);
+    SegmentCostCache.Builder prototype = SegmentCostCache.builder();
+    for (int i = 0; i < NUMBER_OF_SEGMENTS; ++i) {
+      DataSegment segment = createSegment(random.nextInt(30 * 24));


Is that number of days in a month * number of hours in a day? Please leave a comment

leventov · 2017-05-30T16:48:20Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategy.java

+    this.clusterCostCache = Preconditions.checkNotNull(clusterCostCache);
+  }
+
+  protected double computeCost(DataSegment proposalSegment, ServerHolder serverHolder)


Suggested to inline this method, to avoid unnecessary method overloading

agree, fixed

leventov · 2017-05-30T21:27:02Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+  @LifecycleStart
+  public void start()
+  {
+    if (!executorRef.compareAndSet(null, Execs.singleThreaded("CachingCostBalancerStrategy-executor"))) {


Doesn't seem that you need AtomicReference and CAS here, it is not switched back to null anywhere, although part of message "or has been stopped" below means that it was planned.

yes, it was planned to be set to null in stop method after shutdownNow is called for current executor

leventov · 2017-05-30T21:27:15Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+    if (!executorRef.compareAndSet(null, Execs.singleThreaded("CachingCostBalancerStrategy-executor"))) {
+      throw new ISE("CachingCostBalancerStrategyFactory is already started or has been stopped");
+    }
+    if (!started.compareAndSet(false, true)) {


Could you use lifecycleLock?

Please address or comment on this

leventov · 2017-05-30T21:34:42Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+    }
+    serverInventoryView.registerSegmentCallback(
+        executorRef.get(),
+        new ServerView.SegmentCallback()


If this callback may be called in multiple "cycles", this whole construction seems to be racy. Then it need to call clusterCostCacheBuilder.build() in segmentViewInitialized() and save it. For future use in createBalancerStrategy().

Or maybe this factory doesn't create own callbacks, but uses already build server/segment view from CoordinatorServerView directly in createBalancerStrategy()?

Not sure, that understand this comment correctly. What do you think is racy?
All interactions with clusterCostCacheBuilder are done via executor which is singleThreaded. Created clusterCostCache is immutable and can be used by CachingCostBalancerStrategy in multiple threads safely.

See comment below

leventov · 2017-05-30T23:00:31Z

server/src/main/java/io/druid/server/coordinator/cost/ServerCostCache.java

+  {
+    return new Builder();
+  }
+


Extra empty line

leventov · 2017-05-30T23:00:45Z

server/src/main/java/io/druid/server/coordinator/cost/ServerCostCache.java

+    public void addSegment(DataSegment dataSegment)
+    {
+      allSegmentsCostCache.addSegment(dataSegment);
+      segmentsPerDataSource.computeIfAbsent(


IMO formatting like

segmentsPerDataSource .computeIfAbsent(dataSegment.getDataSource(), d -> SegmentCostCache.builder()) .addSegment(dataSegment);

is more readable

leventov · 2017-05-30T23:01:15Z

server/src/main/java/io/druid/server/coordinator/cost/ServerCostCache.java

+    {
+      allSegmentsCostCache.removeSegement(dataSegment);
+      SegmentCostCache.Builder builder = segmentsPerDataSource.get(dataSegment.getDataSource());
+      builder.removeSegement(dataSegment);


Check not null?

leventov · 2017-05-30T23:03:01Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentCostCache.java

+      public void removeSegment(DataSegment dataSegment)
+      {
+        int index = -1;
+        ListIterator<DataSegment> it = dataSegments.listIterator();


Please define it in the loop scope. Also it's type don't need to be ListIterator, could be just iterator.

leventov · 2017-05-30T23:05:02Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentCostCache.java

+        }
+
+        if (index < 0) {
+          throw new ISE("Failed to remove segment from bucket: segment not found");


Why ISE? In other similar places absence is considered normal

leventov · 2017-06-01T22:39:04Z

benchmarks/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyBenchmark.java

+    }
+    segmentsCostCache = prototype.build();
+    for (int i = 0; i < NUMBER_OF_QUERIES; ++i) {
+      DataSegment segment = createSegment(random.nextInt(30 * 24));


Same as above

leventov · 2017-06-01T22:43:13Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+          public ServerView.CallbackAction segmentViewInitialized()
+          {
+            initialized = true;
+            clusterCostCacheBuilder = ClusterCostCache.builder();


This line doesn't solve the problem, because you always construct CachingCostBalancerStrategy from an empty builder.

leventov · 2017-06-01T22:43:37Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+    }
+    serverInventoryView.registerSegmentCallback(
+        executorRef.get(),
+        new ServerView.SegmentCallback()


See comment below

leventov · 2017-06-01T22:44:05Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+    if (!executorRef.compareAndSet(null, Execs.singleThreaded("CachingCostBalancerStrategy-executor"))) {
+      throw new ISE("CachingCostBalancerStrategyFactory is already started or has been stopped");
+    }
+    if (!started.compareAndSet(false, true)) {


Please address or comment on this

leventov · 2017-06-01T22:44:19Z

server/src/main/java/io/druid/server/coordinator/cost/ServerCostCache.java

+  {
+    return new Builder();
+  }
+


leventov · 2017-06-16T23:16:56Z

Also @dgolitsyn please review how does it relate to apache#3485

leventov · 2017-06-23T18:32:10Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategy.java

+import io.druid.timeline.DataSegment;
+
+
+public class CachingCostBalancerStrategy extends CostBalancerStrategy


CachingCostBalancerStrategy is the same algorithm as CostBalancerStrategy, so I suggest just to "add caching" to "CostBalancerStrategy", not introduce another configuration option.

I think the only reason to introduce another configuration option is that CachingCostBalancerStrategy has bigger memory footprint that CostBalancerStrategy. Not sure that all users are ok with such change

How much extra memory it would take on coordinator in our prod cluster?

leventov · 2017-06-23T18:32:40Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+  private static final EmittingLogger log = new EmittingLogger(CachingCostBalancerStrategyFactory.class);
+
+  private final ServerInventoryView serverInventoryView;
+  private final Object lifecycleLock = new Object();


Please use LifecycleLock class

leventov · 2017-06-23T18:35:06Z

server/src/main/java/io/druid/server/coordinator/CachingCostBalancerStrategyFactory.java

+    }
+    if (initialized) {
+      try {
+        CompletableFuture<CachingCostBalancerStrategy> future = CompletableFuture.supplyAsync(


What do you think about replacing all internal maps in ClusterCostCache.Builder with ConcurrentHashMap, and allow concurrent CachingCostBalancerStrategy construction in the same thread as createBalancerStrategy() is called? It simplify code and allow to remove a exception handling here.

This change will require to make ServerCostCache thread-safe as well as SegmentsCostCache. I think that the overall overhead could be significant, so my vote is to keep it as is.

It's updated only from Curator, likely from a single thread, so I don't think it will actually make any performance difference. OTOH it allows to avoid nasty threading and error-catching logic in the factory.

leventov · 2017-06-23T18:35:59Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+
+public class SegmentsCostCache
+{
+  private static final double HALF_LIFE = 1.0; // cost function half-life in hours


Please add code comments, a lot of comments all over this class. See my previous PR comments

leventov · 2017-06-23T18:36:36Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+
+    private Interval toBucketInterval(DataSegment segment)
+    {
+      long start = segment.getInterval().getStartMillis() - (segment.getInterval().getStartMillis() % BUCKET_INTERVAL);


Granularity has logic like this, could it be reused?

leventov · 2017-06-23T18:40:04Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+      return interval;
+    }
+
+    public boolean inCalculationInterval(DataSegment dataSegment)


Please add tests for this method

tests added

leventov · 2017-06-23T18:41:15Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+
+      // avoid calculation for segments outside of LIFE_THRESHOLD
+      if (!inCalculationInterval(dataSegment)) {
+        return cost;


"This situation is already excluded on the level where Bucket.cost() is called, so throwing ISE seems more reasonable to me" - prev comment

leventov · 2017-06-23T18:57:14Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentCostCache.java

+      }
+      // add left-non-overlapping segments
+      if (leftIndex >= 0) {
+        cost += leftSum[leftIndex] * (FastMath.exp(-t1) - FastMath.exp(-t0));


apache#2972 (comment)

leventov · 2017-06-23T18:59:04Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+      }
+      // add all right-overlapping segments
+      int rightIndex = index;
+      while (rightIndex < sortedSegments.size() && sortedSegments.get(rightIndex)


Break on &&

…lity

…CostBalancerStrategy

leventov · 2017-07-06T05:04:06Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+ *
+ * Joint cost for two segments:
+ *
+ *        cost(X, Y) = \int_{x_0}^{x_1} \int_{y_0}^{y_1} e^{-\lambda |x-y|}dxdy


Is there any online service that allows to plot those formulas?

https://www.codecogs.com/latex/eqneditor.php

leventov · 2017-07-06T05:04:56Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+        segments.tailSet(valueAndSegment).forEach(v -> v.leftValue += leftValue);
+        segments.headSet(valueAndSegment).forEach(v -> v.rightValue += rightValue);
+
+        ValueAndSegment lower = segments.lower(valueAndSegment);


Why this? tailSet and headSet are already updated above

leventov · 2017-07-06T05:07:40Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+        segments.tailSet(segmentAndValue).forEach(v -> v.leftValue += leftValue);
+        segments.headSet(segmentAndValue).forEach(v -> v.rightValue += rightValue);
+
+        SegmentAndValue lower = segments.lower(segmentAndValue);


Why this? tailSet and headSet already updated above

add comments for this

leventov · 2017-07-06T05:09:21Z

server/src/main/java/io/druid/server/coordinator/cost/SegmentsCostCache.java

+
+  static class SegmentAndValue implements Comparable<SegmentAndValue>
+  {
+    private final DataSegment dataSegment;


Maybe for sanity add equals() and hashCode() that compare the segment. Since we don't actually expect that the same segment is added, we may check

if (!segments.add(segmentAndValue)) { throw new ISE("expect new segment"); }

…dSum

leventov · 2017-09-28T19:55:41Z

Closing as it is merged into upstream: apache#4731

dgolitsyn and others added 5 commits May 30, 2017 12:52

Add CachingCostBalancerStrategy

69d0bbb

Fail fast if serverInventoryView is null in CachingCostBalancerStrate…

108f12a

…gyFactory

Fix races in CachingCostBalancerStrategyFactory

fe80785

Fix CachingCostBalancerStrategyBenchmark

7a86471

Benchmark fixes

30e6c76

leventov suggested changes May 30, 2017

View reviewed changes

Address review comments

c58ae94

leventov suggested changes Jun 1, 2017

View reviewed changes

dgolitsyn added 3 commits June 22, 2017 19:50

LifecycleLock in CachingCostBalancerStrategyFactory

2dba9ca

Fix typo

231b232

Address review comments

73a716e

leventov reviewed Jun 23, 2017

View reviewed changes

leventov and others added 9 commits June 23, 2017 14:21

Rename ServerView.ServerCallback to ServerRemovedCallback for readabi…

3bc9376

…lity

Use LifecycleLock in CachingCostBalancerStrategyFactory

eeb7b6c

Use DurationGranularity in SegmentCostCache

5975448

Add test for CachingCostBalancerStrategy

3f524d1

Added docs to SegmentsCostCache

034c56a

Include segments that are going to be loaded to total cost in Caching…

43ce54b

…CostBalancerStrategy

Some refactoring and style

342a67b

Typo

cd27ca4

ValueAndSegment -> SegmentAndValue

23fa1e8

leventov reviewed Jul 6, 2017

View reviewed changes

dgolitsyn and others added 5 commits July 6, 2017 13:08

Add docs to SegmentsCostCache and rename SegmentAndValue to SegmentAn…

8dfb995

…dSum

Reduce bucketInterval in SegmentsCostCache from 30 to 15

250a7b7

Fix CachingCostBalancerStrategyBenchmark

5cdb422

Add link to plotter

843c462

Merge branch '0.10.1-mmx' into caching-cost-balancer

cae5e83

leventov closed this Sep 28, 2017

leventov deleted the caching-cost-balancer branch September 28, 2017 19:55

		import io.druid.timeline.DataSegment;


		public class CachingCostBalancerStrategy extends CostBalancerStrategy

Add CachingCostBalancerStrategy #61

Add CachingCostBalancerStrategy #61

Conversation

dgolitsyn commented May 30, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leventov commented Jun 16, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leventov Jul 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leventov commented Sep 28, 2017

dgolitsyn commented May 30, 2017 •

edited

Loading

leventov Jul 6, 2017 •

edited

Loading