POSIX 信号量在高争用/负载下不起作用

Question

在 Linux 内核 4.4.0-57 上使用 C++11，我试图在同一个内核上运行固定 (pthread_setaffinity_np) 的两个忙循环进程（比如 p1、p2）和使用 POSIX 信号量 (semaphore.h) 和 sched_yield() 确保交错执行顺序。但效果并不好。

下面是父代码（父任务），它产生 2 个进程，每个进程都执行子任务代码。

#include <stdio.h>                                                         
#include <cstdlib>                                                         
#include <errno.h>      // errno                                           
#include <iostream>     // cout cerr                                       
#include <semaphore.h>  // semaphore                                       
#include <fcntl.h>      // O_CREAT                                         
#include <unistd.h>     // fork                                            
#include <string.h>     // cpp string                                      
#include <sys/types.h>  //                                                 
#include <sys/wait.h>   // wait()                                          

int init_semaphore(){                                           
   std::string sname = "/SEM_CORE";                                        
   sem_t* sem = sem_open ( sname.c_str(), O_CREAT, 0644, 1 );              
   if ( sem == SEM_FAILED ) {                                              
      std::cerr << "sem_open failed!\n";                                   
      return -1;                                                           
   }                                                                       

   sem_init( sem, 0, 1 );                                                  
   return 0;                                                               
}                                                                          

// Fork and exec child-task.                                                  
// Return pid of child                                                     
int fork_and_exec( std::string pname, char* cpuid ){                       
   int pid = fork();                                                       
   if ( pid == 0) {                                                        
      // Child                                                             
      char* const params[] = { "./child-task", "99", strdup( pname.c_str() ), cpuid, NULL };                              
      execv( params[0], params );                                          
      exit(0);                                                             
   }                                                                       
   else {                                                                  
      // Parent                                                            
      return pid;                                                          
   }                                                                       
}                                                                          

int main( int argc, char* argv[] ) {                                       
   if ( argc <= 1 )                                                        
      printf( "Usage ./parent-task <cpuid> \n" );                        

   char* cpuid = argv[1];                                                  
   std::string pnames[2] = { "p111", "p222" };                             

   init_semaphore();                                                       

   int childid[ 2 ] = { 0 };                                               
   int i = 0;                                                              
   for( std::string pname : pnames ){                                      
      childid[ i ] = fork_and_exec( pname, cpuid ); 
   }                                                                       

   for ( i=0; i<2; i++ )                                                   
      if ( waitpid( childid[i], NULL, 0 ) < 0 )                            
         perror( "waitpid() failed.\n" );                                  

   return 0;                                                               
}

子任务代码如下所示：

#include <cstdlib>                                                              
#include <stdio.h>                                                              
#include <sched.h>                                                              
#include <pthread.h>                                                            
#include <stdint.h>                                                             
#include <errno.h>                                                         
#include <semaphore.h>                                                          
#include <iostream>                                                             
#include <sys/types.h>                                                          
#include <fcntl.h>      // O_CREAT                                              

sem_t* sm;                                                                      

int set_cpu_affinity( int cpuid ) {                                             
   pthread_t current_thread = pthread_self();                                   
   cpu_set_t cpuset;                                                            
   CPU_ZERO( &cpuset );                                                         
   CPU_SET( cpuid, &cpuset );                                                   
   return pthread_setaffinity_np( current_thread,                               
                                  sizeof( cpu_set_t ), &cpuset );               
}                                                                               

int lookup_semaphore() {                                                        
   sm = sem_open( "/SEM_CORE", O_RDWR );                                        
   if ( sm == SEM_FAILED ) {                                                    
      std::cerr << "sem_open failed!" << std::endl ;                            
      return -1;                                                                
   }                                                                            
}                                                                               

int main( int argc, char* argv[] ) {                                            
   printf( "Usage: ./child-task <PRIORITY> <PROCESS-NAME> <CPUID>\n" );            
   printf( "Setting SCHED_RR and priority to %d\n", atoi( argv[1] ) );          

   set_cpu_affinity( atoi( argv[3] ) );                                         

   lookup_semaphore();                                                          

   int res;                                                                     
   uint32_t n = 0;                                                              
   while ( 1 ) {                                                                
      n += 1;                                                                   
      if ( !( n % 1000 ) ) {                                                                                                                         
         res = sem_wait( sm );  

         if( res != 0 ) {                                                       
            printf(" sem_wait %s. errno: %d\n", argv[2], errno);                
         }                                                                   
         printf( "Inst:%s RR Prio %s running (n=%u)\n", argv[2], argv[1], n );  
         fflush( stdout );                                                      

         sem_post( sm );                                                        

         sched_yield();                                                         
      }                                                                         

      sched_yield();                                                            
   }                                                                            

   sem_close( sm );                                                             
}  

