„Konstrukcja uniwersalna” jest klasą opakowania dla obiektu sekwencyjnego, która umożliwia jego linearyzację (silny warunek spójności dla współbieżnych obiektów). Na przykład, tutaj jest dostosowana konstrukcja bez oczekiwania, w Javie, z [1], która zakłada istnienie kolejki bez oczekiwania, która spełnia interfejs WFQ(która wymaga tylko jednorazowej zgody między wątkami) i zakłada Sequentialinterfejs:

public interface WFQ<T> // "FIFO" iteration
{
    int enqueue(T t); // returns the sequence number of t
    Iterable<T> iterateUntil(int max); // iterates until sequence max
}
public interface Sequential
{
    // Apply an invocation (method + arguments)
    // and get a response (return value + state)
    Response apply(Invocation i); 
}
public interface Factory<T> { T generate(); } // generate new default object
public interface Universal extends Sequential {}

public class SlowUniversal implements Universal
{
    Factory<? extends Sequential> generator;
    WFQ<Invocation> wfq = new WFQ<Invocation>();
    Universal(Factory<? extends Sequential> g) { generator = g; } 
    public Response apply(Invocation i)
    {
        int max = wfq.enqueue(i);
        Sequential s = generator.generate();
        for(Invocation invoc : wfq.iterateUntil(max))
            s.apply(invoc);
        return s.apply(i);
    }
}

Ta implementacja nie jest zbyt satysfakcjonująca, ponieważ jest naprawdę powolna (pamiętasz każde wywołanie i musisz odtwarzać je przy każdym zastosowaniu - mamy liniowy czas działania w historii). Czy jest jakiś sposób, żebyśmy mogli rozszerzyć WFQiSequential interfejsy (w rozsądny sposób), abyśmy mogli zapisać pewne kroki podczas stosowania nowego wywołania?

Czy możemy uczynić to bardziej wydajnym (nie liniowym środowiskiem uruchomieniowym w historii, najlepiej zużycie pamięci również maleje) bez utraty właściwości bez czekania?

Wyjaśnienie

„Uniwersalna konstrukcja” to termin, który jestem pewien, że został stworzony przez [1], który akceptuje niebezpieczny, ale zgodny z wątkiem obiekt, który jest generalizowany przez Sequentialinterfejs. Korzystając z kolejki bez oczekiwania, pierwsza konstrukcja oferuje bezpieczną dla wątków, linearyzowalną wersję obiektu, która jest również wolna od oczekiwania (zakłada to determinizm i applyoperacje zatrzymania ).

Jest to nieefektywne, ponieważ metoda polega na tym, że każdy wątek lokalny zaczyna się od czystej tablicy i stosuje każdą zarejestrowaną operację. W każdym razie działa to, ponieważ skutecznie synchronizuje się, używając WFQdo określenia kolejności, w której należy zastosować wszystkie operacje: każde wywołanie wątku applyzobaczy ten sam Sequentialobiekt lokalny , z tą samą sekwencjąInvocation s zastosowaną do niego.

Moje pytanie brzmi, czy możemy (np.) Wprowadzić proces czyszczenia w tle, który aktualizuje „stan początkowy”, abyśmy nie musieli restartować od zera. Nie jest to tak proste, jak posiadanie wskaźnika atomowego ze wskaźnikiem początkowym - tego rodzaju podejścia łatwo tracą gwarancję bez czekania. Podejrzewam, że może tu działać inne podejście oparte na kolejce.

Żargon:

1. bez czekania - bez względu na liczbę wątków lub podejmowanie decyzji przez program planujący, applyzakończy się możliwą do udowodnienia ograniczoną liczbą instrukcji wykonanych dla tego wątku.
2. lock-free - jak wyżej, ale dopuszcza możliwość nieograniczonego czasu wykonania, tylko w przypadku, gdy nieograniczona liczba applyoperacji jest wykonywana w innych wątkach. Zazwyczaj optymistyczne schematy synchronizacji należą do tej kategorii.
3. blokowanie - wydajność na łasce harmonogramu.

Działający przykład, zgodnie z żądaniem (teraz na stronie, która nie wygasa)

[1] Herlihy and Shavit, The Art of Multiprocessor Programming .

