间隔锁实现

Question

我正在寻找间隔锁的实现。给定一个区间(x, y)，如果没有其他人正在获取包含点p 的任何区间，其中x <= p <= y，则线程可以获取锁。

我目前的想法是维护一组现有的授予间隔(x1, y1, x2, y2, ..., xn, yn)，其中x1 < y1 < x2 < y2 < ... < xn < yn 并检查(x, y) 是否与这些间隔中的任何一个重叠。

搜索需要O(logn) 时间，这让我很高兴。但是，当搜索返回有一些重叠时，lock 函数需要以某种方式有效地重试，直到当其他人释放他们的间隔锁时它可以获取锁。忙于等待或睡觉似乎不是一个好主意。

有没有办法有效地实现重试？

Answer 1

我认为问题本质上是关于让线程等待并重试的有效方法。
听听变化怎么样

现有授权间隔数组

仅在更改时重试？
以下不应该被认为是正确的实现（我对线程的经验非常有限），而是对建议机制的演示：

Ranges.java 和 Range.java

//represents all ranges
//see also: https://stackoverflow.com/a/7721388/3992939
public class Ranges {

    private List<Range> ranges = new ArrayList<>();
    private PropertyChangeSupport rangeChangedProperty = new PropertyChangeSupport(this);

    public Range getRange(int rangeStart, int rangeEnd) {

        if(contains(rangeStart) || contains(rangeEnd)) {
            return null;
        }
        Range range = new Range(rangeStart, rangeEnd);
        range.addListener( (observable, oldValue, newValue) -> {
                rangeChangedProperty.firePropertyChange("Range", "-" , "changed");
            }
        );
        ranges.add(range);
        return range;
    }

    private boolean contains(int number){
        for(Range range : ranges) {
            if(range.contains(number)) {return true;}
        }
        return false;
    }

    public boolean removeRange(Range range) {

        boolean isContains = ranges.remove(range);
        rangeChangedProperty.firePropertyChange("Range", "-" , "removed");
        return isContains;
    }

    /**
     * Listen to {@link #rangeChangedProperty}. Fires whenever a range changes
     * or removed.
     * <br/>A client and a listener and when it fires, notify all threads.
     */
    public void addChangeListener(PropertyChangeListener listener) {
        rangeChangedProperty.addPropertyChangeListener(listener);
    }

    //represents a single range
    //It is muttable 
    //can be implemented using ValueRange (https://stackoverflow.com/a/40716042/3992939)
    class Range{

        private SimpleIntegerProperty low = new SimpleIntegerProperty();
        private SimpleIntegerProperty high = new SimpleIntegerProperty();
        private SimpleObjectProperty<int[]> rangeProperty = new SimpleObjectProperty<>();

        private Range(int rangeStart, int rangeEnd){

            low.set(rangeStart) ; high.set(rangeEnd);
            updateRange();
            low.addListener((observable, oldValue, newValue) -> { updateRange(); });
            high.addListener((observable, oldValue, newValue) -> { updateRange(); });
        }

        /**
         * Listen to {@link #rangeProperty} that changes whenever the range changes
         */
        void addListener(ChangeListener<int[]> listener) {
            rangeProperty.addListener(listener);
        }

        private void updateRange() {rangeProperty.set(new int[] {low.get(), high.get()});}

        public int getRangeStart() { return low.get(); }

        public void setRangeStart(int rangeStart) { low.set(rangeStart);}

        public int getRangeEnd() {return high.get();}

        public void setRangeEnd(int rangeEnd) { high.set(rangeEnd);}

        public boolean contains(int number){
            int min = Math.min(low.get(), high.get());
            int max = Math.max(low.get(), high.get());
            return ((number >= min) && (number <= max));
        }
    }
}

GetRange.java

//used to simulate a thread trying to get a range 
public class GetRange implements Runnable{

    private Ranges ranges;
    private int low, high;
    private String id;

    GetRange(Ranges ranges, int low, int high, String id) {
        this.ranges = ranges;
        this.low = low; this.high = high; this.id = id;
    }

