RangeBasedReaderWriterLock { public void acquireWriterLock(long start, long end, BufferPositionParser parser) throws InterruptedException, TimeoutException { long startOffset = parser.address(start); long endOffset = parser.address(end); boolean debug = LOG.isDebugEnabled(); if (debug) LOG.debug("Acquiring writer lock from " + parser.toString(start) + " to " + parser.toString(end)); mutex.lock(); try { boolean timeout = false; while (!readerRanges.isEmpty() && Range.contains(startOffset, endOffset, parser.address(readerRanges.peek()._id.start))) { if ( debug ) { LOG.debug("Entering wait because reader(s) exist: Writer Range: [" + parser.toString(start) + "(Address:" + parser.toString(startOffset) + ")-" + parser.toString(end) + "(Address:" + parser.toString(endOffset) + ")]. Nearest Reader Range :" + readerRanges.peek().toString(parser)); } if (timeout) { LOG.error("timed out waiting for a write lock for [" + parser.toString(start) + "," + parser.toString(end) + "); this: " + this ); throw new TimeoutException(); } for (LockToken token: readerRanges) { LOG.info(token.toString(parser)); } writerWaiting = true; if (!writesPossible.await(MAX_LOCK_WAIT_MS, TimeUnit.MILLISECONDS)) timeout = true; if ( debug ) LOG.debug("Writer coming out of wait"); } writerWaiting = false; writerIn = true; writerRange.start = start; writerRange.end = end; } finally { mutex.unlock(); } } RangeBasedReaderWriterLock(); LockToken acquireReaderLock(long startOffset, long endOffset, BufferPositionParser parser,
                                     String ownerName); void shiftReaderLockStart(LockToken lockId, long newStartOffset, BufferPositionParser parser); void shiftReaderLockStartIfWriterWaiting(LockToken lockId, long newStartOffset, BufferPositionParser parser); void releaseReaderLock(LockToken lockId); void acquireWriterLock(long start, long end, BufferPositionParser parser); void releaseWriterLock(BufferPositionParser parser); String toString(BufferPositionParser parser, boolean doSort); boolean isWriterIn(); boolean isWriterWaiting(); Range getWriterRange(); int getNumReaders(); @Override String toString(); static final String MODULE; static final Logger LOG; }
@Test public void testAcquireWriterLock() throws InterruptedException, TimeoutException { final Logger log = Logger.getLogger("TestRangeBasedReaderWriterLock.testAcquireWriterLock"); final RangeBasedReaderWriterLock lock = new RangeBasedReaderWriterLock(); { LockToken token1 = lock.acquireReaderLock(_parser.setGenId(1000, 1), _parser.setGenId(20000, 1), _parser, "testAcquireWriterLock1"); LockToken token2 = lock.acquireReaderLock(_parser.setGenId(1000,1), _parser.setGenId(20000,1), _parser, "testAcquireWriterLock2"); LockToken token3 = lock.acquireReaderLock(_parser.setGenId(10000,1), _parser.setGenId(20000,1), _parser, "testAcquireWriterLock3"); Runnable writer = new Runnable() { @Override public void run() { try { lock.acquireWriterLock(_parser.setGenId(0,1), _parser.setGenId(1001,1), _parser); } catch (InterruptedException e) { log.error(e); } catch (TimeoutException e) { log.error(e); } } }; Thread writerThread = new Thread(writer); writerThread.start(); try { Thread.sleep(100); } catch ( InterruptedException ie) { throw new RuntimeException(ie); } AssertJUnit.assertTrue(lock.isWriterWaiting()); AssertJUnit.assertFalse(lock.isWriterIn()); lock.releaseReaderLock(token1); try { Thread.sleep(100); } catch ( InterruptedException ie) { throw new RuntimeException(ie); } AssertJUnit.assertTrue(lock.isWriterWaiting()); AssertJUnit.assertFalse(lock.isWriterIn()); lock.releaseReaderLock(token2); try { writerThread.join(1000); } catch (InterruptedException ie) {throw new RuntimeException(ie); } AssertJUnit.assertTrue(lock.isWriterIn()); AssertJUnit.assertFalse(lock.isWriterWaiting()); AssertJUnit.assertFalse(writerThread.isAlive()); lock.releaseReaderLock(token3); lock.releaseWriterLock(_parser); } { LockToken token1 = lock.acquireReaderLock(_parser.setGenId(10, 1), _parser.setGenId(200, 1), _parser, "testAcquireWriterLock4"); LockToken token2 = lock.acquireReaderLock(_parser.setGenId(10,1), _parser.setGenId(2000,1), _parser, "testAcquireWriterLock5"); LockToken token3 = lock.acquireReaderLock(2000, 3000, _parser, "testAcquireWriterLock6"); Runnable writer = new Runnable() { @Override public void run() { try { lock.acquireWriterLock(_parser.setGenId(2000,1), _parser.setGenId(11000,1), _parser); } catch (InterruptedException e) { log.error(e); } catch (TimeoutException e) { log.error(e); } } }; Thread writerThread = new Thread(writer); writerThread.start(); try { Thread.sleep(100); } catch ( InterruptedException ie) { throw new RuntimeException(ie); } AssertJUnit.assertTrue(lock.isWriterWaiting()); AssertJUnit.assertFalse(lock.isWriterIn()); lock.releaseReaderLock(token3); try { writerThread.join(1000); } catch (InterruptedException ie) {throw new RuntimeException(ie); } AssertJUnit.assertTrue(lock.isWriterIn()); AssertJUnit.assertFalse(lock.isWriterWaiting()); AssertJUnit.assertFalse(writerThread.isAlive()); lock.releaseReaderLock(token1); lock.releaseReaderLock(token2); lock.releaseWriterLock(_parser); } }