5

我有一个分配大量内存的 c++ 类。它通过调用一个第三方库来做到这一点，如果它无法分配内存，它会崩溃，有时我的应用程序会在并行线程中创建我的类的多个实例。线程太多我崩溃了。对于解决方案，我最好的想法是确保永远不会有超过三个实例同时运行。（这是一个好主意吗？）而我目前最好的实现方法是使用boost mutex。类似于以下伪代码的内容，

MyClass::MyClass(){
  my_thread_number = -1; //this is a class variable
  while (my_thread_number == -1)
    for (int i=0; i < MAX_PROCESSES; i++)
      if(try_lock a mutex named i){
        my_thread_number = i;
        break;
      }
  //Now I know that my thread has mutex number i and it is allowed to run
}

MyClass::~MyClass(){
    release mutex named my_thread_number
}

如您所见，我不太确定互斥锁的确切语法。所以总结一下，我的问题是

  1. 当我想通过限制线程数来解决内存错误时，我是否走在正确的轨道上？
  2. 如果是，我应该使用互斥锁还是其他方式？
  3. 如果是，我的算法是否合理？
  4. 有没有一个很好的例子来说明如何将 try_lock 与 boost 互斥锁一起使用？

编辑：我意识到我在谈论线程，而不是进程。编辑：我参与构建一个可以在 linux 和 Windows 上运行的应用程序......

4

2 回答 2

5

更新我的另一个答案解决了线程之间的资源调度问题（在问题得到澄清之后）。

它显示了一种信号量方法来协调（许多）工作人员之间的工作，以及一种thread_pool首先限制工作人员并将工作排队的方法。

在 linux（也许还有其他操作系统？）上，您可以使用锁定文件习语（但某些文件系统和旧内核不支持它）。

我建议使用进程间同步对象。

例如，使用名为 semaphore 的 Boost Interprocess：

#include <boost/interprocess/sync/named_semaphore.hpp>
#include <boost/thread.hpp>
#include <cassert>

int main()
{
    using namespace boost::interprocess;
    named_semaphore sem(open_or_create, "ffed38bd-f0fc-4f79-8838-5301c328268c", 0ul);

    if (sem.try_wait())
    {
        std::cout << "Oops, second instance\n";
    }
    else
    {
        sem.post();

        // feign hard work for 30s
        boost::this_thread::sleep_for(boost::chrono::seconds(30));

        if (sem.try_wait())
        {
            sem.remove("ffed38bd-f0fc-4f79-8838-5301c328268c");
        }
    }
}

如果您在后台启动一个副本，新副本将在大约 30 秒内“拒绝”启动（“糟糕，第二个实例”）。

我觉得在这里颠倒逻辑可能更容易。嗯。让我试试。

一段时间过去了

呵呵。这比我想象的要棘手。

问题是，您要确保当您的应用程序被中断或终止时锁不会保留。为了分享便携式处理信号的技术：

#include <boost/interprocess/sync/named_semaphore.hpp>
#include <boost/thread.hpp>
#include <cassert>
#include <boost/asio.hpp>

#define MAX_PROCESS_INSTANCES 3

boost::interprocess::named_semaphore sem(
        boost::interprocess::open_or_create, 
        "4de7ddfe-2bd5-428f-b74d-080970f980be",
        MAX_PROCESS_INSTANCES);

// to handle signals:
boost::asio::io_service service;
boost::asio::signal_set sig(service);

int main()
{

    if (sem.try_wait())
    {
        sig.add(SIGINT);
        sig.add(SIGTERM);
        sig.add(SIGABRT);
        sig.async_wait([](boost::system::error_code,int sig){ 
                std::cerr << "Exiting with signal " << sig << "...\n";
                sem.post();
            });
        boost::thread sig_listener([&] { service.run(); });

        boost::this_thread::sleep_for(boost::chrono::seconds(3));

        service.post([&] { sig.cancel(); });
        sig_listener.join();
    }
    else
    {
        std::cout << "More than " << MAX_PROCESS_INSTANCES << " instances not allowed\n";
    }
}

