Возможное состояние гонки / неопределенное поведение в многопоточной программе

Question

Я написал многопоточную тестовую программу для тестирования моего сервера БД, но похоже, что я столкнулся с каким-то сценарием состояния гонки / неопределенного поведения.

Моя программа использует 'n' количество потоков для ввода 'x' количество записей (IMSI) в базе данных. В потоках я выбираю значение IMSI (для ввода в БД), а затем вызываю API, который вставляет IMSI в БД. Хотя я не получаю никаких ошибок в API «Вставка», тем не менее, не все IMSI вставляются в БД!

Вот программа:

 #include"DB.hpp"
 #include<thread>
 #include<vector>
 #include<string>
 #include<mutex>        
 #include<algorithm>
 #include<iostream>
 using namespace std;

 std::mutex mtx_imsi;
 std::mutex mtx_errorCount;
 std::mutex mtx_addImsi;

 class data
 {
    public:
    static int64_t imsi; //This is stored in the DB
    static int64_t no_of_threads;
    static int64_t no_of_subscribers; //No. of Imsis that will be stored.
    static int64_t error_count; //No. of IMSIs which couldn't be written.
    static vector<string> imsi_list;

    static void get_imsi(int64_t &l_imsi)
    {
        std::lock_guard<mutex> lg(mtx_imsi);

        if(imsi==405862999999999+no_of_subscribers)
           l_imsi=-1;           
        else
           l_imsi=++imsi;
    }

    static void add_count(int64_t l_count)
    {
       std::lock_guard<mutex> lg(mtx_errorCount);
       error_count+=l_count;
    }

    static void add_imsi(vector<string>& list)
    {
       std::lock_guard<mutex> lg(mtx_addImsi);
       for(const auto& x:list)
           imsi_list.push_back(x);
    }
};

int64_t data::imsi(405862999999999); //This is the initial value
int64_t data::no_of_threads;
int64_t data::no_of_subscribers;
int64_t data::error_count=0;
vector<string> data::imsi_list;

int main(int argc, char* argv[])
{
    if(argc!=3)
    {
        cout<<endl<<"Error in input parameters"<<endl;
        cout<<endl<<argv[0]<<"[No_of_threads]   [No_of_subscribers]  [NODE_IP]"<<endl;
        cout<<"e.g. "<<argv[0]<<"10 200000 10.32.129.66"<<endl;
        exit(-1);
    }

    data::no_of_threads=stoi(argv[1]);
    data::no_of_subscribers=stoi(argv[2]);

    DB::InitDBConnection(argv[3]);   //This will initialise the DB connection with the IP

    vector<thread> t;

    for(int i=0;i<data::no_of_threads;i++)
    {
        thread th([&]{
            int64_t errorCount=0,temp_imsi; 
            vector<string> temp_list;

            data::get_imsi(temp_imsi);

            while(temp_imsi!=-1)
            {
                string l_imsi=to_string(temp_imsi);                                             
                temp_list.push_back(l_imsi);

                ReturnCode status=DB::rtInsertImsi(l_imsi);

                if(status!=INSERT_OK)
                    ++errorCount;

                data::get_imsi(temp_imsi);
            }

            data::add_count(errorCount);
            data::add_imsi(temp_list);

        });

        t.push_back(move(th));        
    }

    for(auto &x:t)
    x.join();

    std::sort (data::imsi_list.begin(), data::imsi_list.end());
    cout<<endl<<"IMSI LIST"<<endl;

    // Printing the IMSIs which were entered.
    for(const auto&x:data::imsi_list)
        cout<<x<<endl;

    cout<<endl<<"Number of Imsis used: "<<data::imsi-405862999999999;
    cout<<endl<<"Number of errors: "<<data::error_count;

    return 0;
}

В данный момент я считаю, что в моей функции «Вставка» (которую я вызываю внутри потока) нет ничего плохого, потому что она используется в других многопоточных программах без такого поведения. Что может быть причиной того, что некоторые IMSI не были вставлены? Что-то не так с этой основной программой?

---

Во время публикации этого вопроса я изменил реальный код, чтобы сделать код более понятным (я не знал, что буду удалять строки, содержащие ошибку). Теперь я осознал свою ошибку. В моем реальном коде я не передаю Imsi, полученный из get_imsi (), в мою функцию вставки (которая была бы поточно-ориентированной), вместо этого я использую полученное значение, чтобы заполнить структуру данных и передать эту структуру данных вставке функция. Поскольку я не заполнял структуру данных потокобезопасным способом, я получил замечания, о которых упомянул.

Я хочу удалить вопрос, но из-за запущенной награды я больше не могу это делать!

---

Answer 1

Я не вижу, какой компилятор вы используете, однако с Clang и GCC на Linux и Mac у вас есть возможность активировать дезинфицирующие средства.

