POSIXセマフォが高競合/負荷の下で動作しない

Question

Linuxカーネル4.4.0-57でC++ 11を使用して、2つのビジーループプロセス（例：p1、p2）を固定しようとしています（pthread_setaffinity_np ）を同じコアに置き、POSIXセマフォ（セマフォー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の場合、子プロセスの1つが常にスリープ状態（からまで）になり、他の子プロセスが正しく実行されるということです。しかし、n % 10000を設定すると負荷/競合が少なくなり、両方のプロセスが実行され、インターリーブ出力が出力されますが、これは私の予想される結果です。

これがsemaphore.hの制限であるかどうかは誰にも分かりませんが、プロセスの実行順序を確実にする良い方法がありますか？

Answer 1

私はスレッドとセマフォを使って簡単な例を示しましたが、sched_yieldは作業のために '順番に'ではないスレッドの不必要な起床を避けるのに役立つことがあることに注意してください。私はmutex/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; 
} 

条件変数とmutexでこれを行う方法の1つです。 C++のstdスレッドからpthreadへの変換は簡単です。プロセス間で行うには、プロセス間で共有できるpthread mutex型を使用する必要があります。おそらく、condvarとmutexの両方を共用メモリに置いて、スレッドで以下のようにすることができます。

も参照してくださいpthread_condattr_setpshared manページ（3）または一方 http://manpages.ubuntu.com/manpages/wily/man3/pthread_condattr_setpshared.3posix.html

が、多分ちょうど2つのワーカー・プロセス間のSOCK_STREAM UNIXドメインソケットを使用して、ちょうどまで、ソケット上でブロックすることが簡単ですピアツーピアはあなたにソケットを介してpingを送信します（つまり、1つのcharを送信します）。

#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(); 
}