那里有很多可以解释的。如果您有兴趣，请告诉我。

注意很明显，如果kill -9在您的应用程序中使用（强制终止），那么所有的赌注都将被取消，您必须删除 Name Semaphore 对象或显式解锁它（post()）。

这是我系统上的测试：

sehe@desktop:/tmp$ (for a in {1..6}; do ./test& done; time wait)
More than 3 instances not allowed
More than 3 instances not allowed
More than 3 instances not allowed
Exiting with signal 0...
Exiting with signal 0...
Exiting with signal 0...

real    0m3.005s
user    0m0.013s
sys 0m0.012s
于 2014-03-20T21:50:36.530 回答
2

这是实现您自己的“信号量”的一种简单方法（因为我认为标准库或 boost 没有）。这选择了一种“合作”的方法，工人将互相等待：

#include <boost/thread.hpp>
#include <boost/phoenix.hpp>

using namespace boost;
using namespace boost::phoenix::arg_names;

void the_work(int id)
{
    static int running = 0;
    std::cout << "worker " << id << " entered (" << running << " running)\n";

    static mutex mx;
    static condition_variable cv;

    // synchronize here, waiting until we can begin work
    {
        unique_lock<mutex> lk(mx);
        cv.wait(lk, phoenix::cref(running) < 3);
        running += 1;
    }

    std::cout << "worker " << id << " start work\n";
    this_thread::sleep_for(chrono::seconds(2));
    std::cout << "worker " << id << " done\n";

    // signal one other worker, if waiting
    {
        lock_guard<mutex> lk(mx);
        running -= 1;
        cv.notify_one(); 
    }
}

int main()
{
    thread_group pool;

    for (int i = 0; i < 10; ++i)
        pool.create_thread(bind(the_work, i));

    pool.join_all();
}

现在，我会说最好有一个由 n 个工作人员组成的专用池，轮流从队列中取出他们的工作：

#include <boost/thread.hpp>
#include <boost/phoenix.hpp>
#include <boost/optional.hpp>

using namespace boost;
using namespace boost::phoenix::arg_names;

class thread_pool
{
  private:
      mutex mx;
      condition_variable cv;

      typedef function<void()> job_t;
      std::deque<job_t> _queue;

      thread_group pool;

      boost::atomic_bool shutdown;
      static void worker_thread(thread_pool& q)
      {
          while (auto job = q.dequeue())
              (*job)();
      }

  public:
      thread_pool() : shutdown(false) {
          for (unsigned i = 0; i < boost::thread::hardware_concurrency(); ++i)
              pool.create_thread(bind(worker_thread, ref(*this)));
      }

      void enqueue(job_t job) 
      {
          lock_guard<mutex> lk(mx);
          _queue.push_back(std::move(job));

          cv.notify_one();
      }

      optional<job_t> dequeue() 
      {
          unique_lock<mutex> lk(mx);
          namespace phx = boost::phoenix;

          cv.wait(lk, phx::ref(shutdown) || !phx::empty(phx::ref(_queue)));

          if (_queue.empty())
              return none;

          auto job = std::move(_queue.front());
          _queue.pop_front();

          return std::move(job);
      }

      ~thread_pool()
      {
          shutdown = true;
          {
              lock_guard<mutex> lk(mx);
              cv.notify_all();
          }

          pool.join_all();
      }
};

void the_work(int id)
{
    std::cout << "worker " << id << " entered\n";

    // no more synchronization; the pool size determines max concurrency
    std::cout << "worker " << id << " start work\n";
    this_thread::sleep_for(chrono::seconds(2));
    std::cout << "worker " << id << " done\n";
}

int main()
{
    thread_pool pool; // uses 1 thread per core

    for (int i = 0; i < 10; ++i)
        pool.enqueue(bind(the_work, i));
}

PS。如果您愿意，可以在那里使用 C++11 lambdas 而不是 boost::phoenix。

于 2014-03-21T08:44:16.953 回答