QPS コントロール

Posted on 2014-03-27 tags: java, concurrent

QPS

QPS = Queries per Second, RPS = Requests per Second という言葉がある。読んで字の如し、秒間クエリ数を表す。例えば次のような HTTP アクセスログがあったとする。

127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326

このとき、ある一秒間を含むようなログの行数を数えれば QPS を出すことができる。

$ grep '10/Oct/2000:13:55:36 -0700' access_log | wc -l
123

QPS コントロール

リクエストを送る側が受ける側の性能の問題だったりで「500 QPS くらいで送って」と頼まれるときがある。そこでうまい具合に QPS をコントロールしたいのだが、そんなに自明な問題ではないことに気づいた。素朴には「メインスレッドが 500 個スレッドを作ってそれぞれがリクエストを飛ばし、メインスレッドは1秒寝る」みたいなことをやればいいと考えるかもしれない。

class Main {
    public static void main(String[] args) throws Exception {
        int arg = 0;
        while (true) {
            for (int i = 0; i < 500; ++i) {
                new Thread(new Task(arg++)).run();
            }
            Thread.sleep(1000);
        }
    }

    static class Task implements Runnable {
        private int arg;
        public Task(int arg) {
            this.arg = arg;
        }
        public void run() {
            sendRequest(arg);
        }
    }
}

しかし、これではメインスレッドが Thread や Runnable のインスタンスを作る分遅れ、遅れたところから 1 秒間眠ることになる。これでは正しく QPS コントロールされているとはいえない。ちょっとしたテストならそれでも良いかもしれないが、規模が大きくなるにつれて無視できないことになる。

また、リクエストを送ったがネットワークの気まぐれで失敗した場合にリトライしたいこともある。リトライのロジックを同じスレッドの中で行うとすると次のようになる。

class Main {
    public static void main(String[] args) throws Exception {
        int arg = 0;
        while (true) {
            for (int i = 0; i < 500; ++i) {
                new Thread(new Task(arg++)).run();
            }
            Thread.sleep(1000);
        }
    }

    static class Task implements Runnable {
        private int arg;
        public Task(int arg) {
            this.arg = arg;
        }
        public void run() {
            try {
                sendRequest(arg);
            } catch (Exception e) {
                run();
            }
        }
    }
}

しかしこれではリクエストが失敗したらすぐに同じリクエストが飛ぶことになり、これも QPS が狂ってしまうことになる。

問題

改めてきちんと問題を定義する。

毎秒与えられた数 n 回程度のリクエストを行う。
- ここでは n は一台のマシンが無理なく送れる程度 (〜2000 程度) とする。
失敗したときは m 回までリトライを行う。そのときも QPS は狂わないようにする。
- m 回のリトライでも失敗したときはエラーを吐いて捨てる。

タイマーを使った固定頻度実行

java.util.Timer の scheduleAtFixedRate というメソッドを使う。これを使って 1 秒ごとに実行されるようにしておけば、毎秒きっちり 1 回の頻度で実行されるようになる (イレギュラーなスケジューリングは考えない) 。

import java.util.Timer;
import java.util.TimerTask;

class Main {
    public static void main(String[] args) {
        Timer timer = new Timer();
        Task task = new Task();
        timer.scheduleAtFixedRate(task, 0L, 1000L);
    }

    static class Task extends TimerTask {
        public void run() {
            System.out.println("run!");
        }
    }
}

これで 1 秒に 1 回の割合でタスクが実行されることになる。実際にはこのタスクの中で n 回のリクエストを行いたいので、固定頻度で呼ばれる Runner と実際にリクエストを行う Task を分け、次のようになる。

import java.util.Timer;
import java.util.TimerTask;

class Main {
    static final int n = 500;

    static int arg = 0;

    public static void main(String[] args) {
        Timer timer = new Timer();
        Runner runner = new Runner();
        timer.scheduleAtFixedRate(runner, 0L, 1000L);
    }

    static class Runner extends TimerTask {
        public void run() {
            for (int i = 0; i < n; ++i) {
                new Thread(new Task(arg++)).run();
            }
        }
    }

    static class Task implements Runnable {
        private int arg;
        public Task(int arg) {
            this.arg = arg;
        }
        public void run() {
            sendRequest(arg);
        }
    }
}

リトライ

失敗した時のリトライを同じスレッドで行おうとするとどうしても QPS が狂ってしまうことは明らかである。そこでタスクキューにタスクを追加していき、次の実行の際にまとめて実行するという手段が考えられる。タスクのインスタンスはリクエストを送る際の引数とリトライ回数を保持している。

import java.util.LinkedList;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Queue;

class Main {
    static final int n = 500;
    static final int m = 3;

    static int arg = 0;

    static Queue<Thread> tasks = new LinkedList<Thread>();

    public static void main(String[] args) {
        Timer timer = new Timer();
        Runner runner = new Runner();
        timer.scheduleAtFixedRate(runner, 0L, 1000L);
    }

    static class Runner extends TimerTask {
        public void run() {
            while (tasks.size() < n) {
                tasks.offer(new Thread(new Task(arg++, m)));
            }
            for (int i = 0; i < n; ++i) {
                Thread task = tasks.poll();
                task.run();
            }
        }
    }

    static class Task implements Runnable {
        private int arg;
        private int nRetry;
        public Task(int arg, int nRetry) {
            this.arg = arg;
            this.nRetry = nRetry;
        }
        public void run() {
            try {
                sendRequest(arg);
            } catch (Exception e) {
                if (nRetry > 0) {
                    nRetry--;
                    tasks.offer(new Thread(this));
                } else {
                    System.out.println("arg = " + arg + " tried " + m + " times, but failed.");
                }
            }
        }
    }
}

