Skip to content

Commit

Permalink
New page lock transition: write->flush
Browse files Browse the repository at this point in the history 
It's now possible to atomically undo a write lock, and attempt to grab a flush
lock, on a page in PageList.
This will be useful for adding a PF flag that instructs write page cursors to
flush the page on unpin.
This in turn will be useful for certain stages or phases in the parallel batch
importer, where a group of reader threads are reading in pages ahead of a group
of writer threads who then modify those pages. Currently the reads are fielding
most of the page faulting, and thus also end up doing most of the flushing.
That is, the reads end up doing both the read and the write IO.
  • Loading branch information
@chrisvest
chrisvest committed May 26, 2017
1 parent dd91ee6 commit 9164a44
Show file tree
Hide file tree
Showing 3 changed files with 236 additions and 1 deletion.
24 changes: 23 additions & 1 deletion community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/OffHeapPageLock.java
View file
Expand Up @@ -204,7 +204,7 @@ private static boolean failWriteLock( long s, boolean writeCountOverflow )
return false; return false;
} }


private static long throwWriteLockOverflow( long s ) private static void throwWriteLockOverflow( long s )
{ {
throw new IllegalMonitorStateException( "Write lock counter overflow: " + describeState( s ) ); throw new IllegalMonitorStateException( "Write lock counter overflow: " + describeState( s ) );
} }
Expand Down Expand Up @@ -237,6 +237,28 @@ private static long nextSeq( long s )
return (s & SEQ_IMSK) + (s + 1 & SEQ_MASK); return (s & SEQ_IMSK) + (s + 1 & SEQ_MASK);
} }


public static long unlockWriteAndTryTakeFlushLock( long address )
{
long s, n, r;
do
{
r = 0;
s = getState( address );
if ( (s & CNT_MASK) == 0 )
{
throwUnmatchedUnlockWrite( s );
}
n = nextSeq( s ) - CNT_UNIT;
if ( (n & FAE_MASK) == 0 )
{
r = (n += FLS_MASK);
}
}
while ( !compareAndSetState( address, s, n ) );
UnsafeUtil.storeFence();
return r;
}

