Распараллеливание с минимальными правками в коде

[@leoru]

Я не заметил разницы, если честно.

[@rfq]

Пост не дописан. Это глюк. Я нажимал «в черновики», а он опубликовался. Как его удалить?

[@Utter_step]

Зайти в редактирование, надать «В черновики» ещё раз

[@Pushkoff]

когда расставятся все нужные синхронизации станет хуже чем при однопоточном исполнении.

[@ffriend]

На черновик похоже.

[@KozzyKoder]

Правки действительно минимальны.

[@rfq]

Не знаю как вернуть пост в состояние черновика!

[@rfq]

написал в hangout на google+

[@Deeman]

Внизу редактируемой страницы эта кнопка :)

[@rfq]

Она была, но сейчас нету

[@ffriend]

Нажмите на «Редактировать пост» (синенький гаечный ключ справа от названия поста). В режиме просмотра таких кнопок, естественно, не будет.

[@anjensan]

Значит карма у вас такая.
Придется статью срочно дописывать.

[@rfq]

срочно дописал.

[@anjensan]

По мотивам этого топика можно сделать интересный пост «О том, как я убирал статью в черновики, или в поисках утраченной кнопки».

[@rfq]

Целый пост писать лень, но не могу не высказать удивления тому, что пока я косячил, было много ехидных комментариев, а когда благодаря ffriend статью дописал, все (пораженные?) замолчали.

[@voronaam]

Тривиальное решение просто. Но плюса заслуживает. Люблю когда велосипедов не изобретают.

Upd: по заголовку я было подумал что будет написан Java Agent, который модифицирует байткод класса с рапаралеливаемым методом.

[@anjensan]

Простите, но это типичнейший велосипед (своя реализация single thread executor).

[@anjensan]

Просто первая реакция после прочтения полной версии статьи «лучше бы и не дописывал». Статья уровня «ай, я нашел класс в стандартной библиотеке, надо срочно всем об этом рассказать». Тривиал.

И не могу не заметить, что это не первая Ваша статья на хабре.
Т.е. Вы по идее уже должны знать куда там кликать и что нажимать.

[@rfq]

Не класс, а способ, и не в стандартной библиотеке, а в малоизвестной. И способ, я считаю, нетривиальный. Далеко не любой программист выдумает его сходу, когда придет нужда. А так, глядишь, вспомнит.

[@Sauron]

Немного непонятно что за trueExecutor такой.

[@rfq]

Это реальный Executor, исполняющий задачи. SerialExecutor сам не исполняет, а передает на исполнение реальному в порядке очереди, так что в исполнении каждый раз не более одной задачи от данного SerialExecutor'а. Это позволяет не вводить синхронизацию в распараллеливаемый класс.

[@Sauron]

Вопрос в том, какая именно имплементация Executor предлагается для использования в качестве trueExecutor. Пойдет любой?
В коде то синтаксическая ошибка — trueExecutor не определен.

Если предполагается что можно передать любой Executor извне, может быть проблема с тем что передадут не какой-нибудь ThreadPoolExecutor, а уже SerialExecutor который враппит ThreadPoolExecutor, и получится двойной враппинг.

[@rfq]

Подойдет любой Executor. Даже SerialExecutor, хотя это и бессмысленно. Двойной враппинг замедлит исполнение, но не приведет к ошибкам.

[@anjensan]

Зачем Вы привели свою реализацию SingleThreadExecutor, она, скажем мягко, содержит грубые ошибки?
И чем вам не подошел вариант с java.util.concurrent.Executors.newSingleThreadExecutor?

[@rfq]

Мне не нужен был SingleThreadExecutor ни в каком виде, и я привел свою реализацию SerialExecutor. Это разные вещи. Вы считаете, что мой SerialExecutor содержит ошибки? Какие?

[@anjensan]

Для метода newSingleThreadExecutor в доках явно сказано «все задачи выполняются в одном потоке последовательно, не более одной одновременно». Ну и реализуется интерфейс ExecutorService, а не Executor. Первый является расширением второго, так что все ок. Напишите, пожалуйста, чем же Вам не подошел этот вариант?

Про ошибки: если в ваш SerialExecutor засылать таски из нескольких потоков, то таска может выполнится более одного раза, и даже одновременно.

[@rfq]

ответил в основной ветке.

[@rfq]

anjensan: При использовании SingleThreadExecutor на каждый экземпляр распараллеливаемого класса будет заводиться свой Thread. В общем случае это слишком накладно.

Насчет SerialExecutor и таски из нескольких потоков, там стоит synchronized(tasks) и ужасы, которые вы обещаете, исключены. Если не верите, напишите опровергающий тест.

[@anjensan]

Хм, ваша правда. Ужасы мне померещились :)

[@burjui]

