将 std::atomic<std::shared_ptr<T>> 与非平凡对象正确使用？

Question

我正在尝试通过std::atomic<std::shared_ptr>> 实现无锁包装器，以对容器等重要对象进行操作。 我在这两个主题中找到了一些相关的信息：

• memory fence
• atomic usage

但这仍然不是我需要的。

举个例子：

TEST_METHOD(FechAdd)
    {
        constexpr size_t loopCount = 5000000;
        auto&& container           = std::atomic<size_t>(0);
        auto thread1               = std::jthread([&]()
              {
                  for (size_t i = 0; i < loopCount; i++)
                      container++;
              });

        auto thread2 = std::jthread([&]()
            {
                for (size_t i = 0; i < loopCount; i++)
                    container++;
            });
        thread1.join();
        thread2.join();
        Assert::AreEqual(loopCount * 2, container.load());
    }

这个函数可以正常工作，因为后自增运算符使用了内部fetch_add() 原子操作。

另一方面：

TEST_METHOD(LoadStore)
{
    constexpr size_t loopCount = 5000000;
    auto&& container           = std::atomic<size_t>(0);
    auto thread1               = std::jthread([&]()
          {
              for (size_t i = 0; i < loopCount; i++)
              {
                  auto value = container.load();
                  value++;
                  container.store(value);
              }
          });

    auto thread2 = std::jthread([&]()
        {
            for (size_t i = 0; i < loopCount; i++)
            {
                auto value = container.load();
                value++;
                container.store(value);
            }
        });
    thread1.join();
    thread2.join();
    Assert::AreEqual(loopCount * 2, container.load());
}

而如果我将其替换为 .load() 和 .store() 操作并在这两个操作之间递增，结果就不一样了。 那是两个原子操作，因此这些操作之间无法进行同步。

我的最终目标是通过std::atomic<std::shared_ptr> 加载对象的实际状态，执行一些非常量操作，并通过存储操作再次保存。

TEST_METHOD(AtomicSharedPtr)
{
    constexpr size_t loopCount = 5000000;
    auto&& container           = std::atomic(std::make_shared<std::unordered_set<int>>());
    auto thread1               = std::jthread([&]([[maybe_unused]] std::stop_token token)
          {
              for (size_t i = 0; i < loopCount; i++)
              {
                  // some other lock-free synchronization primitives as barrier, conditions or?
                  auto reader = container.load();
                  reader->emplace(5);
                  container.store(reader);
              }
          });

    auto thread2 = std::jthread([&]([[maybe_unused]] std::stop_token token)
        {
            for (size_t i = 0; i < loopCount; i++)
            {
                // some other lock-free synchronization primitives as barrier, conditions or?
                auto reader = container.load();
                reader->erase(5);
                container.store(reader);
            }
        });
}

我知道第二个线程也只有 shared_ptr 来自对 shared_ptr 的原子和非常量操作，which can only cause data race.

那么关于如何实现一个无锁包装器的任何提示，该包装器将与存储在std::atomic<std::shared_ptr> 中的对象的非常量操作一起工作？

Answer 1

首先，附注。 std::atomic<std::shared_ptr<T>> 提供对指针 的原子访问，并为T 提供任何同步。在这里要注意这一点非常重要。而且您的代码显示您正在尝试同步T，而不是指针，因此atomic 没有按照您的想法进行。要使用std::atomic<std::shared_ptr<T>>，必须将指向的T 视为const。

有两种方法可以以线程安全的方式处理任意数据的读-修改-写。第一个显然是使用锁。这通常执行速度更快，并且由于其简单性，通常错误较少，因此强烈建议使用。如果你真的想用原子操作来做这件事，那会很困难，而且执行起来会更慢。

它通常看起来像这样，在其中制作指向数据的深层副本，对副本进行变异，然后尝试用新数据替换旧数据。如果其他人在此期间更改了数据，则将其全部丢弃并重新开始整个突变。

template<class T, class F>
bool readModifyWrite(std::atomic<std::shared_ptr<T>>& container, F&& function) {
   do {
       const auto&& oldT = container.load();
       //first a deep copy, to enforce immutability
       auto&& newT = std::make_shared(oldT.get());
       //then mutate the T
       if (!function(*newT))
           return false; //function aborted
       //then attempt to save the modified T.
       //if someone else changed the container during our modification, start over
  } while(container.compare_exchange_strong(oldT, newT) == false);
    //Note that this may take MANY tries to eventually succeed.
    return true;
}

然后用法和你的差不多：

auto&& container           = std::atomic(std::make_shared<std::unordered_set<int>>());
auto thread1               = std::jthread([&]([[maybe_unused]] std::stop_token token)
      {
          for (size_t i = 0; i < loopCount; i++)
          {
              readModifyWrite(container, [](auto& reader) {
                 reader.emplace(5);
                 return true;
              });
          }
      });

auto thread2 = std::jthread([&]([[maybe_unused]] std::stop_token token)
    {
        for (size_t i = 0; i < loopCount; i++)
        {
              readModifyWrite(container, [](auto& reader) {
                 reader.erase(5);
                 return true;
              });
        }
    });
}

请注意，由于一个线程正在插入5loopCount 次，另一个正在擦除5 loopCount 次，但它们之间并不同步，因此第一个线程可能会连续写入多次（这是一个集合的无操作），然后第二个线程可能会连续擦除几次（这是一个集合的无操作），所以你真的不能保证这里的最终结果，但是我假设你知道这一点。

但是，如果您想使用突变进行同步，那将变得相当复杂。如果变异函数成功或中止，则必须返回，然后readModifyWrite 的调用者必须处理修改中止的情况。 （注意readModifyWrite实际上是从函数返回值，所以它从修改步骤返回值。写入步骤不影响返回值）

auto thread1               = std::jthread([&]([[maybe_unused]] std::stop_token token)
      {
          for (size_t i = 0; i < loopCount; )
          {
              bool did_emplace = readModifyWrite(container, [](auto& reader) {
                 return reader.emplace(5);
              });
              if (did_emplace) i++;
          }
      });

auto thread2 = std::jthread([&]([[maybe_unused]] std::stop_token token)
    {
        for (size_t i = 0; i < loopCount; )
        {
              bool did_erase = readModifyWrite(container, [](auto& reader) {
                 return reader.erase(5);
              });
              if (did_erase) i++;
        }
    });
}