c++ - программирование - Как отдать приоритет привилегированному потоку в блокировке мьютекса?




многопоточное программирование c++ 11 (6)

Прежде всего: я совершенно новичок в программировании мьютекса / многопоточности, поэтому заранее прошу прощения за любые ошибки ...

У меня есть программа, которая запускает несколько потоков. Потоки (обычно по одному на ядро ​​процессора) много вычисляют и «думают», а затем иногда решают вызвать определенный (общий) метод, который обновляет некоторую статистику. Параллелизм при обновлении статистики управляется с помощью мьютекса:

stats_mutex.lock();
common_area->update_thread_stats( ... );
stats_mutex.unlock();

Теперь к проблеме. Из всех этих тем есть одна конкретная, которая нуждается почти в
приоритет реального времени, потому что это единственный поток, который на самом деле работает.

Под «приоритетом почти в реальном времени» я имею в виду:

Давайте предположим, что поток t0 является «привилегированным», а t1 .... t15 - нормальными. Что происходит сейчас:

  • Поток t1 приобретает блокировку.
  • Поток t2, t3, t0 вызывает метод lock () и ожидает его успешного выполнения.
  • Поток t1 вызывает unlock ()
  • Один (случайно, насколько мне известно) из потоков t2, t3, t0 успешно получает блокировку, а другие продолжают ждать.

Что мне нужно это:

  • Поток t1 приобретает блокировку.
  • Поток t2, t3, t0 вызывает метод lock () и ожидает его успешного выполнения.
  • Поток t1 вызывает unlock ()
  • Поток t0 получает блокировку, так как он привилегирован

Итак, каков наилучший (возможно, самый простой) способ сделать это?

Я думал о том, чтобы иметь переменную bool, называемую "privileged_needs_lock".

Но я думаю, что мне нужен другой мьютекс для управления доступом к этой переменной ... Я не знаю, является ли это правильным способом ...

Дополнительная информация:

  • мои темы используют C ++ 11 (по состоянию на gcc 4.6.3)
  • код должен работать как в Linux, так и в Windows (но в настоящее время тестируется только в Linux).
  • производительность механизма блокировки не является проблемой (моя проблема заключается в вычислениях внутренних потоков, и число потоков всегда будет низким, максимум один или два на ядро ​​процессора)

Любая идея приветствуется. Спасибо

Приведенное ниже решение работает (три мьютекса):

#include <thread>
#include <iostream>
#include "unistd.h"

std::mutex M;
std::mutex N;
std::mutex L;

void lowpriolock(){
  L.lock();
  N.lock();
  M.lock();
  N.unlock();
}

void lowpriounlock(){
  M.unlock();
  L.unlock();
}

void highpriolock(){
  N.lock();
  M.lock();
  N.unlock();
}

void highpriounlock(){
  M.unlock();
}

void hpt(const char* s){
  using namespace std;
  //cout << "hpt trying to get lock here" << endl;
  highpriolock();
  cout << s << endl;
  sleep(2);
  highpriounlock();
}

void lpt(const char* s){
  using namespace std;
  //cout << "lpt trying to get lock here" << endl;
  lowpriolock();
  cout << s << endl;
  sleep(2);
  lowpriounlock();
}

int main(){
std::thread t0(lpt,"low prio t0 working here");
std::thread t1(lpt,"low prio t1 working here");
std::thread t2(hpt,"high prio t2 working here");
std::thread t3(lpt,"low prio t3 working here");
std::thread t4(lpt,"low prio t4 working here");
std::thread t5(lpt,"low prio t5 working here");
std::thread t6(lpt,"low prio t6 working here");
std::thread t7(lpt,"low prio t7 working here");
//std::cout << "All threads created" << std::endl;
t0.join();
t1.join();
t2.join();
t3.join();
t4.join();
t5.join();
t6.join();
t7.join();
return 0;
}

Попробовал приведенное ниже решение, как предложено, но оно не работает (скомпилируйте с помощью «g ++ -std = c ++ 0x -o test test.cpp -lpthread»):

#include <thread>
#include <mutex>

#include "time.h"
#include "pthread.h"

std::mutex l;

void waiter(){
  l.lock();
  printf("Here i am, waiter starts\n");
  sleep(2);
  printf("Here i am, waiter ends\n");
  l.unlock();
}

void privileged(int id){
  usleep(200000);
  l.lock();
  usleep(200000);
  printf("Here i am, privileged (%d)\n",id);
  l.unlock();  
}

void normal(int id){
  usleep(200000);
  l.lock();
  usleep(200000);
  printf("Here i am, normal (%d)\n",id);
  l.unlock();    
}