しかしこの実装では、メインスレッドの他にタスクに失敗したスレッドからもアクセスされるため、それが同時に発生するとまずい。そこで同期化されたタスクキューを使う。

import java.util.concurrent.LinkedBlockingQueue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Queue;

class Main {
    static final int n = 500;
    static final int m = 3;

    static int arg = 0;

    static Queue<Thread> tasks = new LinkedBlockingQueue<Thread>();

    public static void main(String[] args) {
        Timer timer = new Timer();
        Runner runner = new Runner();
        timer.scheduleAtFixedRate(runner, 0L, 1000L);
    }

    static class Runner extends TimerTask {
        public void run() {
            while (tasks.size() < n) {
                tasks.offer(new Thread(new Task(arg++, m)));
            }
            for (int i = 0; i < n; ++i) {
                Thread thread = tasks.poll();
                thread.run();
            }
        }
    }

    static class Task implements Runnable {
        private int arg;
        private int nRetry;
        public Task(int arg, int nRetry) {
            this.arg = arg;
            this.nRetry = nRetry;
        }
        public void run() {
            try {
                sendRequest(arg);
            } catch (Exception e) {
                if (nRetry > 0) {
                    nRetry--;
                    tasks.offer(new Thread(this));
                } else {
                    System.out.println("arg = " + arg + " tried " + m + " times, but failed.");
                }
            }
        }
    }
}

スレッドプールを使う

以上で QPS コントロールの要件は満たされたはずだが、さらに改善の余地がある。何度もスレッドを作っている点だ。これまでの実装ではスレッドが使い捨てになっており、毎回新しくスレッドを作るという点でパフォーマンス的に問題がある。そこで数の固定されたスレッドプールを使う。

java.util.concurrent.ExecutorService はスレッドプールを扱う API である。この invokeAll メソッドを使えば、引数の Iterable に入ったタスクを一気に発火することができる。その後タスクキューをクリアすればよい。

import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Queue;

class Main {
    static final int n = 500;
    static final int m = 3;

    static int arg = 0;

    static LinkedBlockingQueue<Task> tasks = new LinkedBlockingQueue<Task>();

    static ExecutorService exe = Executors.newFixedThreadPool(n);

    public static void main(String[] args) {
        Timer timer = new Timer();
        Runner runner = new Runner();
        timer.scheduleAtFixedRate(runner, 0L, 1000L);
    }

    static class Runner extends TimerTask {
        public void run() {
            while (tasks.size() < n) {
                tasks.offer(new Task(arg++, m));
            }
            try {
                exe.invokeAll(tasks);
                tasks.clear();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    static class Task implements Callable<Void> {
        private int arg;
        private int nRetry;
        public Task(int arg, int nRetry) {
            this.arg = arg;
            this.nRetry = nRetry;
        }
        public Void call() {
            try {
                sendRequest(arg);
            } catch (Exception e) {
                if (nRetry > 0) {
                    nRetry--;
                    tasks.offer(this);
                } else {
                    System.out.println("arg = " + arg + " tried " + m + " times, but failed.");
                }
            }
            return null;
        }
    }
}

しかしこれでも、失敗時のタスクキュー追加のタイミングによって次の問題がある。

invokeAll の直前に追加された場合、追加されたぶんも含めて invoke されてしまう
invokeAll の直後に追加された場合、そのタスクが消えてしまう

解決策として次の2通りがある。

発火前後を synchronized で同期化する
発火前に本当に発火すべきタスクを正しく集める

ここでは後者を採用する。

import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.ArrayList;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Random;

class Main {
    static final int n = 500;
    static final int m = 3;

    static int arg = 0;

    static LinkedBlockingQueue<Task> tasks = new LinkedBlockingQueue<Task>();

    static ExecutorService exe = Executors.newFixedThreadPool(n);

    public static void main(String[] args) {
        Timer timer = new Timer();
        Runner runner = new Runner();
        timer.scheduleAtFixedRate(runner, 0L, 1000L);
    }

    static class Runner extends TimerTask {
        public void run() {
            while (tasks.size() < n) {
                tasks.offer(new Task(arg++, m));
            }
            ArrayList<Task> toInvoke = new ArrayList<Task>();
            for (int i = 0; i < n; ++i) {
                toInvoke.add(tasks.poll());
            }
            try {
                exe.invokeAll(toInvoke);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    static class Task implements Callable<Void> {
        private int arg;
        private int nRetry;
        public Task(int arg, int nRetry) {
            this.arg = arg;
            this.nRetry = nRetry;
        }
        public Void call() {
            try {
                sendRequest(arg);
            } catch (Exception e) {
                if (nRetry > 0) {
                    nRetry--;
                    tasks.offer(this);
                } else {
                    System.out.println("arg = " + arg + " tried " + m + " times, but failed.");
                }
            }
            return null;
        }
    }
}

まとめと課題

失敗したときに何度かやり直すことを許すようなリクエストを送るときの QPS コントロールについて、問題を考察した。

単純にスレッドを作り実行するだけではいけない
失敗時に同じスレッドで実行すると QPS が狂う
スレッド使い捨て問題

これらについてそれぞれ解決策を示し、最終的にこの問題における設計パターンとコーディング例を示した。動作確認に使った動く完全なコード例を Gist にあげておく。

この実装の問題として、多くの場合毎秒処理されるタスクは1秒間のうち最初のほうに固まってしまい、いびつな間隔で実行されてしまうということがある。できれば 1 秒間のうちでうまく分散して実行されるようにしたい。

comments powered by Disqus

ページのトップへ戻る

daimatz.net

QPS コントロール

QPS

QPS コントロール

問題

タイマーを使った固定頻度実行

リトライ

スレッドプールを使う

まとめと課題