Глядя на комментарии, я рекомендую создавать разные варианты сборок, чтобы проверять разные виды проблем. -fsanitize=thread активирует дезинфицирующее средство для нитей. Если вы тестируете свою программу, она должна информировать вас об условиях гонки на самом низком уровне. Он даже сообщает о них, когда они на самом деле не срабатывают. Действительно полезно!

При срабатывании выдает стеки вызовов для доступа и запуска потока. Как и переменная, это примерно.

Простое решение этой проблемы состоит в том, чтобы сделать переменную атомарной, однако, имейте в виду, что это не решает логическую проблему. Так что внимательно посмотрите на то, что на самом деле происходит, и решите большую проблему в конце.

Другие дезинфицирующие средства включают проверку на неопределенное поведение, неправильный доступ к памяти ...

Answer 2

Эта программа, как написано, имеет много проблем. Но гоночные условия и другие подобные неприятности нити не входят в их число

Публичные статические переменные класса - очень плохая идея. Они в основном просто глобальные переменные. И большинство из них не меняются после инициализации. Они должны быть const переменными-членами.

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

Вот как это могло бы быть написано лучше. Я верно повторил большую часть того, что ты сделал. Я не знаю достаточно о том, что ты делаешь, чтобы сделать что-то лучше:

//#include "DB.hpp"
#include <thread>
#include <vector>
#include <string>
#include <mutex>
#include <algorithm>
#include <iostream>
#include <atomic>

using namespace std;

class data
{
 public:
   data(int64_t no_of_threads, int64_t no_of_subscribers, int64_t starting_imsi)
        : no_of_threads_(no_of_threads), no_of_subscribers_(no_of_subscribers),
          starting_imsi_(starting_imsi),
          error_count_(0),
          cur_imsi_(0)
   {
      cur_imsi_ = starting_imsi;
   }

   int64_t next_imsi() {
      if ((cur_imsi_ - starting_imsi_) >= no_of_subscribers_) {
         return -1;
      } else {
         return ++cur_imsi_;
      }
   }

   void add_errors(int64_t l_count)
   {
      lock_guard<mutex> lg(mtx_error_count_);
      error_count_ += l_count;
   }

   void add_imsi_list(vector<string> const & list)
   {
      lock_guard<mutex> lg(mtx_imsi_list_);
      imsi_list_.insert(imsi_list_.end(), list.begin(), list.end());
   }

   void sort_imsi_list()
   {
      // Probably not necessary, but to be thorough.
      lock_guard<mutex> lg(mtx_imsi_list_);
      sort(imsi_list_.begin(), imsi_list_.end());
   }

   int64_t imsis_used() const { return cur_imsi_ - starting_imsi_; }

   int64_t error_count() const {
      lock_guard<mutex> lg(mtx_error_count_);
      return error_count_;
   }

   int64_t thread_count() const { return no_of_threads_; }

   vector<string> const &get_imsi_list() const { return imsi_list_; }

 private:
   const int64_t no_of_threads_;
   const int64_t no_of_subscribers_;
   const int64_t starting_imsi_;
   atomic<int64_t> cur_imsi_;

   mutable mutex mtx_error_count_; // Never const
   int64_t error_count_; //No. of IMSIs which couldn't be written.
   mutable mutex mtx_imsi_list_; // Never const
   vector<string> imsi_list_;
};


int main(int argc, char* argv[])
{
   if (argc != 3)
   {
      cout << endl << "Error in input parameters" << endl;
      cout << endl << argv[0]
           << "[No_of_threads]   [No_of_subscribers]  [NODE_IP]" << endl;
      cout << "e.g. " << argv[0] << "10 200000 10.32.129.66" << endl;
      return 1;
   }

   data imsi_generator(stoi(argv[1]), stoi(argv[2]), 405862999999999);

   // DB::InitDBConnection(argv[3]);   //This will initialise the DB connection with the IP

   vector<thread> t;

   for(int i=0;i<imsi_generator.thread_count();i++)
   {
      t.emplace_back([&imsi_generator]
                     {
                        int64_t errorCount = 0, temp_imsi;
                        vector<string> temp_list;

                        temp_imsi = imsi_generator.next_imsi();

                        while (temp_imsi != -1)
                        {
                           string const l_imsi = to_string(temp_imsi);
                           temp_list.push_back(l_imsi);

                           // ReturnCode status = DB::rtInsertImsi(l_imsi);
                           //
                           // if (status != INSERT_OK)
                           //    ++errorCount;

                           temp_imsi = imsi_generator.next_imsi();
                        }

                        imsi_generator.add_errors(errorCount);
                        imsi_generator.add_imsi_list(temp_list);
                     });
   }

   for (auto &x : t)
      x.join();

   imsi_generator.sort_imsi_list();
   cout << endl << "IMSI LIST" << endl;

   // Printing the IMSIs which were entered.
   for (auto const &x : imsi_generator.get_imsi_list())
      cout << x << endl;

   cout << endl << "Number of Imsis used: " << imsi_generator.imsis_used();
   cout << endl << "Number of errors: " << imsi_generator.error_count();

   return 0;
}