    @Override
    public void run() {
        synchronized (ranges) {
            while(ranges.getRange(low,high) == null) {
                System.out.println("Tread "+ id + " is waiting");
                try {
                    ranges.wait();
                } catch (InterruptedException ex) { ex.printStackTrace();}
            }
        }
        System.out.println("Tread "+ id + " got range. All done");
    }
}

测试是：

//test
public static void main(String[] args) throws InterruptedException {
    Ranges ranges = new Ranges();
    ranges.addChangeListener( (evt) -> {
        synchronized (ranges) {
            ranges.notifyAll();
            System.out.println(evt.getPropertyName() + " "+ evt.getNewValue());
        }
    });

    Range range1 = ranges.getRange(10,15);
    Range range2 = ranges.getRange(20,25);

    new Thread(new GetRange(ranges, 10, 12, "A")).start();
    new Thread(new GetRange(ranges, 21, 28, "B")).start();
    new Thread(new GetRange(ranges, 10, 12, "C")).start();

    Thread.sleep(50);
    System.out.println("-- Changing end of range 1. Threads notifyied and keep waiting -----");
    range1.setRangeEnd(16);   //no thread effected
    Thread.sleep(50);
    System.out.println("-- Changing start of range 1. Threads notifyied and A or C get range -----");
    range1.setRangeStart(13); //effects thread A or C
    Thread.sleep(50);
    System.out.println("-- Removing range 2. Threads notifyied and B get range -----");
    ranges.removeRange(range2);//effects thread B
    Thread.sleep(50);
    System.exit(1);
}

输出：

胎面 A 正在等待
胎面 C 正在等待
胎面 B 正在等待
-- 更改范围的结束 1. 通知线程并保持等待 -----
范围已更改
胎面 B 正在等待
胎面 C 正在等待
胎面 A 正在等待
-- 更改范围的开始 1. 线程通知和 A 或 C 获取范围 -----
范围更改 Tread A 获取范围。全部完成
线程 C 正在等待
胎面 B 正在等待
-- 删除范围 2. 线程通知和 B 获取范围 -----
范围已删除
胎面 B 有范围。全部完成
胎面 C 正在等待

Answer 2

Guava 的Striped locks 可能会引起您的兴趣。

如果您有一个函数int key(int p)，它返回[x_i,y_i] 所属的区间i 的索引p，您可以使用Striped 锁来实现您的目标。

例如，如果我们将点x_1、x_2、...x_n 作为区间边界，那么x_i < x_(i+1) 和x_(i+1) - x_i 在从1 到@ 的所有i 上保持不变987654335@，我们可以使用key(p) = p -> (p - x_1) / n之类的东西。

但是，根据您选择的符号，此假设可能不成立，并且函数 key 并不那么简单 - 但希望锁定条带化解决方案对您有用。

Answer 3

正如@c0der 建议的那样，我做了一个简单地跟踪锁定间隔的实现。

我的代码暗示了一个Range 类...

• 是不可变的
• 有一个下限和上限（扩展到无限范围应该不会太难）
• 正确实现equals() 和hashCode()

RangeLock 类目前只实现了阻塞锁方法。解锁是通过返回的Unlocker 实例完成的。这是为了避免没有获得锁的线程能够解锁给定的Range。

public class RangeLock<T extends Comparable<? super T>> {

    private final SortedSet<Range<T>> locked = new TreeSet<>(Comparator.comparing(Range::lower));
    private final Object lock = new Object();

    public Unlocker lock(Range<T> range) throws InterruptedException {
        synchronized (lock) {
            while (!available(range)) {
                lock.wait();
            }
            locked.add(range);
            return () -> {
                synchronized (lock) {
                    locked.remove(range);
                    lock.notifyAll();
                }
            };
        }
    }

    private boolean available(Range<T> range) {
        SortedSet<Range<T>> tailSet = locked.tailSet(range);
        SortedSet<Range<T>> headSet = locked.headSet(range);
        return (tailSet.isEmpty() || !tailSet.first().overlaps(range)) && (headSet.isEmpty() || !headSet.last().overlaps(range));
    }

    public interface Unlocker {
        void unlock();
    }
}

Answer 4

