Pytanie wywiadu dotyczące synchronizacji wielowątkowej: Znajdź n słów o m wątkach

Czy istnieje sposób, aby ten problem mógł skorzystać z rozwiązania z wieloma wątkami, a nie z jednym wątkiem?

W wywiadzie poproszono mnie o rozwiązanie problemu przy użyciu wielu wątków. Wydaje mi się, że wiele wątków nie przynosi żadnych korzyści.

Oto problem:

Otrzymujesz akapit, który zawiera n liczbę słów, otrzymujesz m wątków. Co musisz zrobić, każdy wątek powinien wydrukować jedno słowo i dać kontrolę nad następnym wątkiem, w ten sposób każdy wątek będzie drukował jedno słowo, w przypadku, gdy nadejdzie ostatni wątek, powinien wywołać pierwszy wątek. Drukowanie będzie powtarzane, aż wszystkie słowa zostaną wydrukowane w akapicie. Wreszcie wszystkie wątki powinny wyjść z gracją. Jakiego rodzaju synchronizacji użyjesz?

Mocno czuję, że nie możemy tutaj wykorzystać nici, ale wierzę, że osoba przeprowadzająca wywiad próbuje zmierzyć moje umiejętności synchronizacji. Czy w tym problemie brakuje czegoś, co sprawiłoby, że wiele wątków miałoby wartość?

Nie potrzebujesz kodu, po prostu zastanów się. Zrealizuję sam.

Wydaje mi się, że prowadzą cię do rozwiązania semaforów. Semafory służą do sygnalizowania kolejnemu wątkowi, że jest ich kolej. Są one używane znacznie rzadziej niż muteksy, i myślę, że dlatego uważają, że to dobre pytanie do rozmowy kwalifikacyjnej. Właśnie dlatego przykład wydaje się wymyślony.

Zasadniczo mtworzyłbyś semafory. Każdy wątek xczeka na semafor, xa następnie x+1wykonuje posty na semaforze po wykonaniu swojej czynności. W pseudokodzie:

loop:
    wait(semaphore[x])
    if no more words:
        post(semaphore[(x+1) % m])
        exit
    print word
    increment current word pointer
    post(semaphore[(x+1) % m])

Moim zdaniem jest to wspaniałe pytanie na rozmowę kwalifikacyjną - przynajmniej przy założeniu, że (1) kandydat ma głęboką wiedzę na temat wątków, oraz (2) osoba przeprowadzająca wywiad ma również głęboką wiedzę i używa pytania do zbadania kandydata. Zawsze możliwe jest, że ankieter szukał konkretnej, wąskiej odpowiedzi, ale kompetentny ankieter powinien szukać:

• Umiejętność odróżnienia pojęć abstrakcyjnych od konkretnego wdrożenia. Wrzucam ten przede wszystkim jako meta-komentarz do niektórych komentarzy. Nie, przetwarzanie jednej listy słów w ten sposób nie ma sensu. Ważna jest jednak abstrakcyjna koncepcja szeregu operacji, które mogą obejmować wiele maszyn o różnych możliwościach.
• Z mojego doświadczenia (prawie 30 lat aplikacji rozproszonych, wieloprocesowych i wielowątkowych), dystrybucja pracy nie jest trudna. Zbieranie wyników i koordynowanie niezależnych procesów to miejsca, w których występuje najwięcej błędów w wątkach (znowu, z mojego doświadczenia). Poprzez sprowadzenie problemu do prostego łańcucha, ankieter może zobaczyć, jak dobrze kandydat myśli o koordynacji. Dodatkowo, osoba przeprowadzająca wywiad ma możliwość zadawania wszelkiego rodzaju dalszych pytań, takich jak „OK, co jeśli każdy wątek musi wysłać swoje słowo do innego wątku w celu odtworzenia”.
• Czy kandydat myśli o tym, jak model pamięci procesora może wpłynąć na implementację? Jeśli wyniki jednej operacji nigdy nie zostaną usunięte z pamięci podręcznej L1, jest to błąd, nawet jeśli nie ma widocznej współbieżności.
• Czy kandydat oddziela wątki od logiki aplikacji?

Ten ostatni punkt jest moim zdaniem najważniejszy. Ponownie, w oparciu o moje doświadczenie, debugowanie kodu wątkowego staje się wykładniczo trudniejsze, jeśli wątki są mieszane z logiką aplikacji (wystarczy spojrzeć na wszystkie pytania Swing dotyczące SO). Uważam, że najlepszy kod wielowątkowy jest pisany jako samodzielny kod jednowątkowy, z wyraźnie określonymi przełączeniami.

