class 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);
    }

    //LOG.info(Thread.currentThread().getName() + "-Asked to acquire reader lock from " + parser.toString(startOffset) + " to " + parser.toString(endOffset));

    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();
    }

  }

  public  RangeBasedReaderWriterLock();

}