11

Eu tenho uma aplicação que cria diversos objetos em cima de um stream de funções. E então eu coleto todos esses objetos gerados em um acumulador. Por exemplo, se eu gerasse strings e acumulasse em uma lista, faria uma coisa assim:

List<Function<T, String>> funcoes = /* inicializa com as diversas funções */;

T objetoSobProcessamento = /* parâmetro arbitrário */
List<String> retornoProcessamento = funcoes.stream().map(f -> f.apply(objetoSobProcessamento))
  .reduce(new ArrayList<>(), (listaAcumulada, novoValor) -> {
            listaAcumulada.add(novoValor);
            return listaAcumulada;
          }, (l1, l2) -> {
            l1.addAll(l2);
            return l1;
          });

O caso real é com outros objetos, ligeiramente mais complexos, mas segue o mesmo espírito

Ao olhar as possíveis operações de reduce em Java, me veio uma dúvida sobre a paralelização da operação de redução, no caso genérico. Olhar essa resposta me deixou com a impressão de que essa operação de reduce deveria ser sequencial, não paralelizável.

Então, minha pergunta é:

  • na operação de acumulação, para aproveitar ao máximo o paralelismo, devo produzir sempre um objeto novo? ou devo sempre reaproveitar o objeto? ou deveria apenas evitar reusar o objeto identidade, gerando novos objetos quando se acumula a primeira vez e reaproveitando esse novo elemento gerado nas conseguintes?

Eu sei que a operação de combinação/merge dos objetos obtidos é perfeitamente paralelizável se e somente se os argumentos passados a ela sejam únicos, como no caso ilustrado abaixo:

t1--A--ui  t2--A--ui  t3--A--ui  t4--A--ui
    |          |          |          |
    u1----M----u2         u3----M----u4
          |                     |
          u5---------M----------u6
                     |
                     u7

Onde A é uma operação de acumulação, M é uma operação de de merge, t? é um objeto do tipo da entrada T, u? é um objeto do tipo do retorno U e ui é o objeto de identidade.

1 Resposta 1

6
+50

Bem, você não disse o que você usa como a classe T e nem o que são essas funções que operam sobre ela para produzir Strings. Então vou inventar uma classe para isso:

class Colorido {
    private void complica() {
        try {
            Thread.sleep((int) (Math.random() * 50));
        } catch (InterruptedException e) {
            // Ignora.
        }
    }
    public String getVermelho() { complica(); return "vermelho"; }
    public String getLaranja()  { complica(); return "laranja";  }
    public String getAmarelo()  { complica(); return "amarelo";  }
    public String getLima()     { complica(); return "lima";     }
    public String getVerde()    { complica(); return "verde";    }
    public String getCiano()    { complica(); return "ciano";    }
    public String getAzul()     { complica(); return "azul";     }
    public String getVioleta()  { complica(); return "violeta";  }
    public String getRosa()     { complica(); return "rosa";     }
    public String getRoxo()     { complica(); return "roxo";     }
    public String getBranco()   { complica(); return "branco";   }
    public String getPreto()    { complica(); return "preto";    }
    public String getMarrom()   { complica(); return "marrom";   }
    public String getOliva()    { complica(); return "oliva";    }
    public String getBege()     { complica(); return "bege";     }
    public String getCinza()    { complica(); return "cinza";    }
    public static List<Function<Colorido, String>> funcoes() {
        return Arrays.asList(
            Colorido::getVermelho, Colorido::getLaranja, Colorido::getAmarelo, Colorido::getLima,
            Colorido::getVerde, Colorido::getCiano, Colorido::getAzul, Colorido::getVioleta,
            Colorido::getRosa, Colorido::getRoxo, Colorido::getBranco, Colorido::getPreto,
            Colorido::getMarrom, Colorido::getOliva, Colorido::getBege, Colorido::getCinza
        );
    }
}

Esta é decerto uma classe bem boba. No entanto, observe que o método funcoes vai devolver uma lista com 16 funções que operam sobre o objeto. Cada função retorna o nome de uma cor diferente, mas todas elas vão demorar algum tempinho para fazer isso graças ao Thread.sleep de tempo aleatório dentro do método complica. Afinal de contas, se queremos testar paralelismo, é importante que as threads não executem de forma demasiadamente rápida para evitarmos que uma comece quando a outra já terminou.

Então, vamos testar o map-reduce que você colocou na pergunta:

public static <T> List<String> mapReduceSequencial(
    List<Function<T, String>> funcoes, // inicializa com as diversas funções
    T objetoSobProcessamento) // parâmetro arbitrário
{
    List<String> retornoProcessamento = funcoes
            .stream()
            .map(f -> f.apply(objetoSobProcessamento))
            .reduce(new ArrayList<>(), (listaAcumulada, novoValor) -> {
                listaAcumulada.add(novoValor);
                return listaAcumulada;
            }, (l1, l2) -> {
                l1.addAll(l2);
                return l1;
            });
    return Collections.unmodifiableList(retornoProcessamento);
}

