|Locks are the mechanism to control concurrent access to data and prevent destructive interaction between operations that use the same resource.|
Tracks tasks (by
Tracks tasks (by
Add the following line to
Refer to Logging.
downgradeLock(blockId: BlockId): Unit
lockForReading( blockId: BlockId, blocking: Boolean = true): Option[BlockInfo]
blockId memory block for reading when the block was registered earlier and no writer tasks use it.
lockForReading prints out the following TRACE message to the logs:
TRACE BlockInfoManager: Task [currentTaskAttemptId] trying to acquire read lock for [blockId]
lockForReading looks up the metadata of the
blockId block (in infos registry).
If no metadata could be found, it returns
None which means that the block does not exist or was removed (and anybody could acquire a write lock).
Otherwise, when the metadata was found, i.e. registered, it checks so-called writerTask. Only when the block has no writer tasks, a read lock can be acquired. If so, the
readerCount of the block metadata is incremented and the block is recorded (in the internal readLocksByTask registry). You should see the following TRACE message in the logs:
TRACE BlockInfoManager: Task [taskAttemptId] acquired read lock for [blockId]
BlockInfo for the
blockId block is returned.
For blocks with
writerTask other than
blocking is enabled,
lockForReading waits (until another thread invokes the
Object.notify method or the
Object.notifyAll methods for this object).
blocking enabled, it will repeat the waiting-for-read-lock sequence until either
None or the lock is obtained.
blocking is disabled and the lock could not be obtained,
None is returned immediately.
lockForWriting( blockId: BlockId, blocking: Boolean = true): Option[BlockInfo]
lockForWriting prints out the following TRACE message to the logs:
TRACE Task [currentTaskAttemptId] trying to acquire write lock for [blockId]
It looks up
blockId in the internal infos registry. When no BlockInfo could be found,
None is returned. Otherwise,
blockId block is checked for
writerTask to be
BlockInfo.NO_WRITER with no readers (i.e.
0) and only then the lock is returned.
When the write lock can be returned,
BlockInfo.writerTask is set to
currentTaskAttemptId and a new binding is added to the internal writeLocksByTask registry. You should see the following TRACE message in the logs:
TRACE Task [currentTaskAttemptId] acquired write lock for [blockId]
If, for some reason,
blockId has a writer or the number of readers is positive (i.e.
BlockInfo.readerCount is greater than
0), the method will wait (based on the input
blocking flag) and attempt the write lock acquisition process until it finishes with a write lock.
(deadlock possible) The method is
None for no
blockId in the internal infos registry or when
blocking flag is disabled and the write lock could not be acquired.
lockNewBlockForWriting( blockId: BlockId, newBlockInfo: BlockInfo): Boolean
lockNewBlockForWriting obtains a write lock for
blockId but only when the method could register the block.
lockNewBlockForWriting prints out the following TRACE message to the logs:
TRACE Task [currentTaskAttemptId] trying to put [blockId]
If some other thread has already created the block, it finishes returning
false. Otherwise, when the block does not exist,
newBlockInfo is recorded in the internal infos registry and the block is locked for this client for writing. It then returns
unlock(blockId: BlockId): Unit
unlock starts by printing out the following TRACE message to the logs:
TRACE BlockInfoManager: Task [currentTaskAttemptId] releasing lock for [blockId]
unlock gets the metadata for
blockId. It may throw a
IllegalStateException if the block was not found.
Otherwise, if the writer task is indeed
NO_WRITER, it is assumed that the
blockId block is locked for reading. The
readerCount counter is decremented for the
blockId block and the read lock removed from the internal readLocksByTask registry for the current task attempt.
In the end,
unlock wakes up all the threads waiting for the
BlockInfoManager (using Java’s Object.notifyAll).
|FIXME What threads could wait?|