如果A（“整体”）有一组B（“部分”），如何将“部分”B从“整体”​​A转移到另一个？

Question

假设我们有整体对象关系。例如，有一个Boss 可以拥有多个Workers。 对于Boss，我们需要以下操作：

• 将新的Worker 添加到其“集合”中
• 将现有的Worker 对象转移到另一个Boss 对象

我在下面提供了一些 C++ 代码来说明这个想法（跳过了一些函数）。

class Worker {
private:
    Boss *owner = nullptr;
    std::string name;

public:
    Worker(std::string name) : name(name), owner(nullptr) { }

    Boss* getOwner() const {
        return this->owner;
    }

    void setOwner(Boss *owner) {
        this->owner = owner;
    }
};

class Boss {
private:
    std::set<Worker *> workers;
    std::string name;

public:
    Boss(std::string name) : name(name) { }

    ~Boss() {
        for (auto worker : this->workers) {
            delete worker;
        }
        this->workers.clear();
    }

    void addWorker(Worker *worker) {
        this->workers.emplace(worker);
        worker->setOwner(this);
    }

    void transferOwnership(Worker *dummyWorker, Boss *newOwner);
};

int main()
{
    Boss b1{ "boss1" };
    Boss b2{ "boss2" };

    b1.addWorker(new Worker{ "w1" });
    b1.addWorker(new Worker{ "w2" });

    b2.addWorker(new Worker{ "w3" });

    Worker temp{ "w2" };
    b1.transferOwnership(&temp, &b2);

    b1.print(); // boss: boss1 having workers: w1
    b2.print(); // boss: boss2 having workers: w2 w3

    return 0;
}

transferOwnership() 函数的串行实现如下：

void Boss::transferOwnership(Worker *dummyWorker, Boss *newOwner) {
    // find existing worker
    Worker *actualWorker = nullptr;
    for (auto worker : this->workers) {
        if (*worker == *dummyWorker) {
            actualWorker = worker;
        }
    }

    if (nullptr == actualWorker) {
        return;
    }

    // remove existing worker from this
    this->workers.erase(actualWorker);

    // add worker to other (newOwner)
    newOwner->workers.emplace(actualWorker);
    actualWorker->setOwner(newOwner);
}

现在，我们想要并行化这个场景。假设有多个线程可以对任何现有的Boss 对象执行任何可用操作（addWorker() 或transferOwnership()）。

问题是如何同步代码，这样我们最后总能得到一致的结果，而不会陷入死锁。

我认为一个解决方案是在Boss 类中添加一个静态互斥变量，在操作开始时将其锁定并在操作结束时结束。这将保护所有 Boss 对象的所有 workers 集。

另一种解决方案是为每个Boss 对象设置一个互斥锁。但是，我认为在这里我们可能会遇到死锁，就像在 transferOwnership() 中一样，我们需要锁定/解锁 Bosses 的互斥锁，我认为在这种情况下我们不能强加特定的锁定顺序。

你能想到其他/更好的解决方案吗？

Answer 1

静态互斥解决方案有效。但是，如果您确实希望每个对象都有一个互斥锁，那么您的事务需要锁定 2 个老板和 1 个工人（根据您的要求，您可能只需要锁定 2 个老板对象）。

为了避免这种情况下的死锁，只需实现一个函数，根据一些众所周知的预定义顺序锁定 3 个互斥锁。它们在内存中的位置可能用于此目的。

你的函数看起来像：

1. 根据内存地址对 2 或 3 个互斥体进行排序
2. 按照该顺序锁定它们（确保不会两次获得相同的互斥锁）

如果您使用的是 C++11，则可以使用 std::lock(std::mutex... mutexes) 来做或多或少相同的事情。

void lock(std::mutex& m1, std::mutex& m2) {
  if (&m1==&m2)
    m1.lock();
  else if (&m1<&m2) {
    m1.lock();
    m2.lock();
  } else {
    m2.lock();
    m1.lock();
  }
}

解锁以相反的顺序完成。