[Feature] Support priority of resource entries

[jialianglinjl]

Issue Description

Type: feature request

Describe what feature you want

Support priority of resource entries.

• Add priority flag support in SphU (ProcessorSlot) (Support prioritized entry in ProcessorSlot interface #255)
• Add priority flag support in Context (implicit for invocation chain)
• Add implementation for prioritized flow control (cluster mode)
• Add optimized implementation for prioritized resource entries (local mode)

[tigerMoon]

hi. can you give more details about why we need priority flag? and the scenario to use it.

[sczyh30]

hi. can you give more details about why we need priority flag? and the scenario to use it.

When requests with two kinds of flag (flag1 and flag2) are coming respectively and simultaneously, we want one kind of request (e.g. flag1) pass first, but the current quota (pass QPS) may be full. Thus some strategies are needed to guarantee the prioritized requests can finally pass rather than rejected.

Perhaps we need a more vivid representation to describe the feature :)

[SugisakiKen]

I write a demo to verify this feature, but maybe there is something wrong with it

/**
 * demo to simulate two operation : read/write , they share same resource
 * read operation use SphU.entryWithPriority() , write operation use SphU.entry()
 * create 25 write thread and 25 read thread,and flow limit count is 20
 * in order to create race condition , use CyclicBarrier to let all thread to start in relatively fair condition
 * when SphU.entryWithPriority() or SphU.entry() success,corresponding counter will add 1
 */
public class FlowQpsReadWritePriorityDemo {

    private static final String WRITE_KEY = "write";
    private static final String READ_KEY = "read";
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(50);
    private static List<String> list = Collections.synchronizedList(new ArrayList<>(30));
    private static AtomicInteger readCounter = new AtomicInteger(0);
    private static AtomicInteger writeCounter = new AtomicInteger(0);
    private static final int threadCountWrite = 25;
    private static final int threadCountRead = 25;
    private static final Object ENTRY_LOCK = new Object();
    private static ConcurrentHashMap<String, Object> lockMap = new ConcurrentHashMap<>();

    public static void main(String[] args) {
        initFlowQpsRule();
        simulateTraffic();
        rouse();
        System.out.println("===== begin to do flow control");
        System.out.println("only 10 requests per second can pass");
    }

    private static void initFlowQpsRule() {
        List<FlowRule> rules = new ArrayList<>();
        FlowRule flowRule = new FlowRule();
        flowRule.setResource(WRITE_KEY);
        flowRule.setCount(20);
        flowRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        rules.add(flowRule);
        FlowRuleManager.loadRules(rules);
    }

    private static void simulateTraffic() {
        for (int i = 0; i < threadCountRead; i++) {
            Thread t = new Thread(new RunTask(WRITE_KEY, true, cyclicBarrier));
            t.setName(String.format("simulate-read-task-%s", i));
            t.start();
        }
        for (int i = 0; i < threadCountWrite; i++) {
            Thread t = new Thread(new RunTask(WRITE_KEY, false, cyclicBarrier));
            t.setName(String.format("simulate-write-task-%s", i));
            t.start();
        }
    }