Oto wyjaśnienie i przykład, jak to osiągnąć. Daj mi znać, jeśli są części, które nie są jasne.

Gist ze źródłem

uniwersalny

Inicjalizacja:

Indeksy wątków są stosowane w sposób przyrostowy atomowy. Jest to zarządzane za pomocą AtomicIntegernazwanego nextIndex. Indeksy te są przypisywane do wątków poprzez ThreadLocalinstancję, która inicjuje się, pobierając następny indeks nextIndexi zwiększając go. Dzieje się tak za pierwszym razem, gdy indeks każdego wątku jest pobierany po raz pierwszy. Utworzono A w ThreadLocalcelu śledzenia ostatniej sekwencji utworzonej przez ten wątek. Zainicjowano 0. Sekwencyjne odniesienie do obiektu fabryki jest przekazywane i przechowywane. AtomicReferenceArrayTworzone są dwa wystąpienia wielkości n. Obiekt ogona jest przypisany do każdego odniesienia, po zainicjowaniu go ze stanu początkowego dostarczonego przez Sequentialfabrykę. nto maksymalna dozwolona liczba wątków. Każdy element w tych tablicach „należy” do odpowiedniego indeksu wątku.

Zastosuj metodę:

Jest to metoda, która wykonuje interesującą pracę. Wykonuje następujące czynności:

• Utwórz nowy węzeł dla tego wywołania: moje
• Ustaw ten nowy węzeł w tablicy zapowiedzi przy indeksie bieżącego wątku

Następnie rozpoczyna się pętla sekwencjonowania. Będzie on działał, dopóki bieżące wywołanie nie zostanie zsekwencjonowane:

1. znajdź węzeł w tablicy zapowiedzi, używając sekwencji ostatniego węzła utworzonego przez ten wątek. Więcej na ten temat później.
2. jeśli węzeł zostanie znaleziony w kroku 2, nie jest jeszcze zsekwencjonowany, kontynuuj go, w przeciwnym razie po prostu skup się na bieżącym wywołaniu. Spróbuje to pomóc tylko jednemu innemu węzłowi na każde wywołanie.
3. Niezależnie od tego, który węzeł został wybrany w kroku 3, próbuj sekwencjonować go po ostatnim zsekwencjonowanym węźle (inne wątki mogą się zakłócać). Niezależnie od sukcesu ustaw bieżące odniesienie nagłówka wątku do sekwencji zwróconej przez decideNext()

Kluczem do opisanej powyżej pętli zagnieżdżonej jest decideNext()metoda. Aby to zrozumieć, musimy spojrzeć na klasę Node.

Klasa węzłowa

Ta klasa określa węzły na liście podwójnie połączonej. W tej klasie nie ma dużo akcji. Większość metod to proste metody wyszukiwania, które powinny być dość oczywiste.

metoda ogona

zwraca to specjalną instancję węzła o sekwencji 0. Po prostu działa jako symbol zastępczy, dopóki wywołanie nie zastąpi jej.

Właściwości i inicjalizacja

• seq: numer sekwencyjny, zainicjowany na -1 (co oznacza, że ​​nie jest porządkowany)
• invocation: wartość wywołania apply(). Ustawiony na budowę.
• next: AtomicReferencedla linku do przesyłania dalej. raz przypisany, nigdy nie zostanie zmieniony
• previous: AtomicReferencedla linku wstecznego przypisanego podczas sekwencjonowania i wyczyszczonego przeztruncate()

Wybierz następny

Ta metoda jest tylko jedna w węźle z nietrywialną logiką. W skrócie, węzeł jest oferowany jako kandydat na następny węzeł na połączonej liście. compareAndSet()Metoda sprawdzi, czy jest to odniesienie jest zerowy, a jeśli tak, to ustawić odniesienie do kandydata. Jeśli odwołanie jest już ustawione, nie robi nic. Ta operacja ma charakter atomowy, więc jeśli dwóch kandydatów zostanie zaoferowanych w tym samym momencie, zostanie wybrany tylko jeden. To gwarantuje, że tylko jeden węzeł zostanie wybrany jako następny. Jeśli wybrany zostanie węzeł kandydujący, jego kolejność zostanie ustawiona na następną wartość, a jego poprzednie łącze zostanie ustawione na ten węzeł.

