From 93d3a22e10f90be828ee87b898caf4c71ccd07a7 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 23 Sep 2010 21:42:01 +0000 Subject: [PATCH] Add weighted request scheduler. patch by Jeremy Hanna; reviewed by Stu Hood for CASSANDRA-1485 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1000640 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + conf/cassandra.yaml | 21 ++++-- .../cassandra/avro/CassandraServer.java | 2 +- .../config/RequestSchedulerOptions.java | 9 ++- .../scheduler/RoundRobinScheduler.java | 65 ++++++++++++++----- .../apache/cassandra/service/ClientState.java | 16 +++-- .../cassandra/thrift/CassandraServer.java | 2 +- 7 files changed, 83 insertions(+), 33 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 0561543646..85256bb712 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -86,6 +86,7 @@ * remove Clock from the Thrift (and Avro) API (CASSANDRA-1501) * Close intra-node sockets when connection is broken (CASSANDRA-1528) * RPM packaging spec file (CASSANDRA-786) + * weighted request scheduler (CASSANDRA-1485) 0.7-beta1 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index d5de5c2497..2b1d6b8564 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -210,9 +210,9 @@ dynamic_snitch: true # not affect inter node communication. # org.apache.cassandra.scheduler.NoScheduler - No scheduling takes place # org.apache.cassandra.scheduler.RoundRobinScheduler - Round robin of -# client requests to a node with a sepearte queue for each -# reques_scheduler_id. The requests are throttled based on the limit set -# in throttle_limit in the requeset_scheduler_options +# client requests to a node with a separate queue for each +# request_scheduler_id. The scheduler is further customized by +# request_scheduler_options as described below. request_scheduler: org.apache.cassandra.scheduler.NoScheduler # Scheduler Options vary based on the type of scheduler @@ -224,12 +224,23 @@ request_scheduler: org.apache.cassandra.scheduler.NoScheduler # running requests can complete. # The value of 80 here is twice the number of # concurrent_reads + concurrent_writes. +# - default_weight -- default_weight is optional and allows for +# overriding the default which is 1. +# - weights -- Weights are optional and will default to 1 or the +# overridden default_weight. The weight translates into how +# many requests are handled during each turn of the +# RoundRobin, based on the scheduler id. +# # request_scheduler_options: # throttle_limit: 80 +# default_weight: 5 +# weights: +# Keyspace1: 1 +# Keyspace2: 5 # request_scheduler_id -- An identifer based on which to perform -# the request scheduling. The current supported option is "keyspace" -request_scheduler_id: keyspace +# the request scheduling. Currently the only valid option is keyspace. +# request_scheduler_id: keyspace # A ColumnFamily is the Cassandra concept closest to a relational table. # diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index 040a6347bb..c1f50d4b52 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -822,7 +822,7 @@ public class CassandraServer implements Cassandra { */ private void schedule() { - requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId()); + requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingValue()); } /** diff --git a/src/java/org/apache/cassandra/config/RequestSchedulerOptions.java b/src/java/org/apache/cassandra/config/RequestSchedulerOptions.java index 601b4a0ae3..1cf12b4618 100644 --- a/src/java/org/apache/cassandra/config/RequestSchedulerOptions.java +++ b/src/java/org/apache/cassandra/config/RequestSchedulerOptions.java @@ -1,4 +1,6 @@ package org.apache.cassandra.config; + +import java.util.Map; /* * * Licensed to the Apache Software Foundation (ASF) under one @@ -25,5 +27,10 @@ package org.apache.cassandra.config; */ public class RequestSchedulerOptions { - public Integer throttle_limit = 80; + public static final Integer DEFAULT_THROTTLE_LIMIT = 80; + public static final Integer DEFAULT_WEIGHT = 1; + + public Integer throttle_limit = DEFAULT_THROTTLE_LIMIT; + public Integer default_weight = DEFAULT_WEIGHT; + public Map weights; } diff --git a/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java b/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java index cf5949d016..da06edfaba 100644 --- a/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java +++ b/src/java/org/apache/cassandra/scheduler/RoundRobinScheduler.java @@ -26,6 +26,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.SynchronousQueue; import org.apache.cassandra.config.RequestSchedulerOptions; +import org.apache.cassandra.utils.Pair; import org.cliffc.high_scale_lib.NonBlockingHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,12 +34,15 @@ import org.slf4j.LoggerFactory; /** * A very basic Round Robin implementation of the RequestScheduler. It handles * request groups identified on user/keyspace by placing them in separate - * queues and servicing a request from each queue in a RoundRobin fashion. + * queues and servicing a request from each queue in a RoundRobin fashion. + * It optionally adds weights for each round. */ public class RoundRobinScheduler implements IRequestScheduler { private static final Logger logger = LoggerFactory.getLogger(RoundRobinScheduler.class); - private final NonBlockingHashMap> queues; + + //The Pair is the weighted queue - the left is the weight and the right is the queue + private final NonBlockingHashMap>> queues; private static boolean started = false; private final Semaphore taskCount; @@ -46,12 +50,18 @@ public class RoundRobinScheduler implements IRequestScheduler // Used by the the scheduler thread so we don't need to busy-wait until there is a request to process private final Semaphore queueSize = new Semaphore(0, false); + private Integer defaultWeight; + private Map weights; + public RoundRobinScheduler(RequestSchedulerOptions options) { assert !started; + defaultWeight = options.default_weight; + weights = options.weights; + taskCount = new Semaphore(options.throttle_limit); - queues = new NonBlockingHashMap>(); + queues = new NonBlockingHashMap>>(); Runnable runnable = new Runnable() { public void run() @@ -70,12 +80,12 @@ public class RoundRobinScheduler implements IRequestScheduler public void queue(Thread t, String id) { - SynchronousQueue queue = getQueue(id); + Pair> weightedQueue = getWeightedQueue(id); try { queueSize.release(); - queue.put(t); + weightedQueue.right.put(t); } catch (InterruptedException e) { @@ -90,14 +100,26 @@ public class RoundRobinScheduler implements IRequestScheduler private void schedule() { + int weight; + SynchronousQueue queue; + Thread t; + queueSize.acquireUninterruptibly(); - for (SynchronousQueue queue : queues.values()) + for (Map.Entry>> request : queues.entrySet()) { - Thread t = queue.poll(); - if (t != null) + weight = request.getValue().left; + queue = request.getValue().right; + //Using the weight, process that many requests at a time (for that scheduler id) + for (int i=0; i getQueue(String id) + private Pair> getWeightedQueue(String id) { - SynchronousQueue queue = queues.get(id); - if (queue != null) + Pair> weightedQueue = queues.get(id); + if (weightedQueue != null) // queue existed - return queue; + return weightedQueue; - SynchronousQueue maybenew = new SynchronousQueue(true); - queue = queues.putIfAbsent(id, maybenew); - if (queue == null) + Pair> maybenew = new Pair(getWeight(id), new SynchronousQueue(true)); + weightedQueue = queues.putIfAbsent(id, maybenew); + if (weightedQueue == null) // created new queue return maybenew; // another thread created the queue - return queue; + return weightedQueue; } Semaphore getTaskCount() { return taskCount; } + + private int getWeight(String weightingVar) + { + return (weights != null && weights.containsKey(weightingVar)) + ? weights.get(weightingVar) + : defaultWeight; + } } diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java index cd536f193f..71719a27f2 100644 --- a/src/java/org/apache/cassandra/service/ClientState.java +++ b/src/java/org/apache/cassandra/service/ClientState.java @@ -38,9 +38,6 @@ import org.apache.cassandra.thrift.InvalidRequestException; public class ClientState { private static Logger logger = LoggerFactory.getLogger(ClientState.class); - - // true if the keyspace should be used as the scheduling id - private final boolean SCHEDULE_ON_KEYSPACE = DatabaseDescriptor.getRequestSchedulerId().equals(RequestSchedulerId.keyspace); // Current user for the session private final ThreadLocal user = new ThreadLocal() @@ -80,11 +77,16 @@ public class ClientState updateKeyspaceAccess(); } - public String getSchedulingId() + public String getSchedulingValue() { - if (SCHEDULE_ON_KEYSPACE) - return keyspace.get(); - return "default"; + String schedulingValue = "default"; + switch(DatabaseDescriptor.getRequestSchedulerId()) + { + case keyspace: + schedulingValue = keyspace.get(); + break; + } + return schedulingValue; } /** diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index f3ac08a989..434d643492 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -615,7 +615,7 @@ public class CassandraServer implements Cassandra.Iface */ private void schedule() { - requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingId()); + requestScheduler.queue(Thread.currentThread(), clientState.getSchedulingValue()); } /**