Во-первых, если очень хочется написать свой SerialExecutor, как недавно сделал я по незнанию, используйте LinkedBlockingQueue — тогда synchronized (tasks) не нужно:
docs.oracle.com/javase/6/docs/api/java/util/concurrent/LinkedBlockingQueue.html

Во-вторых, вот:

public class SerialExecutor {
    private static final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public static void execute(final Runnable task) {
        executorService.submit(task);
    }
}

class ServiceWrapper extends Service {
   public void longJob(final Object arg) {
       SerialExecutor.execute(new Runnable() {
              public void run() {
                   ServiceWrapper.super.longJob(arg);
               }
        });
   }
}

Всё, у вас теперь только один Thread, как вы и хотели. Меньше кода, эффект тот же.

[@rfq]

Если бы я хотел только один Thread, то я бы вовсе не определял SerialExecutor, а использовал бы результат newSingleThreadExecutor напрямую, и кода было бы еще меньше:

class ServiceWrapper extends Service {
   private static final ExecutorService executorService = Executors.newSingleThreadExecutor();

   public void longJob(final Object arg) {
       executorService.execute(new Runnable() {
              public void run() {
                   ServiceWrapper.super.longJob(arg);
               }
        });
   }
}

Но мне SingleThreadExecutor не был нужен, как я и объяснял выше anjensan. Мне нужно чтобы потоков использовалась оптимальное число — чтобы использовать все процессоры, но не плодить по потоку на объект — объектов может быть слишком много. Описать создание такого Executor'а, подходящего на все случаи жизни, невозможно, поэтому его создание оставлено пользователю.

[@burjui]

Видмо, я вас неправильно понял. Тогда можно использовать Executors.newFixedThreadPool(int nThreads), сообщая ему число ядер/процессоров.

[@rfq]

Использование LinkedBlockingQueue или любой другой синхронизированной коллекции менее оптимально, так как требует на обработку каждой задачи 2 обращения к синхронизированному объекту (сначала взять без удаления и только после обработки удалить), вместо одного у меня. Дело в том, что нужно поддерживать признак «объект уже в работе, не передавать Runnable в Executor».

[@burjui]

Что-то я не вкурил, зачем брать задачу из очереди без удаления. Поясните, я действительно не понимаю. Лично мне кажется, что вы неправильно понимаете суть LinkedBlockingQueue. Нормальное её использование предполагает извлечение из очереди объекта и работу с ним. При этом, если очередь пустая, поток, извлекающий данные, замораживается до поступления в очередь новых данных. Добавление новых данных в очередь не замораживает поток, из которого данные добавляются. Если вы передаёте Executor своему SerialExecutor'у с целью выбирать между однопоточной или многопоточной обработкой задач, тогда вообще не нужно писать никакие SerialExecutor'ы, а просто воспользоваться или Executors.newSingleThreadExecutor(), или Executors.newFixedThreadPool(int nThreads). Они сами позаботятся о синхронизации своих очередей, и всё, что вам нужно — это дать им задачу в одном из двух видов:

1. Runnable, если нужно просто закинуть задачу на выполнение.
2. Callable, если нужен результат вычислений. Его можно получить так:
   
   

   class Service {
     public String /* чисто для примера */ longJob(Object arg) {...}
   }
   ...
   final Future<Result> future = someExecutorService.submit(new Callable<Result>() {
       public String call() {
           return someService.longJob(arg);
       }
   });
   // отдаёте ваш future, куда надо
   ...
   // Где надо:
   final String result = future.get();
   


Я всё ещё не понял, какие такие специальные задачи вы решаете, что вам не подходит всё богатство стандартной библиотеки Java.

[@rfq]

Брать задачу из очереди без удаления нужно для того, чтобы показать потоку-поставщику, что SerialExecutor уже работает и не нужно запускать его второй раз. Можно для описания этого состояния завести отдельную переменную, но тогда надо синхронизироваться одновременно по очереди и этой переменной, и внутренняя синхронизация LinkedBlockingQueue оказывается неиспользуемой. Собственно, я так и сделал, и вместо блокирующей очереди использовал несихронизированный LinkedList.

Задача же была такая: распараллелить исполнение отдельных методов с использованием пула потоков, а не расходуя по потоку на вызов метода. При этом вызовы методов, относящихся к одному и тому же объекту, в том числе повторные вызовы одного и того же метода, не должны пересекаться по времени, так как иначе придется вводить дополнительную синхронизацию по доступу к полям объекта.

[@rfq]

И насчет богатства стандартной библиотеки — это богатство имеет существенный изъян — нет средств организации массива задач, исполняемых на пуле потоков. Есть только возможность запустить задачу и дождаться ее завершения с помощью Future, но так ждать можно лишь из потока, а не из задачи, иначе задача заблокирует пуловский поток, что приведет к дедлоку ограниченный пул (thread starvation), а неограниченый пул приведет к конфигурации «поток на задачу», что противоречит самой идее пула потоков. SerialExecutor и призван закрыть одну из дыр стандартной библиотеки.