Powrót do klasy uniwersalnej Zastosuj metodę ...

Po wywołaniu decideNext()ostatniego zsekwencjonowanego węzła (po sprawdzeniu) albo naszym węzłem, albo węzłem z announcetablicy, istnieją dwa możliwe przypadki: 1. Węzeł został pomyślnie zsekwencjonowany 2. Niektóre inne wątki uprzedziły ten wątek.

Następnym krokiem jest sprawdzenie, czy węzeł utworzony dla tego wywołania. Może się tak zdarzyć, ponieważ ten wątek z powodzeniem zsekwencjonował go lub inny wątek podniósł go z announcetablicy i zsekwencjonował dla nas. Jeśli nie został zsekwencjonowany, proces jest powtarzany. W przeciwnym razie wywołanie kończy się przez wyczyszczenie tablicy announce dla indeksu tego wątku i zwrócenie wartości wynikowej wywołania. Tablica zapowiedzi została wyczyszczona, aby zagwarantować, że nie pozostały żadne odniesienia do węzła, które uniemożliwiłyby zbieranie śmieci i dlatego wszystkie węzły na liście połączonej od tego momentu pozostają aktywne na stosie.

Oceń metodę

Teraz, gdy węzeł wywołania został pomyślnie zsekwencjonowany, wywołanie musi zostać ocenione. Aby to zrobić, pierwszym krokiem jest upewnienie się, że wywołania poprzedzające to zostały ocenione. Jeśli tego nie zrobią, ten wątek nie będzie czekał, ale wykona tę pracę natychmiast.

Metoda surePrior

ensurePrior()Sposób to działa sprawdzając poprzedni węzeł w połączonej listy. Jeśli jego stan nie jest ustawiony, poprzedni węzeł zostanie oceniony. Węzeł, że jest to rekurencyjne. Jeśli węzeł poprzedzający poprzedni węzeł nie został oceniony, wywoła funkcję oceny dla tego węzła i tak dalej.

Teraz, gdy wiadomo, że poprzedni węzeł ma stan, możemy go ocenić. Ostatni węzeł jest pobierany i przypisywany do zmiennej lokalnej. Jeśli to odwołanie ma wartość null, oznacza to, że jakiś inny wątek uprzedził ten i ocenił już ten węzeł; ustawienie to stan. W przeciwnym razie stan poprzedniego węzła jest przekazywany do Sequentialmetody stosowania obiektu wraz z wywołaniem tego węzła. Zwrócony stan jest ustawiany w węźle i truncate()wywoływana jest metoda, usuwając łącze zwrotne z węzła, ponieważ nie jest już potrzebne.

Metoda MoveForward

Metoda przesuwania do przodu spróbuje przenieść wszystkie odniesienia do tego węzła, jeśli nie wskazują już czegoś dalej. Ma to na celu zapewnienie, że jeśli wątek przestanie dzwonić, jego głowa nie zachowa odniesienia do węzła, który nie jest już potrzebny. compareAndSet()Metoda będzie upewnić się, że tylko zaktualizować węzeł jeśli jakiś inny wątek nie zmienił się, ponieważ została pobrana.

Ogłoś tablicę i pomoc

Kluczem do uczynienia tego podejścia bez czekania, a nie po prostu bez blokady, jest to, że nie możemy zakładać, że harmonogram wątków da pierwszeństwo każdemu wątkowi, gdy będzie tego potrzebował. Jeśli każdy wątek po prostu spróbuje zsekwencjonować swoje własne węzły, możliwe jest, że wątek może być ciągle opróżniany pod obciążeniem. Aby uwzględnić tę możliwość, każdy wątek najpierw spróbuje „pomóc” innym wątkom, których sekwencjonowanie może nie być możliwe.

Podstawową ideą jest to, że ponieważ każdy wątek z powodzeniem tworzy węzły, przypisane sekwencje rosną monotonicznie. Jeśli wątek lub wątki ciągle przeczą innemu wątkowi, indeks używany do znajdowania niesekwencjonowanych węzłów w announcetablicy przesunie się do przodu. Nawet jeśli każdy wątek, który obecnie próbuje sekwencjonować dany węzeł, jest stale przejmowany przez inny wątek, ostatecznie wszystkie wątki będą próbowały sekwencjonować ten węzeł. Aby to zilustrować, skonstruujemy przykład z trzema wątkami.