Ele é invocado assim:

System.out.println(mapReduceSequencial(Colorido.funcoes(), new Colorido()));

Eis a saída (acrescentei a quebra de linha só para facilitar a leitura):

[vermelho, laranja, amarelo, lima, verde, ciano, azul, violeta,
 rosa, roxo, branco, preto, marrom, oliva, bege, cinza]

Observe que a ordem das cores é a mesma da lista.

E se alterarmos aquele .stream() para .parallelStream(), o que acontece? Neste caso, o resultado vai variar por causa do Thread.sleep aleatório e por causa do indeterminismo inerente de ver várias threads simultâneas, mas em um teste que fiz o resultado foi esse:

[verde, rosa, roxo, ciano, azul, branco, violeta, preto,
vermelho, laranja, amarelo, marrom, oliva, lima, verde, rosa,
roxo, ciano, azul, branco, violeta, preto, vermelho, laranja,
amarelo, marrom, oliva, lima, bege, cinza, verde, rosa,
roxo, ciano, azul, branco, violeta, preto, vermelho, laranja,
amarelo, marrom, oliva, lima, verde, rosa, roxo, ciano,
azul, branco, violeta, preto, vermelho, laranja, amarelo, marrom,
oliva, lima, bege, cinza, verde, rosa, roxo, ciano,
azul, branco, violeta, preto, vermelho, laranja, amarelo, marrom,
oliva, lima, verde, rosa, roxo, ciano, azul, branco,
violeta, preto, vermelho, laranja, amarelo, marrom, oliva, lima,
bege, cinza, verde, rosa, roxo, ciano, azul, branco,
violeta, preto, vermelho, laranja, amarelo, marrom, oliva, lima,
verde, rosa, roxo, ciano, azul, branco, violeta, preto,
vermelho, laranja, amarelo, marrom, oliva, lima, bege, cinza]

Bem, esse resultado claramente está muito errado. Isso significa que a implementação do seu reduce não funciona em Streams paralelos (mas tenha calma que daqui a pouco explico como consertar isso). Observe que há várias subsequências se repetindo nesse resultado aí.

Olhando a documentação do método Stream<T>.reduce(U, BiFunction<U, ? super T, U>, BinaryOperator<U>) cujos parâmetros são chamados respectivamente de identity, accumulator e combiner, encontramos isso:

This is equivalent to:

 U result = identity;
 for (T element : this stream)
     result = accumulator.apply(result, element)
 return result;

but is not constrained to execute sequentially.

Traduzindo para o português:

Isto é equivalente a [código] mas não limitado a executar sequencialmente.

Vamos dar uma olhada primeiro nessa linha:

U result = identity;

Isso significa que sua ArrayList é o primeiro valor de result.

Agora essa linha dentro do for:

    result = accumulator.apply(result, element)

Isso significa que a partir da lista dada, uma outra lista será produzida e com base nela uma outra lista e assim sucessivamente até os elementos acabarem. Ocorre que no seu caso a lista produzida é sempre a mesma, e portanto ao invés de você produzir novos resultados com base nos resultados parciais anteriores, você está na verdade modificando os resultados parciais anteriores e os devolvendo como se fossem novos.

O resultado é que ao fazer isso em paralelo, quando o combinador for utilizado, você vai ter que l1 e l2 apontam para o mesmo objeto, que é a ArrayList, e com isso, quando você chama l1.addAll(l2), você vai adicionar a lista a ela mesma, o que significa que vai duplicar todos os elementos existentes nela! Junte isso com o fato das threads em paralelo estarem demorando tempos aleatórios/imprevisíveis e o resultado é aquela bagunça que foi produzida.

Voltando lá no javadoc, também está escrito isso:

This means that for all u, combiner(identity, u) is equal to u.

Traduzindo:

Isso significa que para todo u, combiner(identity, u) é igual a u.

Isso significa, de uma forma algébrica, que identity deve ser o elemento neutro da operação combiner, assim como 0 é o elemento neutro da soma e 1 é o elemento neutro da multiplicação. Qualquer coisa combinada com identity deve resultar nessa mesma coisa.

O combiner é o seu último lambda que faz o addAll. Em seus dois lambdas você altera a lista, e portanto você altera o que é considerado identity. Logo, ao invocar combiner(identity, u) você acaba tendo uma lista que não é igual ao que o u era antes e sim uma que equivale ao u dobrado. Isso significa que os seus lambdas violam as regras definidas pela documentação do reduce ao violar a regra do elemento neutro do combiner. Assim sendo, é óbvio que coisas ruins vão acontecer.