Mając to na uwadze, moim podejściem byłoby nadanie każdemu wątkowi dwóch kolejek: jednej dla danych wejściowych, drugiej dla danych wyjściowych. Wątek blokuje się podczas odczytywania kolejki wejściowej, usuwa pierwsze słowo z ciągu i przekazuje pozostałą część ciągu do swojej kolejki wyjściowej. Niektóre funkcje tego podejścia:

• Kod aplikacji odpowiada za odczytywanie kolejki, robienie czegoś z danymi i pisanie kolejki. Nie ma znaczenia, czy jest ona wielowątkowa, czy nie, czy kolejka jest kolejką w pamięci na jednym komputerze, czy kolejką opartą na TCP między komputerami, które znajdują się po przeciwnych stronach świata.
• Ponieważ kod aplikacji jest napisany jakby jednowątkowy, można go testować w sposób deterministyczny, bez potrzeby dużej ilości rusztowań.
• Podczas fazy wykonywania kod aplikacji jest właścicielem przetwarzanego ciągu. Nie musi dbać o synchronizację z równolegle wykonującymi się wątkami.

Mimo to kompetentny ankieter może nadal badać wiele szarych obszarów:

• „OK, ale chcemy poznać Twoją wiedzę na temat prymitywów współbieżności; czy możesz zaimplementować kolejkę blokującą?” Pierwszą odpowiedzią powinno być oczywiście użycie gotowej kolejki blokującej z wybranej platformy. Jeśli jednak rozumiesz wątki, możesz utworzyć implementację kolejki w kilkunastu liniach kodu, używając dowolnych operacji podstawowych synchronizacji obsługiwanych przez platformę.
• „Co jeśli jeden krok w tym procesie zajmuje bardzo dużo czasu?” Powinieneś pomyśleć o tym, czy chcesz mieć ograniczoną, czy nieograniczoną kolejkę wyjściową, jak możesz obsługiwać błędy i wpływ na ogólną przepustowość, jeśli masz opóźnienie.
• Jak skutecznie kolejkować ciąg źródłowy. Niekoniecznie problem, jeśli masz do czynienia z kolejkami w pamięci, ale może to być problem, jeśli poruszasz się między komputerami. Możesz również eksplorować opakowania tylko do odczytu na podstawowej niezmiennej tablicy bajtów.

Wreszcie, jeśli masz doświadczenie w programowaniu współbieżnym, możesz porozmawiać o niektórych frameworkach (np. Akka for Java / Scala), które już stosują ten model.

Pytania do wywiadu są czasami podstępnymi pytaniami, mającymi na celu skłonienie Cię do zastanowienia się nad problemem, który próbujesz rozwiązać. Zadawanie pytań dotyczących pytania jest integralną częścią podejścia do każdego problemu, czy to w prawdziwym świecie, czy w wywiadzie. W Internecie krąży wiele filmów wideo na temat tego, jak podchodzić do pytań w wywiadach technicznych (w szczególności Google i Microsoft).

„Po prostu spróbuj odpowiedzieć i wynoś się stamtąd…”

Zbliżanie się do wywiadów z tym schematem myślenia doprowadzi cię do zbombardowania każdego wywiadu dla każdej firmy, w której warto pracować.

Jeśli nie uważasz, że dużo zyskujesz (jeśli cokolwiek z wątków), powiedz im to. Powiedz im, dlaczego według ciebie nie ma żadnych korzyści. Porozmawiaj z nimi. Wywiady techniczne mają być otwartą platformą do dyskusji. Możesz w końcu dowiedzieć się czegoś o tym, jak to może być przydatne. Nie posuwaj się naprzód na ślepo, próbując wdrożyć coś, o czym kazał ci twój ankieter.

• Najpierw tokenizuj akapit za pomocą odpowiednich ograniczników i dodaj słowa do kolejki.

• Utwórz N liczby wątków i przechowuj je w puli wątków.

• Iteruj po puli wątków, uruchom wątek i poczekaj na
  dołączenie wątku. I rozpocznij następny wątek, gdy pierwszy wątek się skończy i tak dalej.

• Każdy wątek powinien po prostu odpytać kolejkę i wydrukować ją.

• Gdy wszystkie wątki zostaną użyte w puli wątków, zacznij od początku puli.