int main(){
  std::thread tw(waiter);
  std::thread t1(normal,1);
  std::thread t0(privileged,0);
  std::thread t2(normal,2);

  sched_param sch;
  int policy; 

  pthread_getschedparam(t0.native_handle(), &policy, &sch);
  sch.sched_priority = -19;
  pthread_setschedparam(t0.native_handle(), SCHED_FIFO, &sch);

  pthread_getschedparam(t1.native_handle(), &policy, &sch);
  sch.sched_priority = 18;
  pthread_setschedparam(t1.native_handle(), SCHED_FIFO, &sch);

  pthread_getschedparam(t2.native_handle(), &policy, &sch);
  sch.sched_priority = 18;
  pthread_setschedparam(t2.native_handle(), SCHED_FIFO, &sch);

  tw.join();
  t1.join();
  t0.join();
  t2.join();

  return 0;  
}


Немного измененный ответ ecatmur , добавив 4-й мьютекс для одновременной обработки нескольких потоков с высоким приоритетом (обратите внимание, что в моем исходном вопросе это не требовалось ):

#include <thread>
#include <iostream>
#include "unistd.h"

std::mutex M; //data access mutex
std::mutex N; // 'next to access' mutex
std::mutex L; //low priority access mutex
std::mutex H; //hptwaiting int access mutex

int hptwaiting=0;

void lowpriolock(){
  L.lock();
  while(hptwaiting>0){
    N.lock();
    N.unlock();
  }
  N.lock();
  M.lock();
  N.unlock();
}

void lowpriounlock(){
  M.unlock();
  L.unlock();
}

void highpriolock(){
  H.lock();
  hptwaiting++;
  H.unlock();
  N.lock();
  M.lock();
  N.unlock();
}

void highpriounlock(){
  M.unlock();
  H.lock();
  hptwaiting--;
  H.unlock();
}

void hpt(const char* s){
  using namespace std;
  //cout << "hpt trying to get lock here" << endl;
  highpriolock();
  cout << s << endl;
  usleep(30000);
  highpriounlock();
}

void lpt(const char* s){
  using namespace std;
  //cout << "lpt trying to get lock here" << endl;
  lowpriolock();
  cout << s << endl;
  usleep(30000);
  lowpriounlock();
}

int main(){
std::thread t0(lpt,"low  prio t0  working here");
std::thread t1(lpt,"low  prio t1  working here");
std::thread t2(hpt,"high prio t2  working here");
std::thread t3(lpt,"low  prio t3  working here");
std::thread t4(lpt,"low  prio t4  working here");
std::thread t5(lpt,"low  prio t5  working here");
std::thread t6(hpt,"high prio t6  working here");
std::thread t7(lpt,"low  prio t7  working here");
std::thread t8(hpt,"high prio t8  working here");
std::thread t9(lpt,"low  prio t9  working here");
std::thread t10(lpt,"low  prio t10 working here");
std::thread t11(lpt,"low  prio t11 working here");
std::thread t12(hpt,"high prio t12 working here");
std::thread t13(lpt,"low  prio t13 working here");
//std::cout << "All threads created" << std::endl;
t0.join();
t1.join();
t2.join();
t3.join();
t4.join();
t5.join();
t6.join();
t7.join();
t8.join();
t9.join();
t10.join();
t11.join();
t12.join();
t13.join();
return 0;
}

Как вы думаете? Это нормально? Это правда, что семафор мог бы лучше обрабатывать подобные вещи, но с мьютексами мне гораздо проще управлять.


Попробуйте что-то вроде следующего. Вы можете сделать класс потокобезопасным синглтоном и даже сделать его функтором.

#include <pthread.h>
#include <semaphore.h>
#include <map>

class ThreadPrioFun
{
    typedef std::multimap<int, sem_t*> priomap_t;
public:
    ThreadPrioFun()
    {
        pthread_mutex_init(&mtx, NULL);
    }
    ~ThreadPrioFun()
    {
        pthread_mutex_destroy(&mtx);
    }
    void fun(int prio, sem_t* pSem)
    {
        pthread_mutex_lock(&mtx);
        bool bWait = !(pm.empty());
        priomap_t::iterator it = pm.insert(std::pair<int, sem_t*>(prio, pSem) );
        pthread_mutex_unlock(&mtx);

        if( bWait ) sem_wait(pSem);

        // do the actual job
        // ....
        //

        pthread_mutex_lock(&mtx);
        // done, remove yourself
        pm.erase(it);
        if( ! pm.empty() )
        {
             // let next guy run:
            sem_post((pm.begin()->second));
        }
        pthread_mutex_unlock(&mtx);
    }
private:
    pthread_mutex_t mtx;
    priomap_t pm;
};

