Linux 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 предоставляет системный вызов под названием futex , на котором основаны многие утилиты параллелизма, включая недавние реализации pthread_mutex.Всякий раз, когда вы пишете код, вы всегда должны учитывать, является ли использование существующей реализации или написание ее самостоятельно лучшим выбором для вашего проекта.

Выше приведена реализация блокировки (мьютекс, семафор разрешений 1) на основе futex иописание семантики в man futex (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 на основе futex описана в статье Ульриха Дреппера «Futexes is tricky»

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;
};

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

У вас есть

if (x == 2 || CompareAssign (x, 1, 2))

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

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

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

Answer 3

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

Исходное состояние этого сценария имеет:

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

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

     x = 1

                       x = -1

     Wake()

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