Несколько потоков, способных одновременно получать стадо

У меня создалось впечатление, что flock (2) является потокобезопасным, я недавно столкнулся с ситуацией в коде, где несколько потоков могут получить блокировку в том же файле, которые все синхронизированы с использованием получения эксклюзивной блокировки с помощью c api flock. Процесс 25554 представляет собой многопоточное приложение, которое имеет 20 потоков, количество потоков, имеющих блокировку для одного и того же файла, изменяется, когда происходит тупиковая ситуация. Многопоточное приложение testEvent приложения является писателем для файла, где был нажат читатель из файла. К сожалению, lsof не печатает значение LWP, поэтому я не могу найти, какие потоки хранят блокировку. Когда происходит указанное ниже условие, процесс и потоки застревают в вызове стаи, отображаемом при вызове pstack или strace на pid 25569 и 25554. Любые предложения о том, как преодолеть это в RHEL 4.x.

Одна вещь, которую я хотел обновить, - это стая, которая не ошибается все время, когда скорость обмена сообщениями составляет более 2 Мбит/с только тогда, когда я попадаю в эту тупиковую проблему со ставкой, ниже этой скорости tx все является файлом. Я сохранил константу num_threads= 20, size_of_msg= 1000bytes и просто изменил количество сообщений tx в секунду, начиная с 10 сообщений до 100 сообщений, которое составляет 20 * 1000 * 100 = 2 mbps, когда я увеличиваю количество сообщения до 150, тогда возникает проблема стайки.

Я просто хотел спросить, каково ваше мнение о 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

Многопоточная тестовая программа, которая вызовет функцию 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;

Вышеуказанный метод будет вызываться в pthreads в основной функции теста.

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

}

Вот источник писателя/читателя, из-за соображений собственности, которые я не хотел просто вырезать и вставлять, источник записи будет обращаться к нескольким потокам процесса

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

В стороне читателя есть процесс демона без потоков.

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