Поскольку приоритеты потоков не работают для вас:

Создайте 2 взаимных блокировки, обычную блокировку и блокировку приоритета.

Обычные потоки должны сначала заблокировать нормальную блокировку, а затем блокировку приоритета. Приоритетный поток только должен заблокировать приоритетную блокировку:

Mutex mLock;
Mutex mPriLock;


doNormal()
{
   mLock.lock();
   pthread_yield();
   doPriority();
   mLock.unlock();
}

doPriority()
{
   mPriLock.lock();
   doStuff();
   mPriLock.unlock();
}

Я могу думать о трех методах, использующих только потоковые примитивы:

Тройной мьютекс

Три мьютекса будут работать здесь:

  • мьютекс данных ('M')
  • мьютекс следующего доступа ('N') и
  • мьютекс доступа с низким приоритетом ('L')

Шаблоны доступа:

  • Потоки с низким приоритетом: блокировка L, блокировка N, блокировка M, разблокировка N, {делать вещи}, разблокировка M, разблокировка L
  • Поток с высоким приоритетом: блокировка N, блокировка M, разблокировка N, {делать вещи}, разблокировать M

Таким образом, доступ к данным защищен, и поток с высоким приоритетом может опередить потоки с низким приоритетом в доступе к ним.

Мьютекс, условная переменная, атомный флаг

Примитивный способ сделать это с помощью условной переменной и атомарного:

  • Мутекс М;
  • Кондвар С;
  • atomic bool hpt_waiting;

Шаблоны доступа к данным:

  • Поток с низким приоритетом: блокировка M, в то время как (hpt_waiting) ожидание C на M, {do stuff}, трансляция C, разблокировка M
  • Поток с высоким приоритетом: hpt_waiting: = true, блокировка M, hpt_waiting: = false, {do stuff}, трансляция C, разблокировка M

Мьютекс, условная переменная, два неатомарных флага

В качестве альтернативы вы можете использовать два неатомных bools с condvar; в этом методе мьютекс / condvar защищает флаги, а данные защищаются не мьютексом, а флагом:

  • Мутекс М;
  • Кондвар С;
  • bool data_held, hpt_waiting;

  • Поток с низким приоритетом: блокировка M, в то время как (hpt_waiting или data_held) ожидает C на M, data_held: = true, разблокировать M, {делать вещи}, блокировать M, data_held: = false, передавать C, разблокировать M

  • Поток с высоким приоритетом: блокировка M, hpt_waiting: = true, в то время как (data_held) ожидает C на M, data_held: = true, {делать вещи}, блокировка M, data_held: = false, hpt_waiting: = false, трансляция C, разблокировка M

#include <thread>
#include <mutex>
#include <condition_variable>
#include <cassert>

class priority_mutex {
  std::condition_variable cv_;
  std::mutex gate_;
  bool locked_;
  std::thread::id pr_tid_; // priority thread
public:
  priority_mutex() : locked_(false) {}
  ~priority_mutex() { assert(!locked_); }
  priority_mutex(priority_mutex&) = delete;
  priority_mutex operator=(priority_mutex&) = delete;

  void lock(bool privileged = false) {
    const std::thread::id tid = std::this_thread::get_id();
    std::unique_lock<decltype(gate_)> lk(gate_);
    if (privileged)
      pr_tid_ = tid;
    cv_.wait(lk, [&]{
      return !locked_ && (pr_tid_ == std::thread::id() || pr_tid_ == tid);
    });
    locked_ = true;
  }

  void unlock() {
    std::lock_guard<decltype(gate_)> lk(gate_);
    if (pr_tid_ == std::this_thread::get_id())
      pr_tid_ = std::thread::id();
    locked_ = false;
    cv_.notify_all();
  }
};

ЗАМЕЧАНИЕ. Этот priority_mutex обеспечивает несправедливое планирование потоков. Если привилегированный поток часто получает блокировку, другие непривилегированные потоки могут почти не планироваться.

Пример использования:

#include <mutex>
priority_mutex mtx;

void privileged_thread()
{
  //...
  {
    mtx.lock(true);  // acquire 'priority lock'
    std::unique_lock<decltype(mtx)> lk(mtx, std::adopt_lock);
    // update shared state, etc.
  }
  //...
}

void normal_thread()
{
  //...
  {
    std::unique_lock<decltype(mtx)> lk(mtx);  // acquire 'normal lock'
    // do something
  }
  //...
}






mutex