W punkcie początkowym wszystkie głowy i trzy elementy wątku są wskazywane w tailwęźle. lastSequenceDla każdej nici 0.

W tym momencie Wątek 1 jest wykonywany za pomocą wywołania. Sprawdza tablicę zapowiedzi pod kątem jej ostatniej sekwencji (zero), która jest węzłem, który ma obecnie indeksować. Sekwencjonuje węzeł i lastSequencejest ustawiony na 1.

Wątek 2 jest teraz wykonywany z wywołaniem, sprawdza tablicę zapowiedzi w ostatniej sekwencji (zero) i widzi, że nie potrzebuje pomocy, więc próbuje sekwencjonować swoje wywołanie. Udaje się, a teraz lastSequencejest ustawiony na 2.

Wątek 3 jest teraz wykonywany, a także widzi, że węzeł w announce[0]jest już zsekwencjonowany i sekwensuje własne wywołanie. Teraz lastSequencejest ustawiony na 3.

Teraz wątek 1 jest wywoływany ponownie. Sprawdza tablicę ogłoszeń w indeksie 1 i stwierdza, że ​​jest już zsekwencjonowana. Jednocześnie wywoływany jest wątek 2 . Sprawdza tablicę ogłoszeń w indeksie 2 i stwierdza, że ​​jest już zsekwencjonowana. Zarówno Wątek 1, jak i Wątek 2 próbują teraz sekwencjonować własne węzły. Wątek 2 wygrywa i sekwencjonuje jego wywołanie. To lastSequencejest ustawiony na 4. Tymczasem trzy wątek został wywołany. Sprawdza indeks lastSequence(mod 3) i stwierdza, że ​​węzeł w announce[0]nie został zsekwencjonowany. Wątek 2 jest ponownie wywoływany w tym samym czasie, gdy wątek 1 jest podczas drugiej próby. Wątek 1znajduje nie wywoływane sekwencyjne wywołanie, w announce[1]którym jest węzeł właśnie utworzony przez wątek 2 . Próbuje sekwencjonować wywołanie wątku 2 i kończy się powodzeniem. Wątek 2 znajduje swój własny węzeł announce[1]i został zsekwencjonowany. Ustawia to na lastSequence5. Wątek 3 jest następnie wywoływany i stwierdza, że ​​węzeł, w którym umieszczony announce[0]jest wątek 1, nadal nie jest sekwencjonowany i próbuje to zrobić. W międzyczasie wywołano także wątek 2, który uprzedza wątek 3. Sekwencjonuje swój węzeł i ustawia go na lastSequence6.

Słaby wątek 1 . Mimo że wątek 3 próbuje go sekwencjonować, oba wątki są ciągle udaremniane przez program planujący. Ale w tym momencie. Wątek 2 wskazuje teraz również na announce[0](6 mod 3). Wszystkie trzy wątki są ustawione tak, aby próbować sekwencjonować to samo wywołanie. Bez względu na to, który wątek się powiedzie, następnym węzłem, który ma być zsekwencjonowany, będzie oczekujące wywołanie wątku 1, tj. Węzła, do którego się odwołuje announce[0].

To jest nieuniknione. Aby wątki zostały uprzednio opróżnione, inne wątki muszą być węzłami sekwencjonowania, a gdy to robią, będą stale poruszać się do lastSequenceprzodu. Jeśli węzeł danego wątku nie jest ciągle sekwencjonowany, ostatecznie wszystkie wątki będą wskazywały na jego indeks w tablicy zapowiedzi. Żaden wątek nie zrobi nic innego, dopóki węzeł, któremu próbuje pomóc, nie zostanie zsekwencjonowany, najgorszym scenariuszem jest to, że wszystkie wątki wskazują ten sam niesekwencjonowany węzeł. Dlatego czas wymagany do sekwencjonowania dowolnego wywołania jest funkcją liczby wątków, a nie wielkości danych wejściowych.

