多个线程能够同时获得群集

Question

我的印象是flock(2)是线程安全的,我最近在代码中遇到了这种情况,其中多个线程能够锁定同一个文件,这些文件都与使用获取独占锁使用同步c api flock.过程25554是具有20个线程的多线程应用程序,当发生死锁时,锁定到同一文件的线程数量变化.多线程应用程序testEvent是文件的编写者,其中push是文件中的reader.不幸的是,lsof它不会打印LWP值,所以我找不到哪个是持有锁的线程.当下面提到的情况发生时,进程和线程都停留在flock 调用上,如pid 25569和25554上的调用pstack或strace调用所示.有关如何在RHEL 4.x中解决此问题的任何建议.

我想要更新的一件事是flock不会一直行为不端,当消息的tx速率超过2 mbps时我才会遇到flock的死锁问题,低于tx速率一切都是文件.我保持num_threads= 20,size_of_msg= 1000字节不变,只是将每秒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_funclib函数的多线程测试程序.

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

Answer 1

flock(2)被记录为"阻塞,如果另一个进程持有不兼容的锁",并且"flock()创建的锁与打开的文件表条目相关联",所以应该预期flock多个线程的-ed锁定相同的过程不相互作用.(flock文档中没有提到线程).

因此,解决方案对您来说应该很简单:将一个pthread_mutex_t与每个flock文件描述符相关联,并flock使用该互斥锁保护调用.如果您想要读取与写入锁定,也可以使用pthread_rwlock_t.