    private static void rouse() {
        Thread rouser = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("begin to statistic!!!");
                    System.out.println(String.format("pass result:{%s}", list.toString()));
                    System.out.println(String.format("write pass:%s,read pass:%s",
                            writeCounter.get(), readCounter.get()));
                    reset();
                    for (Map.Entry<String, Object> e : lockMap.entrySet()) {
                        synchronized (e.getValue()) {
                            e.getValue().notify();
                        }
                    }
                }
            }
        });
        rouser.setName("sentinel-rouser");
        rouser.start();
    }

    private static void reset() {
        list.clear();
        cyclicBarrier.reset();
        readCounter.set(0);
        writeCounter.set(0);
    }

    static class RunTask extends CyclicBarrierRunnable {
        private String resource;
        private boolean prioritized;

        public RunTask(String resource, boolean prioritized, CyclicBarrier cyclicBarrier) {
            super(cyclicBarrier);
            this.resource = resource;
            this.prioritized = prioritized;
        }

        @Override
        public void execute() {
            Entry entry = null;
            try {
                synchronized (ENTRY_LOCK) {
                    if (!prioritized) {
                        entry = SphU.entry(resource);
                        writeCounter.addAndGet(1);
                    } else {
                        entry = SphU.entryWithPriority(resource);
                        readCounter.addAndGet(1);
                    }
                }
                list.add(Thread.currentThread().getName());
            } catch (Exception e2) {
                // biz exception
            } finally {
                if (entry != null) {
                    entry.exit();
                }
            }
        }
    }

    static abstract class CyclicBarrierRunnable implements Runnable {
        private CyclicBarrier cyclicBarrier;
        private final Object NOTIFIER = new Object();

        public CyclicBarrierRunnable(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        public abstract void execute();

        public void run() {
            lockMap.put(Thread.currentThread().getName(), this.NOTIFIER);
            while (true) {
                try {
                    cyclicBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
                execute();
                try {
                    synchronized (NOTIFIER) {
                        NOTIFIER.wait();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

At the beginning ,console look like this , read operation use entryWithPriority() but not work as expected(read operation has higher priority to acquire)

pass result:{[simulate-write-task-24, simulate-write-task-22, simulate-write-task-23, simulate-write-task-21, simulate-write-task-20, simulate-write-task-19, simulate-write-task-18, simulate-write-task-17, simulate-write-task-16, simulate-write-task-15, simulate-write-task-14, simulate-write-task-13, simulate-write-task-12, simulate-write-task-11, simulate-write-task-10, simulate-write-task-9, simulate-write-task-8, simulate-write-task-7, simulate-write-task-6, simulate-write-task-5]}
write pass:20,read pass:0
begin to statistic!!!
pass result:{[simulate-read-task-24, simulate-write-task-0, simulate-write-task-7, simulate-write-task-1, simulate-read-task-12, simulate-read-task-15, simulate-read-task-3, simulate-read-task-2, simulate-read-task-9, simulate-write-task-13, simulate-read-task-17, simulate-read-task-23, simulate-write-task-17, simulate-write-task-8, simulate-read-task-4, simulate-write-task-24, simulate-write-task-19, simulate-write-task-16, simulate-write-task-5, simulate-write-task-15]}
write pass:11,read pass:9
begin to statistic!!!
pass result:{[simulate-read-task-20, simulate-write-task-23, simulate-write-task-10, simulate-write-task-9, simulate-write-task-17, simulate-write-task-21, simulate-write-task-7, simulate-write-task-2, simulate-write-task-4, simulate-read-task-12, simulate-read-task-15, simulate-read-task-3, simulate-read-task-6, simulate-write-task-19, simulate-read-task-11, simulate-read-task-16, simulate-read-task-14, simulate-read-task-2, simulate-read-task-19, simulate-read-task-0]}
write pass:9,read pass:11
begin to statistic!!!
pass result:{[simulate-read-task-0, simulate-write-task-21, simulate-write-task-22, simulate-write-task-11, simulate-write-task-20, simulate-write-task-9, simulate-write-task-6, simulate-write-task-17, simulate-write-task-18, simulate-write-task-7, simulate-write-task-2, simulate-write-task-4, simulate-read-task-10, simulate-write-task-3, simulate-read-task-7, simulate-read-task-11, simulate-read-task-6, simulate-read-task-15, simulate-read-task-13, simulate-read-task-17]}
write pass:12,read pass:8

After running several times , console output will look like this ,write pass rate is much higher than read pass , it's strange ,but I can't figure it out. Is that this feature not suitable for this scene? or there is something wrong with demo?

begin to statistic!!!
pass result:{[simulate-read-task-24, simulate-write-task-21, simulate-write-task-22, simulate-write-task-24, simulate-write-task-23, simulate-write-task-0, simulate-write-task-10, simulate-write-task-20, simulate-write-task-11, simulate-write-task-12, simulate-write-task-13, simulate-write-task-14, simulate-write-task-9, simulate-write-task-15, simulate-write-task-16, simulate-write-task-17, simulate-write-task-6, simulate-write-task-18, simulate-write-task-5, simulate-write-task-19]}
write pass:19,read pass:1
begin to statistic!!!
pass result:{[simulate-read-task-24, simulate-write-task-22, simulate-write-task-21, simulate-write-task-23, simulate-write-task-24, simulate-write-task-0, simulate-write-task-10, simulate-write-task-20, simulate-write-task-12, simulate-write-task-11, simulate-write-task-13, simulate-write-task-14, simulate-write-task-9, simulate-write-task-15, simulate-write-task-16, simulate-write-task-17, simulate-write-task-6, simulate-write-task-18, simulate-write-task-5, simulate-write-task-8]}
write pass:19,read pass:1

Environment: macOS Mojave 10.14.3