wiele wątków mogących uzyskać stado w tym samym czasie

Question

Miałem wrażenie, że flock(2) jest bezpieczny dla wątków, ostatnio przebiegłem przez skrzynkę w kodzie, gdzie wiele wątków jest w stanie uzyskać blokadę tego samego pliku, które są wszystko zsynchronizowane z wykorzystaniem uzyskania wyłącznego zamka za pomocą stada cipi. Proces 25554 jest aplikacją wielowątkową, która ma 20 wątków, liczba wątków mających blokadę do tego samego pliku zmienia się w przypadku wystąpienia impasu. Aplikacja z wieloma gwintami jest zapisywana do pliku, gdzie było naciśnięcie czytnika z pliku. Niestety, lsof nie drukuje wartości LWP, więc nie mogę znaleźć, które są gwinty, które trzymają blokadę. Gdy wystąpi poniższy warunek, zarówno proces, jak i wątki zostaną zablokowane w wywołaniu flock, jak to jest wyświetlane przez wywołanie w pidzie 25569 i 25554. Wszelkie sugestie dotyczące tego rozwiązania w RHEL 4.x.

Jedną z rzeczy, którą chciałem zaktualizować, jest to, że flok nie działa przez cały czas, kiedy wskaźnik tx wiadomości wynosi więcej niż 2 Mb/s tylko wtedy dostaję się do tego problemu zakleszczenia z flokiem, poniżej tego współczynnika tx wszystko jest plikiem. Utrzymałem stałą liczbę num_threads = 20, size_of_msg = 1000bajtów i zmieniłem liczbę wiadomości na sekundę od 10 wiadomości na 100, czyli 20 * 1000 * 100 = 2 Mb/s, gdy zwiększam liczbę wiadomości do 150, problem stada.

Chciałam zapytać, jaka jest Twoja opinia na temat pliku flockfile c api.

sudo lsof filename.txt 
    COMMAND  PID  USER  FD  TYPE  DEVICE  SIZE NODE  NAME 
    push   25569 root  11u  REG  253.4  1079 49266853 filename.txt 
    testEvent 25554 root  27uW  REG  253.4  1079 49266853 filename.txt 
    testEvent 25554 root  28uW  REG  253.4  1079 49266853 filename.txt 
    testEvent 25554 root  29uW  REG  253.4  1079 49266853 filename.txt 
    testEvent 25554 root  30uW  REG  253.4  1079 49266853 filename.txt 

Wielowątkowość program testowy, który wywoła funkcję write_data_lib_func lib.

void* sendMessage(void *arg) { 

int* numOfMessagesPerSecond = (int*) arg; 
std::cout <<" Executing p thread id " << pthread_self() << std::endl; 
while(!terminateTest) { 
    Record *er1 = Record::create(); 
    er1.setDate("some data"); 

    for(int i = 0 ; i <=*numOfMessagesPerSecond ; i++){ 
    ec = _write_data_lib_func(*er1); 
    if(ec != SUCCESS) { 
     std::cout << "write was not successful" << std::endl; 

    } 

    } 
    delete er1; 
    sleep(1); 
} 

return NULL; 

Powyższa metoda zostanie wywołana w pthreads w głównej funkcji testu.

for (i=0; i<_numThreads ; ++i) { 
    rc = pthread_create(&threads[i], NULL, sendMessage, (void *)&_num_msgs); 
    assert(0 == rc); 

}

Oto źródło pisarz/czytelnik, z powodu własnych powodów nie chcą po prostu wyciąć i wkleić, źródło pisarz będzie dostępne z wielu wątków w procesie

int write_data_lib_func(Record * rec) {  
if(fd == -1) { 
    fd = open(fn,O_RDWR| O_CREAT | O_APPEND, 0666); 
} 
if (fd >= 0) { 
    /* some code */ 

    if(flock(fd, LOCK_EX) < 0) { 
    print "some error message"; 
    } 
    else { 
    if(maxfilesize) { 
     off_t len = lseek (fd,0,SEEK_END); 
     ... 
     ... 
     ftruncate(fd,0); 
     ... 
     lseek(fd,0,SEEK_SET); 
    } /* end of max spool size */ 
    if(writev(fd,rec) < 0) { 
    print "some error message" ; 
    } 

    if(flock(fd,LOCK_UN) < 0) { 
    print some error message; 
    } 

Po stronie czytelnika jest proces demona bez wątków.

int readData() { 
    while(true) { 
     if(fd == -1) { 
     fd= open (filename,O_RDWR); 
     } 
     if(flock (fd, LOCK_EX) < 0) { 
     print "some error message"; 
     break; 
     } 
     if(n = read(fd,readBuf,readBufSize)) < 0) { 
     print "some error message" ; 
     break; 
     } 
     if(off < n) { 
     if (off <= 0 && n > 0) { 
      corrupt_file = true; 
     } 
     if (lseek(fd, off-n, SEEK_CUR) < 0) { 
      print "some error message"; 
     } 
     if(corrupt_spool) { 
      if (ftruncate(fd,0) < 0) { 
      print "some error message"; 
      break; 
      } 
     } 
     } 
     if(flock(fd, LOCK_UN) < 0) 
     print some error message ; 
     } 
    }  
} 

Answer 1

flock(2) jest opisany jako „blokowanie jeśli niezgodne blokada jest utrzymywana przez inny proces ” i „blokuje utworzone przez stado() są związane z wpisu tablicy otwarty plik”, więc należy spodziewać że flock -ed blokuje przez kilka wątków tego samego procesu nie współdziałają. (flock dokumentacja nie wspomina nici).

Stąd rozwiązanie powinno być proste dla Ciebie: skojarzyć jeden pthread_mutex_t do każdego flock -able deskryptor pliku i chronić wezwanie do flock z tym mutex. Możesz również użyć pthread_rwlock_t, jeśli chcesz blokować czytanie lub zapisywanie.

Answer 2

Na stronie Linux man dla trzody (2):

Blokady utworzone za pomocą flock() są skojarzone z otwartym wejściem tabeli plików. Oznacza to, że zduplikowane deskryptory plików (utworzone na przykład przez fork (2) lub dup (2)) odnoszą się do tej samej blokady, a ta blokada może zostać zmodyfikowana lub zwolniona przy użyciu dowolnego z tych deskryptorów. Ponadto blokada jest uwalniana albo przez jawną operację LOCK_UN na dowolnym z tych zduplikowanych deskryptorów , albo gdy wszystkie takie deskryptory zostały zamknięte.

Ponadto stada zamki nie „stos”, więc w przypadku próby uzyskania blokady już trzymać, rozmowa stado jest noop że wraca natychmiast bez blokowania i bez zmiany stanu blokady w jakikolwiek sposób .

Od wątków wewnątrz deskryptorów plików współużytkowania procesu, można flokować plik wiele razy z różnych wątków, i nie będzie blokować, ponieważ blokada jest już wstrzymana.

również z informacji dodatkowej trzody (2):

hodowlanej (fcntl) i (2) mają różne zamki semantykę względem rozgałęzionych procesów i DUP (2). W systemach, w których funkcja flock() jest używana pod numerem fcntl (2), semantyka flock() będzie inna od tej opisanej na tej stronie podręcznika: .