Почему многопоточная версия программы работает дольше?

Question

[Bonce]

Здравствуйте! Осваиваю многопоточность на джаве. Решаю простую задачу: есть входной поток, нужно его продублировать в 2 разных выходных потока. Мое решение:

private static class Tee {
        private BufferedReader reader;
        private PrintWriter fWriter;
        private PrintWriter sWriter;
        boolean done;
        String buffer;
        int readCount;
        int firstWriteCount;
        int secondWriteCount;

        public Tee(BufferedReader reader, PrintWriter fWriter, PrintWriter sWriter) {
            this.reader = reader;
            this.fWriter = fWriter;
            this.sWriter = sWriter;
        }

        public void teeWhileInsideCo() throws InterruptedException {
            Thread reader = new Thread(new LineReader());
            Thread fWriter = new Thread(new LineWriter(this.fWriter, 0));
            Thread sWriter = new Thread(new LineWriter(this.sWriter, 1));

            reader.start();
            fWriter.start();
            sWriter.start();

            reader.join();
            fWriter.join();
            sWriter.join();
        }

        private class LineReader implements Runnable {

            @Override
            public void run() {
                String line;
                while (true) {
                    try {
                        line = reader.readLine();
                    } catch (IOException e) {
                        line = null;
                    }

                    while (readCount != firstWriteCount || readCount != secondWriteCount) {
                        try {
                            Thread.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    if (line == null) {
                        done = true;
                        break;
                    }

                    buffer = line;
                    readCount++;
                }
            }
        }

        private class LineWriter implements Runnable {
            private PrintWriter writer;
            private int number;

            public LineWriter(PrintWriter writer, int n) {
                System.out.println(Tee.this);
                this.writer = writer;
                this.number = n;
            }

            @Override
            public void run() {
                while (true) {
                    while (!done && ((number == 0 ? firstWriteCount : secondWriteCount) == readCount)) {
                        try {
                            Thread.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    if (done) {
                        break;
                    }

                    writer.println(buffer);
                    if (number == 0) {
                        firstWriteCount++;
                    } else {
                        secondWriteCount++;
                    }
                }
                writer.flush();
            }
        }
    }

На тестовых данных данная версия работает в 30 раз дольше, чем последовательная версия программы. Подскажите, что я делаю не так? Как именно надо стрелять, чтобы не попадать себе в ногу?

Comments

[Dmtm]

вместо buffer - пару очередей бы, на каждого потребителя - свою

[antego]

Вы жалуетесь на производительность, но при этом специально останавливаете потоки. Попробуйте убрать thread.sleep

Судить про корректность работы вашего кода нельзя, он полон гонок. Изучайте, что такое synchronized и volatile.

[Bonce]

antego, добавил на примитивы volatile. Код стал выполняться быстро и тесты стали проходить

[Bonce]

antego, можете указать на состояние гонки? я с многопоточностью очень слабо знаком. Начал прочтение книги, но там без привязки к языку.
Здесь как я вижу основная проблема в условиях циклов while, которые выполняют что-то вроде await по предикату. Видимо, без volatile потоки не так читали и писали в примитивы, как того бы хотелось

[antego]

Bonce, все гонки расписывать не буду, потому что их много.

Например readCount - если оно не обозначено как volatile, то потоки LineWriter могут не увидеть записи из LineReader и будут читать данные из кэша цпу.

[Bonce]

antego, то есть ситуация, когда в одном потоке readCount=n, а другие потоки по-прежнему считают readCount=(n-1). Спасибо.