在子任务代码中，我有if ( !( n % 1000 ) ) 来试验减少等待和发布信号量时的争用/负载。我得到的结果是，当n % 1000 时，其中一个子进程将始终处于睡眠状态（来自top），而另一个子进程正常执行。但是，如果我设置n % 10000，即减少负载/争用，两个进程将运行并交错打印输出，这是我的预期结果。

有谁知道这是 semaphore.h 的限制还是有更好的方法来确保进程执行顺序？

Answer 1

更新：我用线程和信号量做了一个简单的例子，注意 sched_yield 可能有助于避免不必要的线程唤醒，这些线程不是“依次”工作，但屈服不是保证。我还展示了一个使用互斥锁/condvar 的示例，该示例保证可以正常工作，无需让步。

#include <stdexcept>
#include <semaphore.h>
#include <pthread.h>
#include <thread>
#include <iostream>

using std::thread;
using std::cout;

sem_t sem;
int count = 0;

const int NR_WORK_ITEMS = 10;

void do_work(int worker_id)
{
    cout << "Worker " << worker_id << '\n';
}

void foo(int work_on_odd)
{
    int result;
    int contention_count = 0;
    while (count < NR_WORK_ITEMS)
    {
        result = sem_wait(&sem);
        if (result) {
            throw std::runtime_error("sem_wait failed!");
        }
        if (count % 2 == work_on_odd)
        {
            do_work(work_on_odd);
            count++;
        }
        else
        {
            contention_count++;
        }
        result = sem_post(&sem);
        if (result) {
            throw std::runtime_error("sem_post failed!");
        }
        result = sched_yield();
        if (result < 0) {
            throw std::runtime_error("yield failed!");
        }
    }
    cout << "Worker " << work_on_odd << " terminating. Nr of redundant wakeups from sem_wait: " <<
        contention_count << '\n';
}

int main()
{
    int result = sem_init(&sem, 0, 1);

    if (result) {
        throw std::runtime_error("sem_init failed!");
    }

    thread t0 = thread(foo, 0);
    thread t1 = thread(foo, 1);

    t0.join();
    t1.join();

    return 0;
}

这是使用条件变量和互斥锁的一种方法。从 C++ std 线程转换为 pthread 应该是微不足道的。要在进程之间执行此操作，您必须使用可以在进程之间共享的 pthread 互斥锁类型。也许 condvar 和 mutex 都可以放在共享内存中，以实现我在下面使用线程所做的相同事情。

另见手册页 pthread_condattr_setpshared (3) 或 http://manpages.ubuntu.com/manpages/wily/man3/pthread_condattr_setpshared.3posix.html

另一方面，在两个工作进程之间使用 SOCK_STREAM unix 域套接字可能更简单，然后阻塞套接字，直到对等工作人员通过套接字对您执行 ping 操作（即发送一个字符）。

#include <cassert>
#include <iostream>
#include <thread>
#include <condition_variable>
#include <unistd.h>

using std::thread;
using std::condition_variable;
using std::mutex;
using std::unique_lock;
using std::cout;

condition_variable cv;
mutex mtx;
int count;

void dowork(int arg)
{
    std::thread::id this_id = std::this_thread::get_id();

    cout << "Arg: " << arg << ", thread id: " << this_id << '\n';
}

void tfunc(int work_on_odd)
{
    assert(work_on_odd < 2);

    auto check_can_work = [&count, &work_on_odd](){ return ((count % 2) ==
                                                            work_on_odd); };
    while (count < 10)
    {
        unique_lock<mutex> lk(mtx);
        cv.wait (lk, check_can_work);
        dowork(work_on_odd);
        count++;
        cv.notify_one();
        // Lock is unlocked automatically here, but with threads and condvars,                                                                                                                                                                                
        // it is actually better to unlock manually before notify_one.                                                                                                                                                                                        
    }
}

int main()
{
    count = 0;
    thread t0 = thread(tfunc, 0);
    thread t1 = thread(tfunc, 1);
    sleep(1);
    cv.notify_one();

    t0.join();
    t1.join();
}