Moja poprzednia odpowiedź tak naprawdę nie odpowiada poprawnie na pytanie, ale ponieważ OP uważa je za przydatne, pozostawię je bez zmian. Na podstawie kodu w linku w pytaniu, oto moja próba. Przeprowadziłem tylko bardzo podstawowe testy na tym, ale wydaje się, że poprawnie oblicza średnie. Informacje zwrotne mile widziane, czy jest to odpowiednio wolne od oczekiwania.

UWAGA : usunąłem interfejs uniwersalny i uczyniłem go klasą. Posiadanie Universal składającego się z Sekwencyjności, a także bycie jednym, wydaje się niepotrzebną komplikacją, ale może czegoś mi brakuje. W klasie średniej zaznaczyłem zmienną stanu jako volatile. Nie jest to konieczne, aby kod działał. Zachowaj ostrożność (dobry pomysł na wątki) i nie pozwól, aby każdy wątek wykonał wszystkie obliczenia (jeden raz).

Sekwencyjny i fabryczny

public interface Sequential<E, S, R>
{ 
  R apply(S priorState);

  S state();

  default boolean isApplied()
  {
    return state() != null;
  }
}

public interface Factory<E, S, R>
{
   S initial();

   Sequential<E, S, R> generate(E input);
}

uniwersalny

import java.util.concurrent.ConcurrentLinkedQueue;

public class Universal<I, S, R> 
{
  private final Factory<I, S, R> generator;
  private final ConcurrentLinkedQueue<Sequential<I, S, R>> wfq = new ConcurrentLinkedQueue<>();
  private final ThreadLocal<Sequential<I, S, R>> last = new ThreadLocal<>();

  public Universal(Factory<I, S, R> g)
  { 
    generator = g;
  }

  public R apply(I invocation)
  {
    Sequential<I, S, R> newSequential = generator.generate(invocation);
    wfq.add(newSequential);

    Sequential<I, S, R> last = null;
    S prior = generator.initial(); 

    for (Sequential<I, S, R> i : wfq) {
      if (!i.isApplied() || newSequential == i) {
        R r = i.apply(prior);

        if (i == newSequential) {
          wfq.remove(last.get());
          last.set(newSequential);

          return r;
        }
      }

      prior = i.state();
    }

    throw new IllegalStateException("Houston, we have a problem");
  }
}

Średni

public class Average implements Sequential<Integer, Average.State, Double>
{
  private final Integer invocation;
  private volatile State state;

  private Average(Integer invocation)
  {
    this.invocation = invocation;
  }

  @Override
  public Double apply(State prior)
  {
    System.out.println(Thread.currentThread() + " " + invocation + " prior " + prior);

    state = prior.add(invocation);

    return ((double) state.sum)/ state.count;
  }

  @Override
  public State state()
  {
    return state;
  }

  public static class AverageFactory implements Factory<Integer, State, Double> 
  {
    @Override
    public State initial()
    {
      return new State(0, 0);
    }

    @Override
    public Average generate(Integer i)
    {
      return new Average(i);
    }
  }

  public static class State
  {
    private final int sum;
    private final int count;

    private State(int sum, int count)
    {
      this.sum = sum;
      this.count = count;
    }

    State add(int value)
    {
      return new State(sum + value, count + 1);
    }

    @Override
    public String toString()
    {
      return sum + " / " + count;
    }
  }
}

Kod demonstracyjny

private static final int THREADS = 10;
private static final int SIZE = 50;

public static void main(String... args)
{
  Average.AverageFactory factory = new Average.AverageFactory();

  Universal<Integer, Average.State, Double> universal = new Universal<>(factory);

  for (int i = 0; i < THREADS; i++)
  {
    new Thread(new Test(i * SIZE, universal)).start();
  }
}

static class Test implements Runnable
{
  final int start;
  final Universal<Integer, Average.State, Double> universal;

  Test(int start, Universal<Integer, Average.State, Double> universal)
  {
    this.start = start;
    this.universal = universal;
  }

  @Override
  public void run()
  {
    for (int i = start; i < start + SIZE; i++)
    {
      System.out.println(Thread.currentThread() + " " + i);

      System.out.println(System.nanoTime() + " " + Thread.currentThread() + " " + i + " result " + universal.apply(i));
    }
  }
}

Wprowadziłem kilka zmian w kodzie, gdy go tutaj publikowałem. Powinno być OK, ale daj mi znać, jeśli masz z tym problemy.