C中的引用计数

Question

尝试使用引用计数在纯 C 中的线程之间传递结构。我有 pthreads 和 gcc atomics 可用。我可以让它工作，但我正在寻找防弹。

一开始，我使用了结构本身拥有的 pthread 互斥锁：

struct item {
  int ref;
  pthread_mutex_t mutex;
};

void ref(struct item *item) {
  pthread_mutex_lock(&item->mutex);
  item->ref++;
  pthread_mutex_unlock(&item->mutex);
}

void unref(struct item *item) {
  pthread_mutex_lock(&item->mutex);
  item->ref--;
  pthread_mutex_unlock(&item->mutex);
  if (item->ref <= 0)
    free(item);
}

struct item *alloc_item(void) {
  struct item *item = calloc(1, sizeof(*item));
  return item;
}

但是，意识到互斥锁不应该归项目所有：

static pthread_mutex_t mutex;
struct item {
  int ref;
};

void ref(struct item *item) {
  pthread_mutex_lock(&mutex);
  item->ref++;
  pthread_mutex_unlock(&mutex);
}

void unref(struct item *item) {
  pthread_mutex_lock(&mutex);
  item->ref--;
  if (item->ref <= 0)
    free(item);
  pthread_mutex_unlock(&mutex);
}

struct item *alloc_item(void) {
  struct item *item = calloc(1, sizeof(*item));
  return item;
}

然后，进一步实现的指针是按值传递的，所以我现在有了：

static pthread_mutex_t mutex;
struct item {
  int ref;
};

void ref(struct item **item) {
  pthread_mutex_lock(&mutex);
  if (item != NULL) {
    if (*item != NULL) {
      (*item)->ref++;
    }
  }
  pthread_mutex_unlock(&mutex);
}

void unref(struct item **item) {
  pthread_mutex_lock(&mutex);
  if (item != NULL) {
    if (*item != NULL) {
      (*item)->ref--;
      if ((*item)->ref == 0) {
        free((*item));
        *item = NULL;
      }
    }
  }
  pthread_mutex_unlock(&mutex);
}

struct item *alloc_item(void) {
  struct item *item = calloc(1, sizeof(*item));
  if (item != NULL)
    item->ref = 1;
  return item;
}

这里有什么逻辑错误吗？谢谢！

Answer 1

我不知道通用解决方案。

如果能够将其减少为引用计数的原子加/减，那就太好了。事实上，大多数时候这就是所有需要的......所以逐步通过互斥锁或任何伤害。

但真正的问题是同时管理引用计数和指向项目的指针。

当一个线程来到ref()一个项目时，它是如何找到它的？如果它不存在，大概它必须创建它。如果它已经存在，它必须避免其他线程在引用计数增加之前释放它。

所以...您的void ref(struct item** item) 工作基于互斥锁保护struct item** 指针...当您持有互斥锁时，没有其他线程可以更改指针 - 所以只有一个线程可以创建项目（并将计数递增 0->1），并且只有一个线程可以销毁该项目（在递减计数 1->0 之后）。

据说，计算机科学中的许多问题都可以通过引入新的间接级别来解决，这就是这里发生的事情。问题是所有线程如何获得该项目的地址 - 考虑到它可能（轻轻地和突然地）消失了？答案：发明一个间接级别。

但是，现在我们假设指向该项目的指针本身不会消失。如果指向项目的指针可以保持一个进程全局（静态存储持续时间），这可以很容易地实现。如果指向该项目的指针是分配的存储持续时间对象的（部分），那么我们必须确保这个更高级别的对象以某种方式被锁定 - 以便指向该项目的指针的地址在使用时是“稳定的” .也就是说，更高级别的对象不会在内存中移动，也不会在我们使用它时被销毁！

因此，锁定互斥锁后的检查if (item == NULL) 是可疑的。如果互斥锁还保护指向该项的指针，则该互斥锁需要在建立指向该项的指针的地址之前被锁定——在这种情况下，在锁定之后检查为时已晚。或者指向该项目的指针的地址以其他方式受到保护（可能由另一个互斥锁） - 在这种情况下，可以在锁定之前完成检查（并将其移动到那里可以清楚地了解互斥锁保护的内容，以及它确实不保护）。

但是，如果项目是较大数据结构的一部分，并且该结构已锁定，则您可能（很好）根本不需要锁定来覆盖指向该项目的指针。这取决于...正如我所说，我不知道一般的解决方案。

我有一些大型的动态数据结构（哈希表、队列、树等），它们由多个线程共享。大多数情况下，线程会查找并保留项目一段时间。当系统繁忙时，它非常繁忙，并且可以将项目的销毁推迟到事情变得更安静时。所以我在大型结构上使用读/写锁，对引用计数使用原子加/减，并使用垃圾收集器对项目进行实际销毁。这里的要点是，引用计数的（显然简单且自包含的）递增/递减机制的选择取决于如何管理项目的创建和销毁，以及线程如何拥有指向一个项目（毕竟这是引用计数的计数）。

---

如果您手头有 128 位原子操作，您可以将 64 位地址和 64 位引用计数放在一起，然后执行以下操作：

ref:   bar = fetch_add(*foo, 1) ;
       ptr = bar >> 64 ;
       if (ptr == NULL)
         {
            if (bar & 0xF...F)
              ...create item etc.
            else
              ...wait for item
         } ;

unref: bar = fetch_sub(*foo, 1) ;
       if ((bar & 0xF...F) == 0)
         {
            if (cmp_xchg(*foo, bar, (NULL << 64) | 0)))
              ...free(bar >> 64) ;
         } ;

其中foo 是 128 位组合的 ptr/ref-count（其存在受到某些外部手段的保护）——假设 64 位 ptr 和 64 位计数——而bar 是一个局部变量那种形式，ptr 是一个 void*。

如果找到指针NULL触发项目创建，那么第一个将计数从0->1移动的线程知道他们是谁，并且在项目创建之前到达的任何线程，以及指针集合，也知道是谁他们是并且可以等待。设置指针需要cmp_xchg()，然后创建者会发现有多少线程在等待。

这种机制将引用计数从项目中移出，并将其与项目的地址捆绑在一起，这看起来很简洁——尽管您现在在操作时需要项目的地址，以及引用的地址当您对项目的引用计数进行操作时，对项目进行操作。

这将替换您的 ref 和 unref 函数中的互斥锁...但不能解决引用本身如何受到保护的问题。