Jak powiedziałeś, nie sądzę, aby ten scenariusz przyniósł wiele korzyści, jeśli w ogóle, dzięki wątkom. Najprawdopodobniej jest wolniejszy niż implementacja z jednym wątkiem.

Jednak moją odpowiedzią byłoby, aby każdy wątek w ciasnej pętli próbował uzyskać dostęp do blokady, która kontroluje dostęp do indeksu tablicy słów. Każdy wątek chwyta blokadę, pobiera indeks, pobiera odpowiednie słowo z tablicy, drukuje je, zwiększa indeks, a następnie zwalnia blokadę. Wątki wychodzą, gdy indeks znajduje się na końcu tablicy.

Coś takiego:

while(true)
{
    lock(index)
    {
        if(index >= array.length())
          break;
        Console.WriteLine(array[index]);
        index++;
    }
}

Myślę, że powinno to osiągnąć jeden wątek po drugim wymaganiu, ale kolejność wątków nie jest gwarantowana. Ciekawi mnie też inne rozwiązania.

Użyj interfejsów API warunku oczekiwania / sygnału, aby rozwiązać ten problem.

Powiedzmy, że pierwszy wątek wybiera 1 słowo, a tymczasem wszystkie wątki czekają na sygnał. Wydrukuj pierwszy wątek 1. słowo i wygeneruje sygnał do następnego wątku, a następnie wydrukuj drugi wątek drugie słowo i wygeneruje sygnał do 3. wątku i tak dalej.

#include <iostream>
#include <fstream>
#include <pthread.h>
#include <signal.h>
pthread_cond_t cond[5] = {PTHREAD_COND_INITIALIZER,};
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

using namespace std;

string gstr;

void* thread1(void*)
{
    do {
    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&cond[0],&mutex);
    cout <<"thread1 :"<<gstr<<endl;
    pthread_mutex_unlock(&mutex);
    }while(1);
}

void* thread2(void*)
{
    do{
    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&cond[1],&mutex);
    cout <<"thread2 :"<<gstr<<endl;
    pthread_mutex_unlock(&mutex);
    }while(1);
}

void* thread3(void*)
{
    do{
    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&cond[2],&mutex);
    cout <<"thread3 :"<<gstr<<endl;
    pthread_mutex_unlock(&mutex);
    }while(1);
}

void* thread4(void*)
{
    do{
    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&cond[3],&mutex);
    cout <<"thread4 :"<<gstr<<endl;
    pthread_mutex_unlock(&mutex);
    }while(1);
}

void* thread5(void*)
{
    do{
    pthread_mutex_lock(&mutex);
    pthread_cond_wait(&cond[4],&mutex);
    cout <<"thread5 :"<<gstr<<endl;
    pthread_mutex_unlock(&mutex);
    }while(1);
}

int main()
{
    pthread_t t[5];
    void* (*fun[5])(void*);
    fun[0]=thread1;
    fun[1]=thread2;
    fun[2]=thread3;
    fun[3]=thread4;
    fun[4]=thread5;

    for (int i =0 ; i < 5; ++i)
    {
        pthread_create(&t[i],NULL,fun[i],NULL);
    }
    ifstream in;
    in.open("paragraph.txt");
    int i=0;
    while(in >> gstr)
    {

        pthread_cond_signal(&cond[i++]);
        if(i == 5)
            i=0;
        usleep(10);
    }
    for (int i =0 ; i < 5; ++i)
    {
        int ret = pthread_cancel(t[i]);
        if(ret != 0)
            perror("pthread_cancel:");
        else
            cout <<"canceled\n";
    }
    pthread_exit(NULL);
}

[Użyte tutaj terminy mogą być specyficzne dla wątków POSIX]

Aby rozwiązać ten problem, powinno być również możliwe zastosowanie muteksu FIFO.

Gdzie używać:

Załóżmy, że dwa wątki T1 i T2 próbują wykonać sekcję krytyczną. Obaj nie mają wiele do zrobienia poza tym krytycznym punktem i trzymają zamki przez długi czas. Tak więc T1 może zablokować, wykonać i odblokować oraz zasygnalizować T2 do wybudzenia. Ale zanim T2 może się obudzić i uzyskać blokadę, T1 ponownie blokuje i wykonuje. W ten sposób T2 może czekać bardzo długo, zanim faktycznie otrzyma blokadę lub może nie być.

