过滤器锁定算法

Question

我正忙着研究n线程互斥的过滤器锁定算法，我似乎无法理解代码的第17行。我知道它是在某种条件下旋转的，但不完全确定这些条件是什么。更具体地说，(∃k != me) 意味着什么。

1 class Filter implements Lock {
2 int[] level;
3 int[] victim;
4 public Filter(int n) {
5     level = new int[n];
6     victim = new int[n]; // use 1..n-1
7     for (int i = 0; i < n; i++) {
8         level[i] = 0;
9     }
10 }
11 public void lock() {
12     int me = ThreadID.get();
13     for (int i = 1; i < n; i++) { //attempt level 1
14     level[me] = i;
15     victim[i] = me;
16     // spin while conflicts exist
17     while ((∃k != me) (level[k] >= i && victim[i] == me)) {};
18     }
19 }
20 public void unlock() {
21     int me = ThreadID.get();
22     level[me] = 0;
23 }
24 }

Answer 1

我的阅读

(∃k != me) (level[k] >= i && victim[i] == me)

是“除了

k

之外还存在一些

me

，使得

level[k] >= i && victim[i] == me

”。

循环不断旋转，直到没有

k

条件成立。

这是陈述同一件事的另一种方式：

boolean conflicts_exist = true;
while (conflicts_exist) {
   conflicts_exist = false;
   for (int k = 1; k < n; k++) {
      if (k != me && level[k] >= i && victim[i] == me) {
         conflicts_exist = true;
         break;
      }
   }
}

Answer 2

可以写成：

for (int k = 0; k < n; k++) {
      while ((k != me) && (level[k] >= i && victim[i] == me)) {
           //spin wait
      }
}

Answer 3

我可以向你展示我的实现。 过滤器类：

package org.example.ch2.example4;

import org.example.ch2.Lock;
import org.example.ch2.ThreadID;

public class Filter implements Lock {
    private volatile int[] level;
    private volatile int[] victim;

    Filter(int n) {
        level = new int[n];
        victim = new int[n];
        for (int i = 0; i < n; i++) {
            level[i] = 0;
        }
    }

    @Override
    public void lock() {
        int me = ThreadID.get();
        int n = level.length;

        for(int i = 1; i < n; i++) {
            level[me] = i; // I'm interested
            victim[i] = me; // you go first

            for(int k = 0; k < n; k++) { // check other thread ids
                if(k == me)
                    continue; // skip myself

                // spin while conflict exists
                while(level[k] >= i && victim[i] == me) {} // wait
            }
        }

    }

    @Override
    public void unlock() {
        int me = ThreadID.get();

        level[me] = 0;
    }
}

锁定界面：

package org.example.ch2;

public interface Lock {
    public void lock();
    public void unlock();
}

ThreadID 类：

package org.example.ch2;

import java.util.HashMap;

public class ThreadID {
    private static final HashMap<Long, Integer> map = new HashMap<>();

    public static synchronized int get() {
        long threadId = Thread.currentThread().threadId();
        if(!map.containsKey(threadId)) {
            int key = map.size();
            map.put(threadId, key);
        }

        return map.get(threadId);
    }
}

运行示例：

package org.example.ch2.example4;

import org.example.ch2.Lock;

public class Example {
    private static int count = 0;

    public static void main(String[] args) {
        int n = 9;
        Lock lock = new Filter(n);

        Thread[] threads = new Thread[n];

        for (int i = 0; i < n; i++) {
            threads[i] = new Thread(() -> {
                lock.lock();
                System.out.println(++count);
                lock.unlock();
            });
        }

        for (Thread thread : threads) {
            thread.start();
        }

        // pause before you press Enter
        new java.util.Scanner(System.in).nextLine();
    }
}

输出：

1
2
3
4
5
6
7
8
9