@Test public void pickExecutorTask() { Lock lock = new ReentrantLock(); ExecutorTask pickedTask; pickedTask = TaskRouter.getInstance().pickExecutorTask(lock); Assert.assertEquals(pickedTask.getSourceKey().toString(), "focus8"); ConsumeQueue.getInstance().offer(pickedTask); pickedTask = TaskRouter.getInstance().pickExecutorTask(lock); Assert.assertEquals(pickedTask.getSourceKey().toString(), "focus7"); ConsumeQueue.getInstance().offer(pickedTask); Mockito.when(task4.getSourceKey()).thenReturn(new SourceKey("focus4")); Mockito.when(task6.getSourceKey()).thenReturn(new SourceKey("focus6")); pickedTask = TaskRouter.getInstance().pickExecutorTask(lock); Assert.assertEquals(pickedTask.getSourceKey().toString(), "focus6"); ConsumeQueue.getInstance().offer(pickedTask); Mockito.when(task5.getLockType()).thenReturn(LockType.VIRTUAL_SEG); pickedTask = TaskRouter.getInstance().pickExecutorTask(lock); Assert.assertEquals(pickedTask.getSourceKey().toString(), "focus_point"); ConsumeQueue.getInstance().offer(pickedTask); Mockito.when(task4.getLockType()).thenReturn(LockType.VIRTUAL_SEG); Mockito.when(task4.getSourceKey()).thenReturn(new SourceKey("focus_point")); pickedTask = TaskRouter.getInstance().pickExecutorTask(lock); Assert.assertNull(pickedTask); }
public synchronized ExecutorTask pickExecutorTask(Lock lock) { MultiSkipList<ExecutorTask> idleTasks = TaskRouter.getInstance().getIdleTasks(); ExecutorTask taskPicked = null; lock.lock(); try { taskConflict.initVirtualLocks(ConsumeQueue.getInstance().getTaskList()); Iterator<ExecutorTask> iterator = idleTasks.iterator(); while (iterator.hasNext()) { ExecutorTask curTask = iterator.next(); if (isQualified(curTask)) { taskPicked = curTask; break; } } } finally { lock.unlock(); } taskConflict.finishVirtualLocks(); if (taskPicked != null && TaskRouter.getInstance().remove(taskPicked)) { return taskPicked; } else { return null; } }
TaskRouter { public synchronized ExecutorTask pickExecutorTask(Lock lock) { MultiSkipList<ExecutorTask> idleTasks = TaskRouter.getInstance().getIdleTasks(); ExecutorTask taskPicked = null; lock.lock(); try { taskConflict.initVirtualLocks(ConsumeQueue.getInstance().getTaskList()); Iterator<ExecutorTask> iterator = idleTasks.iterator(); while (iterator.hasNext()) { ExecutorTask curTask = iterator.next(); if (isQualified(curTask)) { taskPicked = curTask; break; } } } finally { lock.unlock(); } taskConflict.finishVirtualLocks(); if (taskPicked != null && TaskRouter.getInstance().remove(taskPicked)) { return taskPicked; } else { return null; } } }
TaskRouter { public synchronized ExecutorTask pickExecutorTask(Lock lock) { MultiSkipList<ExecutorTask> idleTasks = TaskRouter.getInstance().getIdleTasks(); ExecutorTask taskPicked = null; lock.lock(); try { taskConflict.initVirtualLocks(ConsumeQueue.getInstance().getTaskList()); Iterator<ExecutorTask> iterator = idleTasks.iterator(); while (iterator.hasNext()) { ExecutorTask curTask = iterator.next(); if (isQualified(curTask)) { taskPicked = curTask; break; } } } finally { lock.unlock(); } taskConflict.finishVirtualLocks(); if (taskPicked != null && TaskRouter.getInstance().remove(taskPicked)) { return taskPicked; } else { return null; } } private  TaskRouter(); }
TaskRouter { public synchronized ExecutorTask pickExecutorTask(Lock lock) { MultiSkipList<ExecutorTask> idleTasks = TaskRouter.getInstance().getIdleTasks(); ExecutorTask taskPicked = null; lock.lock(); try { taskConflict.initVirtualLocks(ConsumeQueue.getInstance().getTaskList()); Iterator<ExecutorTask> iterator = idleTasks.iterator(); while (iterator.hasNext()) { ExecutorTask curTask = iterator.next(); if (isQualified(curTask)) { taskPicked = curTask; break; } } } finally { lock.unlock(); } taskConflict.finishVirtualLocks(); if (taskPicked != null && TaskRouter.getInstance().remove(taskPicked)) { return taskPicked; } else { return null; } } private  TaskRouter(); static TaskRouter getInstance(); synchronized boolean addTasks(List<ExecutorTask> taskList); MultiSkipList<ExecutorTask> getIdleTasks(); synchronized boolean moveTask(ExecutorTask task, int position); synchronized Pair<Long, List<ExecutorTask>> getTaskView(boolean refresh); synchronized boolean remove(ExecutorTask task); synchronized ExecutorTask pickExecutorTask(Lock lock); synchronized void clear(); }
TaskRouter { public synchronized ExecutorTask pickExecutorTask(Lock lock) { MultiSkipList<ExecutorTask> idleTasks = TaskRouter.getInstance().getIdleTasks(); ExecutorTask taskPicked = null; lock.lock(); try { taskConflict.initVirtualLocks(ConsumeQueue.getInstance().getTaskList()); Iterator<ExecutorTask> iterator = idleTasks.iterator(); while (iterator.hasNext()) { ExecutorTask curTask = iterator.next(); if (isQualified(curTask)) { taskPicked = curTask; break; } } } finally { lock.unlock(); } taskConflict.finishVirtualLocks(); if (taskPicked != null && TaskRouter.getInstance().remove(taskPicked)) { return taskPicked; } else { return null; } } private  TaskRouter(); static TaskRouter getInstance(); synchronized boolean addTasks(List<ExecutorTask> taskList); MultiSkipList<ExecutorTask> getIdleTasks(); synchronized boolean moveTask(ExecutorTask task, int position); synchronized Pair<Long, List<ExecutorTask>> getTaskView(boolean refresh); synchronized boolean remove(ExecutorTask task); synchronized ExecutorTask pickExecutorTask(Lock lock); synchronized void clear();  }