Podemos alterar algumas coisas nos seus lambdas para restaurar o comportamento/contrato exigido pelo reduce:

public static <T> List<String> mapReduceParalelo2(
    List<Function<T, String>> funcoes, // inicializa com as diversas funções
    T objetoSobProcessamento) // parâmetro arbitrário
{
    List<String> retornoProcessamento = funcoes
            .parallelStream()
            .map(f -> f.apply(objetoSobProcessamento))
            .reduce(Collections.emptyList(), (listaAcumulada, novoValor) -> {
                List<String> novaLista = listaAcumulada.isEmpty() ? new ArrayList<>() : listaAcumulada;
                novaLista.add(novoValor);
                return novaLista;
            }, (l1, l2) -> {
                if (l1.isEmpty()) return l2;
                if (l2.isEmpty()) return l1;
                l1.addAll(l2);
                return l1;
            });
    return Collections.unmodifiableList(retornoProcessamento);
}

O nome do método é mapReduceParalelo2 porque eu vou chamar o que produziu o resultado bagunçado de mapReduceParalelo1. No método parReduceParalelo2, note que o identity é Collections.emptyList(). Fiz isso para enfatizar que o identity não deve ser alterado (tentar fazer isso resultaria numa exceção).

O accumulator testa se a lista dada é um identity para evitar problemas, criando uma nova se não for. Se a lista dada não for um identity, então ela pode ser seguramente alterada pois a lista original não seria reutilizada em lugar nenhum se uma nova fosse criada.

Já o combiner, verifica se uma das listas é um identity, retornando a outra neste caso. Se nenhuma delas for o identity, então não há problema em fazer-se o addAll ao invés de criar uma nova pois nenhuma das listas originais seriam reutilizadas depois.

A moral da história é que o identity sempre é reutilizado, enquanto os demais resultados parciais não são. Se o identity fosse sempre recriado ao invés de reutilizado, o mapReduceParalelo1 funcionaria.

Isso pode ter ficado meio complicado porque na verdade o reduce foi projetado para trabalhar com dados imutáveis, ao menos com um identity imutável. Embora seja possível usar o reduce para coisas mutáveis, o resultado é esse troço meio complicado e confuso que saiu. Já o collect é diferente. O collect foi de fato projetado para trabalhar com coisas mutáveis. Por exemplo:

public static <T> List<String> mapCollectParalelo(
    List<Function<T, String>> funcoes, // inicializa com as diversas funções
    T objetoSobProcessamento) // parâmetro arbitrário
{
    List<String> retornoProcessamento = funcoes
            .parallelStream()
            .map(f -> f.apply(objetoSobProcessamento))
            .collect(ArrayList::new, List::add, List::addAll);
    return Collections.unmodifiableList(retornoProcessamento);
}

Observe a simplicidade dos parâmetros do método collect. Observe que todos os três parâmetros são method-references. O primeiro, ao invés de ser um objeto fixo para servir de identity sempre sendo reutilizado, é o Supplier que é usado para criar novas listas, no caso o construtor de ArrayList. O segundo parâmetro (accumulator) é o que adiciona elementos à lista, ou seja o método add (e o resultado dele é void, pois a operação esperada aqui é uma mutação). O terceiro parâmetro (combiner) que combina elementos é o addAll (também void). Observe que é usado parallelStream(). O resultado é o que se espera (e se fosse sequencial, também é):

[vermelho, laranja, amarelo, lima, verde, ciano, azul, violeta,
 rosa, roxo, branco, preto, marrom, oliva, bege, cinza]

Talvez você perceba que juntar ArrayList::new, List::add e List::addAll seja um caso bastante frequente e corriqueiro. Será que não dá para encapsular esses três num objeto só? Bem, é para isso que serve a interface Collector. E o método collect(Collector) é a versão de collect que utiliza tal interface. Para obter então um Collector que cria a lista usando esses três parâmetros, basta usar o método Collectors.toList(). Logo, você pode usar o collect simplesmente assim:

    List<String> retornoProcessamento = funcoes
            .parallelStream()
            .map(f -> f.apply(objetoSobProcessamento))
            .collect(Collectors.toList());

Esse último deve ser o código que você queria. Observe que é muito mais simples e direto do que o que você postou em sua pergunta!

Coloquei o código completo e compilável no GitHub.

Por fim, quanto à resposta do OnoSendai que você linkou na pergunta, aquilo é sobre JavaScript, e não Java. O modelo do JavaScript é totalmente single-threaded, então não há muito o que se falar de paralelismo lá.

1
  • 1
    Eu creio que você tenha matado a minha próxima pergunta: quando usar reduce e quando usar collect Commented 23/07/2018 às 9:55

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .