múltiples subprocesos capaces de obtener rebaño al mismo tiempo

tenía la impresión de que rebaño(2) es seguro para subprocesos, recientemente me encontré con el caso en el código, donde múltiples subprocesos pueden obtener un bloqueo en el mismo archivo que están todos sincronizados con el uso de obtener un bloqueo exclusivo usando el rebaño de cap api. El proceso 25554 es una aplicación de subprocesos múltiples que tiene 20 subprocesos, la cantidad de subprocesos que tienen un bloqueo en el mismo archivo varía cuando ocurre el interbloqueo. La aplicación multiproceso evento de prueba es el escritor del archivo, donde estaba el impulso es el lector del archivo. Desafortunadamente, el lsof no imprime el valor de LWP, por lo que no puedo encontrar cuáles son los subprocesos que mantienen el bloqueo. Cuando ocurre la condición mencionada a continuación, tanto el proceso como los subprocesos se atascan en la llamada de rebaño, como lo muestra el pstack o strace llame al pid 25569 y 25554. Cualquier sugerencia sobre cómo superar esto en RHEL 4.x.

Una cosa que quería actualizar es que flock no se porta mal todo el tiempo, cuando la tasa de transmisión de los mensajes es más de 2 mbps solo entonces me meto en este problema de interbloqueo con flock, por debajo de esa tasa de transmisión todo es archivo. he guardado el num_threads = 20, size_of_msg = 1000 bytes constantes y solo varié la cantidad de mensajes tx por segundo, comenzando de 10 mensajes a 100 mensajes, que es 20 * 1000 * 100 = 2 mbps, cuando aumento la cantidad de mensajes a 150, entonces ocurre un problema de rebaño.

Solo quería preguntarle cuál es su opinión acerca de 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

El programa de prueba de subprocesos múltiples que llamará al write_data_lib_func función de liberación.

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;

El método anterior se llamará en los pthreads en la función principal de la prueba.

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

}

Aquí está la fuente del escritor/lector, debido a razones de propiedad, no quería simplemente cortar y pegar, la fuente del escritor accederá a múltiples subprocesos en un proceso

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; 
   } 

En el lado del lector de las cosas hay un proceso daemon sin subprocesos.

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 ;
      }  
   }     
}

rebaño(2) está documentado como “bloqueo si un bloqueo incompatible está en manos de otro proceso” y con “los bloqueos creados por flock() están asociados con una entrada de tabla de archivo abierto”, por lo que se debe esperar que flockLos bloqueos -ed por varios subprocesos del mismo proceso no interactúan. (los flock la documentación no menciona hilos).

Por lo tanto, la solución debería ser simple para usted: asocie uno pthread_mutex_t a cada flock-able descriptor de archivo, y proteger la llamada a flock con ese mutex. También podrías usar pthread_rwlock_t si desea un bloqueo de lectura frente a escritura.

Desde la página del manual de Linux para flock(2):

Los bloqueos creados por flock() están asociados con una entrada de tabla de archivos abiertos. Esto significa que los descriptores de archivos duplicados (creados, por ejemplo, por fork(2) o dup(2)) se refieren al mismo bloqueo, y este bloqueo puede modificarse o liberarse utilizando cualquiera de estos descriptores. Además, el bloqueo se libera mediante una operación LOCK_UN explícita en cualquiera de estos descriptores duplicados, o cuando se han cerrado todos esos descriptores.

Además, los bloqueos de bandada no se ‘apilan’, por lo que si intenta adquirir un bloqueo que ya tiene, la llamada de bandada es un noop que regresa inmediatamente sin bloquear y sin cambiar el estado de bloqueo de ninguna manera.

Dado que los subprocesos dentro de un proceso comparten descriptores de archivo, puede agrupar el archivo varias veces desde diferentes subprocesos y no se bloqueará, ya que el bloqueo ya está retenido.

También de las notas sobre flock(2):

Los bloqueos flock() y fcntl(2) tienen una semántica diferente con respecto a los procesos bifurcados y dup(2). En los sistemas que implementan flock() mediante fcntl(2), la semántica de flock() será diferente de las descritas en esta página del manual.