jihoonson commented on a change in pull request #4550: Prioritized locking
URL: https://github.com/apache/incubator-druid/pull/4550#discussion_r331343307
##########
File path:
indexing-service/src/test/java/io/druid/indexing/overlord/TaskLockboxTest.java
##########
@@ -207,4 +258,206 @@ public void testSyncFromStorage() throws
EntryExistsException
Assert.assertEquals(beforeLocksInStorage, afterLocksInStorage);
}
+
+ @Test
+ public void testDoInCriticalSectionWithSharedLock() throws Exception
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-01-02");
+ final Task task = NoopTask.create();
+ lockbox.add(task);
+ Assert.assertTrue(lockbox.tryLock(TaskLockType.SHARED, task,
interval).isOk());
+
+ Assert.assertFalse(
+ lockbox.doInCriticalSection(
+ task,
+ Collections.singletonList(interval),
+ CriticalAction.<Boolean>builder().onValidLocks(() ->
true).onInvalidLocks(() -> false).build()
+ )
+ );
+ }
+
+ @Test
+ public void testDoInCriticalSectionWithExclusiveLock() throws Exception
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-01-02");
+ final Task task = NoopTask.create();
+ lockbox.add(task);
+ final TaskLock lock = lockbox.tryLock(TaskLockType.EXCLUSIVE, task,
interval).getTaskLock();
+ Assert.assertNotNull(lock);
+
+ Assert.assertTrue(
+ lockbox.doInCriticalSection(
+ task,
+ Collections.singletonList(interval),
+ CriticalAction.<Boolean>builder().onValidLocks(() ->
true).onInvalidLocks(() -> false).build()
+ )
+ );
+ }
+
+ @Test
+ public void testDoInCriticalSectionWithSmallerInterval() throws Exception
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-02-01");
+ final Interval smallInterval = Intervals.of("2017-01-10/2017-01-11");
+ final Task task = NoopTask.create();
+ lockbox.add(task);
+ final TaskLock lock = lockbox.tryLock(TaskLockType.EXCLUSIVE, task,
interval).getTaskLock();
+ Assert.assertNotNull(lock);
+
+ Assert.assertTrue(
+ lockbox.doInCriticalSection(
+ task,
+ Collections.singletonList(smallInterval),
+ CriticalAction.<Boolean>builder().onValidLocks(() ->
true).onInvalidLocks(() -> false).build()
+ )
+ );
+ }
+
+ @Test
+ public void testPreemptionAndDoInCriticalSection() throws Exception
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-01-02");
+ for (int i = 0; i < 5; i++) {
+ final Task task = NoopTask.create();
+ lockbox.add(task);
+ taskStorage.insert(task, TaskStatus.running(task.getId()));
+ Assert.assertTrue(lockbox.tryLock(TaskLockType.SHARED, task,
interval).isOk());
+ }
+
+ final Task highPriorityTask = NoopTask.create(100);
+ lockbox.add(highPriorityTask);
+ taskStorage.insert(highPriorityTask,
TaskStatus.running(highPriorityTask.getId()));
+ final TaskLock lock = lockbox.tryLock(TaskLockType.EXCLUSIVE,
highPriorityTask, interval).getTaskLock();
+ Assert.assertNotNull(lock);
+
+ Assert.assertTrue(
+ lockbox.doInCriticalSection(
+ highPriorityTask,
+ Collections.singletonList(interval),
+ CriticalAction.<Boolean>builder().onValidLocks(() ->
true).onInvalidLocks(() -> false).build()
+ )
+ );
+ }
+
+ @Test
+ public void testDoInCriticalSectionWithRevokedLock() throws Exception
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-01-02");
+ final Task lowPriorityTask = NoopTask.create("task1", 0);
+ final Task highPriorityTask = NoopTask.create("task2", 10);
+ lockbox.add(lowPriorityTask);
+ lockbox.add(highPriorityTask);
+ taskStorage.insert(lowPriorityTask,
TaskStatus.running(lowPriorityTask.getId()));
+ taskStorage.insert(highPriorityTask,
TaskStatus.running(highPriorityTask.getId()));
+
+ final TaskLock lowPriorityLock = lockbox.tryLock(TaskLockType.EXCLUSIVE,
lowPriorityTask, interval).getTaskLock();
+ Assert.assertNotNull(lowPriorityLock);
+ Assert.assertTrue(lockbox.tryLock(TaskLockType.EXCLUSIVE,
highPriorityTask, interval).isOk());
+
Assert.assertTrue(Iterables.getOnlyElement(lockbox.findLocksForTask(lowPriorityTask)).isRevoked());
+
+ Assert.assertFalse(
+ lockbox.doInCriticalSection(
+ lowPriorityTask,
+ Collections.singletonList(interval),
+ CriticalAction.<Boolean>builder().onValidLocks(() ->
true).onInvalidLocks(() -> false).build()
+ )
+ );
+ }
+
+ @Test(timeout = 5000L)
+ public void testAcquireLockAfterRevoked() throws EntryExistsException,
InterruptedException
+ {
+ final Interval interval = Intervals.of("2017-01-01/2017-01-02");
+ final Task lowPriorityTask = NoopTask.create("task1", 0);
+ final Task highPriorityTask = NoopTask.create("task2", 10);
+ lockbox.add(lowPriorityTask);
+ lockbox.add(highPriorityTask);
+ taskStorage.insert(lowPriorityTask,
TaskStatus.running(lowPriorityTask.getId()));
+ taskStorage.insert(highPriorityTask,
TaskStatus.running(highPriorityTask.getId()));
+
+ final TaskLock lowPriorityLock = lockbox.lock(TaskLockType.EXCLUSIVE,
lowPriorityTask, interval).getTaskLock();
+ Assert.assertNotNull(lowPriorityLock);
+ Assert.assertTrue(lockbox.tryLock(TaskLockType.EXCLUSIVE,
highPriorityTask, interval).isOk());
+
Assert.assertTrue(Iterables.getOnlyElement(lockbox.findLocksForTask(lowPriorityTask)).isRevoked());
+
+ lockbox.unlock(highPriorityTask, interval);
+
+ // Acquire again
+ final LockResult lockResult = lockbox.lock(TaskLockType.EXCLUSIVE,
lowPriorityTask, interval);
+ Assert.assertFalse(lockResult.isOk());
Review comment:
Oh, I see. Perhaps this could be related to the lame task state tracking in
the overlord. :sob:
I'm not sure if it's the best idea but I think we can change the return type
of `isReady()` to some enum value which could be one of `READY`, `NOT_READY`,
and `CANNOT_READY`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