这是我对支持读写锁的 IntervalLock 的实现。读取可能会获取范围重叠的锁，而如果其范围与任何其他读取或写入重叠，则写入必须等待。基本思想是使用interval tree 来存储范围。在给定时间，每个范围可能持有一个写锁或多个读锁。必须小心地从树中插入和删除范围，以防止出现任何竞争条件。该代码使用了来自here 的区间树实现。

SemaphoreInterval.java

package intervallock;

import java.util.ArrayList;
import java.util.concurrent.Semaphore;

import datastructures.Interval;

public class SemaphoreInterval implements Interval {
    private ArrayList<Semaphore> semaphores;
    private int start;
    private int end;
    private int mode;

    public SemaphoreInterval(int start, int end, int mode) {
        this.semaphores = new ArrayList<>(1);
        this.start = start;
        this.end = end;
        this.mode = mode;
    }

    public int getMode() {
        return mode;
    }

    public ArrayList<Semaphore> getSemaphores() {
        return semaphores;
    }

    @Override
    public int start() {
        return start;
    }

    @Override
    public int end() {
        return end+1;
    }
}

IntervalLock.java

package intervallock;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Semaphore;

import datastructures.IntervalTree;

/**
 * An implementation of Interval Lock
 * 
 * @author Hieu
 *
 */
public class IntervalLock {
    public IntervalTree<SemaphoreInterval> tree;
    private Semaphore treeLock;
    private int maxPermits;

    public static final int READ = 0;
    public static final int WRITE = 1;

    public IntervalLock(int maxPermits) {
        tree = new IntervalTree<>();
        treeLock = new Semaphore(1);
        this.maxPermits = maxPermits;
    }

    /**
     * Acquire a lock on range [start, end] with the specified mode.
     * @param start The start of the interval
     * @param end   The end of the interval
     * @param mode  The mode, either IntervalLock.READ or IntervalLock.WRITE.
     * @return The SemaphoreInterval instance.
     */
    public SemaphoreInterval acquire(int start, int end, int mode) {
        SemaphoreInterval si = new SemaphoreInterval(start, end, mode);
        Set<Semaphore> semaphores = new HashSet<>();

        try {
            treeLock.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace(System.out);
            System.exit(-1);
        }

        Iterator<SemaphoreInterval> overlappers = tree.overlappers(si);
        while (overlappers.hasNext()) {
            SemaphoreInterval i = overlappers.next();
            if (i == null) {
                System.out.println("Error: Getting a null interval");
                System.exit(-1);
            }

            if (i.compareTo(si) == 0)
                continue;

            switch (i.getMode()) {
            case READ:
                if (mode == WRITE) 
                    semaphores.addAll(i.getSemaphores());
                break;
            case WRITE:
                semaphores.addAll(i.getSemaphores());
                break;
            }
        }

        SemaphoreInterval result = tree.insert(si);
        if (result != null)
            si = result;
        si.getSemaphores().add(new Semaphore(0));
        treeLock.release();

        for (Semaphore s: semaphores) {
            try {
                s.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace(System.out);
                System.exit(-1);
            }
        }

        return si;
    }

    /**
     * Release the range lock hold on specified SemaphoreInterval.
     * @param si The semaphore interval returned by the acquire().
     */
    public void release(SemaphoreInterval si) {
        try {
            treeLock.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace(System.out);
            System.exit(-1);
        }
        if (si.getSemaphores() == null || si.getSemaphores().size() == 0) {
            System.out.println("Error: Empty array of semaphores");
            treeLock.release();
            return;
        }
        Semaphore sm = si.getSemaphores().remove(0);
        if (si.getSemaphores().size() == 0) {
            boolean success = tree.delete(si);
            if (!success) {
                System.out.println("Error: Cannot remove an interval.");
                treeLock.release();
                return;
            }
        }
        treeLock.release();
        sm.release(maxPermits);
    }
}

用法

// init the lock with the max permits per semaphore (should be the max number of threads)
public static final IntervalLock lock = new IntervalLock(1000);

// ...

// acquire the lock on range [a, b] (inclusive), with mode (either IntervalLock.READ or IntervalLock.WRITE)
// it returns a SemaphoreInterval instance
SemaphoreInterval si = lock.acquire(a, b, mode);

// ...

// release the acquired lock
lock.release(si);