RangeBasedReaderWriterLock { public LockToken acquireReaderLock(long startOffset, long endOffset, BufferPositionParser parser, String ownerName) throws InterruptedException, TimeoutException { boolean debug = LOG.isDebugEnabled(); if (debug) { LOG.debug("Asked to acquire reader lock from " + parser.toString(startOffset) + " to " + parser.toString(endOffset) + " for " + ownerName); } Range readerRange = new Range(startOffset, endOffset); mutex.lock(); try { boolean timeout = false; while (writerIn && writerRange.intersects(readerRange)) { if ( debug ) { LOG.debug("Waiting for reads to be possible since writer is In. Reader Range is :" + readerRange.toString(parser) + ". Writer Range is :" + writerRange.toString(parser)); } if (timeout) throw new TimeoutException(); if (!readsPossible.await(MAX_LOCK_WAIT_MS, TimeUnit.MILLISECONDS)) timeout = true; if ( debug ) { LOG.info("Waiting for reads to be possible: coming out of wait"); } } LockToken returnVal = new LockToken(readerRange, ownerName); readerRanges.add(returnVal); if (debug) { LOG.debug("Returning with reader lock from " + parser.toString(startOffset) + " to " + parser.toString(endOffset)); } return returnVal; } 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 testAcquireReaderLock() throws Exception { RangeBasedReaderWriterLock lock = new RangeBasedReaderWriterLock(); LockToken token = lock.acquireReaderLock(10, 200,_parser, "testAcquireReaderLock"); AssertJUnit.assertEquals(10, token._id.start); AssertJUnit.assertEquals(200, token._id.end); }