/** /**
* Grab the exclusive lock if it is immediately available. Exclusive locks will invalidate any overlapping * Grab the exclusive lock if it is immediately available. Exclusive locks will invalidate any overlapping
* optimistic read lock, and fail write and flush locks. If any write or flush locks are currently taken, or if * optimistic read lock, and fail write and flush locks. If any write or flush locks are currently taken, or if
Expand Down
5 changes: 5 additions & 0 deletions community/io/src/main/java/org/neo4j/io/pagecache/impl/muninn/PageList.java
View file
Expand Up @@ -213,6 +213,11 @@ public void unlockWrite( long pageRef )
OffHeapPageLock.unlockWrite( offLock( pageRef ) ); OffHeapPageLock.unlockWrite( offLock( pageRef ) );
} }


public long unlockWriteAndTryTakeFlushLock( long pageRef )
{
return OffHeapPageLock.unlockWriteAndTryTakeFlushLock( offLock( pageRef ) );
}

public boolean tryExclusiveLock( long pageRef ) public boolean tryExclusiveLock( long pageRef )
{ {
return OffHeapPageLock.tryExclusiveLock( offLock( pageRef ) ); return OffHeapPageLock.tryExclusiveLock( offLock( pageRef ) );
Expand Down
208 changes: 208 additions & 0 deletions community/io/src/test/java/org/neo4j/io/pagecache/impl/muninn/PageListTest.java
View file
Expand Up @@ -838,6 +838,22 @@ public void releasingFlushLockMustLowerModifiedFlagIfSuccessful() throws Excepti
assertFalse( pageList.isModified( pageRef ) ); assertFalse( pageList.isModified( pageRef ) );
} }


@Test
public void loweredModifiedFlagMustRemainLoweredAfterReleasingFlushLock() throws Exception
{
pageList.unlockExclusive( pageRef );
assertTrue( pageList.tryWriteLock( pageRef ) );
pageList.unlockWrite( pageRef );
assertTrue( pageList.isModified( pageRef ) );
long s = pageList.tryFlushLock( pageRef );
pageList.unlockFlush( pageRef, s, true );
assertFalse( pageList.isModified( pageRef ) );

s = pageList.tryFlushLock( pageRef );
pageList.unlockFlush( pageRef, s, true );
assertFalse( pageList.isModified( pageRef ) );
}

@Test @Test
public void releasingFlushLockMustNotLowerModifiedFlagIfUnsuccessful() throws Exception public void releasingFlushLockMustNotLowerModifiedFlagIfUnsuccessful() throws Exception
{ {
Expand Down Expand Up @@ -977,6 +993,198 @@ public void allowExclusiveLockedPageToExplicitlyLowerModifiedFlag() throws Excep
pageList.unlockExclusive( pageRef ); pageList.unlockExclusive( pageRef );
} }


@Test
public void unlockWriteAndTryTakeFlushLockMustTakeFlushLock() throws Exception
{
pageList.unlockExclusive( pageRef );
assertTrue( pageList.tryWriteLock( pageRef ) );
long flushStamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertThat( flushStamp, is( not( 0L ) ) );
assertThat( pageList.tryFlushLock( pageRef ), is( 0L ) );
pageList.unlockFlush( pageRef, flushStamp, true );
}

@Test( expected = IllegalMonitorStateException.class )
public void unlockWriteAndTryTakeFlushLockMustThrowIfNotWriteLocked() throws Exception
{
pageList.unlockExclusive( pageRef );
pageList.unlockWriteAndTryTakeFlushLock( pageRef );
}

@Test( expected = IllegalMonitorStateException.class )
public void unlockWriteAndTryTakeFlushLockMustThrowIfNotWriteLockedButExclusiveLocked() throws Exception
{
// exclusive lock implied by constructor
pageList.unlockWriteAndTryTakeFlushLock( pageRef );
}

@Test
public void unlockWriteAndTryTakeFlushLockMustFailIfFlushLockIsAlreadyTaken() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.tryFlushLock( pageRef );
assertThat( stamp, is( not( 0L ) ) );
long secondStamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertThat( secondStamp, is( 0L ) );
pageList.unlockFlush( pageRef, stamp, true );
}

@Test
public void unlockWriteAndTryTakeFlushLockMustReleaseWriteLockEvenIfFlushLockFails() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long flushStamp = pageList.tryFlushLock( pageRef );
assertThat( flushStamp, is( not( 0L ) ) );
assertThat( pageList.unlockWriteAndTryTakeFlushLock( pageRef ), is( 0L ) );
long readStamp = pageList.tryOptimisticReadLock( pageRef );
assertTrue( pageList.validateReadLock( pageRef, readStamp ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockMustReleaseWriteLockWhenFlushLockSucceeds() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
assertThat( pageList.unlockWriteAndTryTakeFlushLock( pageRef ), is( not( 0L ) ) );
long readStamp = pageList.tryOptimisticReadLock( pageRef );
assertTrue( pageList.validateReadLock( pageRef, readStamp ) );
}

@Test
public void unlockWriteAndTrueTakeFlushLockMustRaiseModifiedFlag() throws Exception
{
assertFalse( pageList.isModified( pageRef ) );
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
assertTrue( pageList.isModified( pageRef ) );
assertThat( pageList.unlockWriteAndTryTakeFlushLock( pageRef ), is( not( 0L ) ) );
assertTrue( pageList.isModified( pageRef ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockAndThenUnlockFlushMustLowerModifiedFlagIfSuccessful() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertTrue( pageList.isModified( pageRef ) );
pageList.unlockFlush( pageRef, stamp, true );
assertFalse( pageList.isModified( pageRef ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockAndThenUnlockFlushMustNotLowerModifiedFlagIfFailed() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertTrue( pageList.isModified( pageRef ) );
pageList.unlockFlush( pageRef, stamp, false );
assertTrue( pageList.isModified( pageRef ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockWithOverlappingWriterAndThenUnlockFlushMustNotLowerModifiedFlag()
throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
assertTrue( pageList.tryWriteLock( pageRef ) ); // two write locks, now
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef ); // one flush, one write lock
assertThat( stamp, is( not( 0L ) ) );
pageList.unlockWrite( pageRef ); // one flush, zero write locks
assertTrue( pageList.isModified( pageRef ) );
pageList.unlockFlush( pageRef, stamp, true ); // flush is successful, but had one overlapping writer
assertTrue( pageList.isModified( pageRef ) ); // so it's still modified
}

@Test
public void unlockWriteAndTryTakeFlushLockAndThenUnlockFlushWithOverlappingWriterMustNotLowerModifiedFlag()
throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef ); // one flush lock
assertThat( stamp, is( not( 0L ) ) );
assertTrue( pageList.isModified( pageRef ) );
assertTrue( pageList.tryWriteLock( pageRef ) ); // one flush and one write lock
pageList.unlockFlush( pageRef, stamp, true ); // flush is successful, but have one overlapping writer
pageList.unlockWrite( pageRef ); // no more locks, but a writer started within flush section ...
assertTrue( pageList.isModified( pageRef ) ); // ... and overlapped unlockFlush, so it's still modified
}

@Test
public void unlockWriteAndTryTakeFlushLockAndThenUnlockFlushWithContainedWriterMustNotLowerModifiedFlag()
throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef ); // one flush lock
assertThat( stamp, is( not( 0L ) ) );
assertTrue( pageList.isModified( pageRef ) );
assertTrue( pageList.tryWriteLock( pageRef ) ); // one flush and one write lock
pageList.unlockWrite( pageRef ); // back to one flush lock
pageList.unlockFlush( pageRef, stamp, true ); // flush is successful, but had one overlapping writer
assertTrue( pageList.isModified( pageRef ) ); // so it's still modified
}

@Test
public void unlockWriteAndTryTakeFlushLockThatSucceedsMustPreventOverlappingExclusiveLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
assertFalse( pageList.tryExclusiveLock( pageRef ) );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertFalse( pageList.tryExclusiveLock( pageRef ) );
pageList.unlockFlush( pageRef, stamp, true );
assertTrue( pageList.tryExclusiveLock( pageRef ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockThatFailsMustPreventOverlappingExclusiveLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
assertFalse( pageList.tryExclusiveLock( pageRef ) );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertFalse( pageList.tryExclusiveLock( pageRef ) );
pageList.unlockFlush( pageRef, stamp, false );
assertTrue( pageList.tryExclusiveLock( pageRef ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockThatSucceedsMustPreventOverlappingFlushLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertThat( pageList.tryFlushLock( pageRef ), is( 0L ) );
pageList.unlockFlush( pageRef, stamp, true );
assertThat( pageList.tryFlushLock( pageRef ), is( not( 0L ) ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockThatFailsMustPreventOverlappingFlushLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long stamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertThat( pageList.tryFlushLock( pageRef ), is( 0L ) );
pageList.unlockFlush( pageRef, stamp, false );
assertThat( pageList.tryFlushLock( pageRef ), is( not( 0L ) ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockMustNotInvalidateReadersOverlappingWithFlushLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long flushStamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
long readStamp = pageList.tryOptimisticReadLock( pageRef );
assertTrue( pageList.validateReadLock( pageRef, readStamp ) );
pageList.unlockFlush( pageRef, flushStamp, true );
assertTrue( pageList.validateReadLock( pageRef, readStamp ) );
}

@Test
public void unlockWriteAndTryTakeFlushLockMustInvalidateReadersOverlappingWithWriteLock() throws Exception
{
pageList.unlockExclusiveAndTakeWriteLock( pageRef );
long readStamp = pageList.tryOptimisticReadLock( pageRef );
long flushStamp = pageList.unlockWriteAndTryTakeFlushLock( pageRef );
assertFalse( pageList.validateReadLock( pageRef, readStamp ) );
pageList.unlockFlush( pageRef, flushStamp, true );
assertFalse( pageList.validateReadLock( pageRef, readStamp ) );
}

// xxx ---[ Page state tests ]--- // xxx ---[ Page state tests ]---


@Test @Test
Expand Down

0 comments on commit 9164a44

Please sign in to comment.