Линукс 3.0:ошибка взаимоблокировки futex-lock?

Question

// SubFetch(x,y) = atomically x-=y and return x (__sync_sub_and_fetch)
// AddFetch(x,y) = atomically x+=y and return x (__sync_add_and_fetch)
// CompareWait(x, y) = futex(&x, FUTEX_WAIT, y) wait on x if x == y
// Wake(x, y) = futex(&x, FUTEX_WAKE, y) wake up y waiters

struct Lock
{
Lock() : x(1) {}

void lock()
{
    while (true)
    {
        if (SubFetch(x, 1) == 0)
            return;

        x = -1;

        CompareWait(x, -1);
    }
}

void unlock()
{
    if (AddFetch(x, 1) == 1)
        return;

    x = 1;

    Wake(x, 1);
}

private:
    int x;
};

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

Выше представлена ​​реализация блокировки (мьютекс, 1 семафор подсчета разрешений), основанная на фьютексе и описании семантики в мужской фьютекс(7)

Похоже, что он содержит ошибку взаимоблокировки, из-за которой после того, как несколько потоков пытаются заблокировать и разблокировать его несколько тысяч раз, потоки могут перейти в состояние, где x == -1, и все потоки застревают в CompareWait, однако никто не удерживает замок.

Кто-нибудь может увидеть, где ошибка?

Обновлять: Я немного удивлен, что futex(7)/семантика настолько нарушена.Я полностью переписал Lock следующим образом...это правильно сейчас?

// CompareAssign(x,y,z) atomically: if (x == y) {x = z; ret true; } else ret false;

struct Lock
{
Lock() : x(0) {}

void lock()
{
    while (!CompareAssign(x, 0, 1))
        if (x == 2 || CompareAssign(x, 1, 2))
            CompareWait(x, 2);
}

void unlock()
{
    if (SubFetch(x, 1) == 0)
        return;

    x = 0;

    Wake(x, 1);
}

private:
int x;
};

Идея здесь в том, что x имеет следующие три состояния:

0: unlocked
1: locked & no waiters
2: locked & waiters

Answer 1

Проблема в том, что вы явно присваиваете значение -1 x если SubFetch не удается установить блокировку.Это связано с разблокировкой.

1. Резьба 1 фиксируется. x==0.
2. Поток 2 пытается получить блокировку.Тот Самый SubFetch Наборы x до -1, а затем поток 2 приостанавливается.
3. Резьба 1 освобождает блокировку.Тот Самый AddFetch Наборы x равняется 0, поэтому затем код явно устанавливает x до 1 и вызывает Wake.
4. Поток 2 просыпается и устанавливает x до -1, а затем вызывает CompareWait.

Поток 2 теперь застрял в ожидании, с x установлено значение -1, но поблизости нет никого, кто мог бы его разбудить, так как поток 1 уже снял блокировку.

Answer 2

правильная реализация Mutex на основе FOTEX описана в бумаге Ульриха Дреппер «Футексики сложно»

http://people.redhat.com/drepper/futex.pdf .

Он включает в себя не только код, но и очень подробное объяснение того, почему это правильно.Код из бумаги:

class mutex
{
 public:
 mutex () : val (0) { }
 void lock () {
   int c;
   if ((c = cmpxchg (val, 0, 1)) != 0)
     do {
       if (c == 2 || cmpxchg (val, 1, 2) != 0)
         futex_wait (&val, 2);
     } while ((c = cmpxchg (val, 0, 2)) != 0);
 }
 void unlock () {
//NOTE: atomic_dec returns the value BEFORE the operation, unlike your SubFetch !
   if (atomic_dec (val) != 1) {
     val = 0;
     futex_wake (&val, 1);
   }
 }
 private:
   int val;
};

.

Сравнивая код в документе с вашим кодом, я замечаю разницу

У вас есть

Если (x== 2 || сравниватьsign (x, 1, 2))

Использование значения futex непосредственно тогда как drepper использует возвращаемое значение из предыдущего сравнения ().Эта разница, вероятно, повлияет на только производительность.

Ваш код разблокировки тоже отличается, но кажется семантически эквивалентным.

В любом случае я настоятельно рекомендую вам следить за кодом Дреппера к букве.Эта бумага выдерживала испытание временем и получила много рецензирования.Вы ничего не получаете от качения.

Answer 3

Как насчет этого сценария с тремя потоками: A, B и C?

Начальное состояние этого сценария имеет:

• нить А, удерживающая замок
• поток B пока не борется за блокировку
• резьба С внутрь CompareWait()
• x == -1 с того момента, когда C не удалось получить блокировку

        A                  B                C
    ==============   ================  ===============
     AddFetch()
     (so x == 0)
                       SubFetch()
                       (so x == -1)

     x = 1

                       x = -1

     Wake()

На данный момент независимо от того, разблокированы ли B или C, они не получат результата 0 когда они SubFetch().