Jak to działa / Jak wdrożyć:

Miej muteks do zablokowania. Zainicjuj dane specyficzne dla wątku (TSD) dla każdego wątku w węźle zawierającym identyfikator wątku i semafor. Ponadto mają dwie zmienne - własność (PRAWDA lub FAŁSZ lub -1), właściciel (identyfikator wątku właściciela). Ponadto utrzymuj kolejkę kelnerów i wskaźnik kelnera Ostatni wskazujący na ostatni węzeł w kolejce kelnerów.

działanie blokady:

node = get_thread_specific_data(node_key);
lock(mutex);
    if(!owned)
    {
        owned = true;
        owner = self;
        return success;
    }

    node->next = nullptr;
    if(waiters_queue == null) waiters_queue = node;
    else waiters_last->next = node;

    waiters_last = node;
unlock(mutex);
sem_wait(node->semaphore);

lock(mutex);
    if(owned != -1) abort();
    owned = true;
    owner = self;
    waiters_queue = waiters_queue->next;
 unlock(mutex);

operacja odblokowania:

lock(mutex);
    owner = null;
    if(waiters_queue == null)
    {
        owned = false;
        return success;
    }
    owned = -1;
    sem_post(waiters_queue->semaphore);
unlock(mutex);

Interesujące pytanie. Oto moje rozwiązanie w Javie wykorzystujące SynchronousQueue do utworzenia kanału spotkania między wątkami:

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Stack;
import java.util.concurrent.SynchronousQueue;

public class FindNWordsGivenMThreads {

    private static final int NUMBER_OF_WORDS = 100;
    private static final int NUMBER_OF_THREADS = 5;
    private static final Stack<String> POISON_PILL = new Stack<String>();

    public static void main(String[] args) throws Exception {
        new FindNWordsGivenMThreads().run();
    }

    private void run() throws Exception {
        final Stack<String> words = loadWords();
        SynchronousQueue<Stack<String>> init = new SynchronousQueue<Stack<String>>();
        createProcessors(init);
        init.put(words);
    }

    private void createProcessors(SynchronousQueue<Stack<String>> init) {
        List<Processor> processors = new ArrayList<Processor>();

        for (int i = 0; i < NUMBER_OF_THREADS; i++) {

            SynchronousQueue in;
            SynchronousQueue out;

            if (i == 0) {
                in = init;
            } else {
                in = processors.get(i - 1).getOut();
            }

            if (i == (NUMBER_OF_THREADS - 1)) {
                out = init;
            } else {
                out = new SynchronousQueue();
            }

            Processor processor = new Processor("Thread-" + i, in, out);
            processors.add(processor);
            processor.start();

        }

    }

    class Processor extends Thread {

        private SynchronousQueue<Stack<String>> in;
        private SynchronousQueue<Stack<String>> out;

        Processor(String name, SynchronousQueue in, SynchronousQueue out) {
            super(name);
            this.in = in;
            this.out = out;
        }

        @Override
        public void run() {

            while (true) {

                Stack<String> stack = null;
                try {
                    stack = in.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                if (stack.empty() || stack == POISON_PILL) {
                    System.out.println(Thread.currentThread().getName() + " Done!");
                    out.offer(POISON_PILL);
                    break;
                }

                System.out.println(Thread.currentThread().getName() + " " + stack.pop());
                out.offer(stack);
            }
        }

        public SynchronousQueue getOut() {
            return out;
        }
    }

    private Stack<String> loadWords() throws Exception {

        Stack<String> words = new Stack<String>();

        BufferedReader reader = new BufferedReader(new FileReader(new File("/usr/share/dict/words")));
        String line;
        while ((line = reader.readLine()) != null) {
            words.push(line);
            if (words.size() == NUMBER_OF_WORDS) {
                break;
            }
        }
        return words;
    }
}

Powiedziałbym, że na tego rodzaju pytanie bardzo trudno jest odpowiedzieć, ponieważ wymaga najlepszego sposobu zrobienia czegoś całkowicie głupiego. Mój mózg po prostu nie działa w ten sposób. Nie może znaleźć rozwiązania głupich pytań. Mój mózg od razu powiedziałby, że w tych warunkach używanie wielu wątków jest bezcelowe, więc użyłbym jednego wątku.

Następnie poprosiłbym ich, aby zadali pytanie dotyczące wątków w prawdziwym świecie lub pozwolili mi podać przykład poważnego wątku w prawdziwym świecie.