alluxio.worker.block

Class DefaultBlockWorker

java.lang.Object

alluxio.worker.AbstractWorker

alluxio.worker.block.DefaultBlockWorker

All Implemented Interfaces:

Server<WorkerNetAddress>, BlockWorker, SessionCleanable, Worker

@NotThreadSafe
public final class DefaultBlockWorker
extends AbstractWorker
implements BlockWorker

The class is responsible for managing all top level components of the Block Worker. This includes: Periodic Threads: BlockMasterSync (Worker to Master continuous communication) Logic: DefaultBlockWorker (Logic for all block related storage operations)

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	DefaultBlockWorker.Metrics This class contains some metrics related to the block worker.
class	DefaultBlockWorker.StorageChecker StorageChecker periodically checks the health of each storage path and report missing blocks to BlockWorker.

Method Summary

Modifier and Type	Method and Description
void	abortBlock(long sessionId, long blockId) Aborts the temporary block created by the session.
void	accessBlock(long sessionId, long blockId) Access the block for a given session.
void	cleanupSession(long sessionId) Cleans up the session.
void	clearMetrics() Clears the worker metrics.
void	closeUfsBlock(long sessionId, long blockId) Closes a UFS block for a client session.
void	commitBlock(long sessionId, long blockId, boolean pinOnCreate) Commits a block to Alluxio managed space.
void	commitBlockInUfs(long blockId, long length) Commits a block in UFS.
String	createBlock(long sessionId, long blockId, String tierAlias, String medium, long initialBytes) Creates a block in Alluxio managed space for short-circuit writes.
void	createBlockRemote(long sessionId, long blockId, String tierAlias, String medium, long initialBytes) Creates a block for non short-circuit writes or caching requests.
void	freeSpace(long sessionId, long availableBytes, String tierAlias) Frees space to make a specific amount of bytes available in a best-effort way in the tier.
BlockMeta	getBlockMeta(long sessionId, long blockId, long lockId) Gets the metadata of a specific block from local storage.
BlockStore	getBlockStore()
Set<Class<? extends Server>>	getDependencies()
FileInfo	getFileInfo(long fileId) Gets the file information.
String	getName()
BlockHeartbeatReport	getReport() Gets a report for the periodic heartbeat to master.
Map<ServiceType,GrpcService>	getServices()
BlockStoreMeta	getStoreMeta() Gets the metadata for the entire block store.
BlockStoreMeta	getStoreMetaFull() Similar as BlockWorker.getStoreMeta() except that this also contains full blockId list.
BlockWriter	getTempBlockWriterRemote(long sessionId, long blockId) Opens a BlockWriter for an existing temporary block for non short-circuit writes or cache requests.
BlockMeta	getVolatileBlockMeta(long blockId) Gets the metadata of a block given its blockId or throws IOException.
AtomicReference<Long>	getWorkerId()
boolean	hasBlockMeta(long blockId) Checks if the storage has a given block.
long	lockBlock(long sessionId, long blockId) Obtains a read lock the block.
long	lockBlockNoException(long sessionId, long blockId) Obtains a read lock the block without throwing an exception.
void	moveBlock(long sessionId, long blockId, String tierAlias) Moves a block from its current location to a target location, currently only tier level moves are supported.
void	moveBlockToMedium(long sessionId, long blockId, String mediumType) Moves a block from its current location to a target location, with a specific medium type.
boolean	openUfsBlock(long sessionId, long blockId, Protocol.OpenUfsBlockOptions options) Opens a UFS block.
String	readBlock(long sessionId, long blockId, long lockId) Gets the path to the block file in local storage.
BlockReader	readBlockRemote(long sessionId, long blockId, long lockId) Gets the block reader for the block for non short-circuit reads.
BlockReader	readUfsBlock(long sessionId, long blockId, long offset, boolean positionShort) Gets a block reader to read a UFS block for non short-circuit reads.
void	removeBlock(long sessionId, long blockId) Frees a block from Alluxio managed space.
void	requestSpace(long sessionId, long blockId, long additionalBytes) Request an amount of space for a block in its storage directory.
void	sessionHeartbeat(long sessionId) Handles the heartbeat from a client.
void	start(WorkerNetAddress address) Runs the block worker.
void	stop() Stops the block worker.
void	unlockBlock(long lockId) Releases the lock with the specified lock id.
boolean	unlockBlock(long sessionId, long blockId) Releases the lock with the specified session and block id.
void	updatePinList(Set<Long> pinnedInodes) Sets the pinlist for the underlying block store.

Methods inherited from class alluxio.worker.AbstractWorker

close, getExecutorService

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface alluxio.Server

close

Method Detail

getDependencies

public Set<Class<? extends Server>> getDependencies()

Specified by:

getDependencies in interface Server<WorkerNetAddress>

Returns:

the server dependencies

getName

public String getName()

Specified by:

getName in interface Server<WorkerNetAddress>

Returns:

the server's name

getBlockStore

public BlockStore getBlockStore()

Specified by:

getBlockStore in interface BlockWorker

Returns:

the worker data service bind host

getServices

public Map<ServiceType,GrpcService> getServices()

Specified by:

getServices in interface Server<WorkerNetAddress>

Returns:

a map from service names to gRPC serviced that serve RPCs for this server

getWorkerId

public AtomicReference<Long> getWorkerId()

Specified by:

getWorkerId in interface BlockWorker

Returns:

the worker id

start

public void start(WorkerNetAddress address)
           throws IOException

Runs the block worker. The thread must be called after all services (e.g., web, dataserver) started. BlockWorker doesn't support being restarted!

Specified by:

start in interface Server<WorkerNetAddress>

Overrides:

start in class AbstractWorker

Parameters:

address - the start options

Throws:

IOException

stop

public void stop()
          throws IOException

Stops the block worker. This method should only be called to terminate the worker. BlockWorker doesn't support being restarted!

Specified by:

stop in interface Server<WorkerNetAddress>

Overrides:

stop in class AbstractWorker

Throws:

IOException

abortBlock

public void abortBlock(long sessionId,
                       long blockId)
                throws BlockAlreadyExistsException,
                       BlockDoesNotExistException,
                       InvalidWorkerStateException,
                       IOException

Description copied from interface: BlockWorker

Aborts the temporary block created by the session.

Specified by:

abortBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be aborted

Throws:

BlockAlreadyExistsException - if blockId already exists in committed blocks

BlockDoesNotExistException - if the temporary block cannot be found

InvalidWorkerStateException - if blockId does not belong to sessionId

IOException

accessBlock

public void accessBlock(long sessionId,
                        long blockId)
                 throws BlockDoesNotExistException

Description copied from interface: BlockWorker

Access the block for a given session. This should be called to update the evictor when necessary.

Specified by:

accessBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to access

Throws:

BlockDoesNotExistException - this exception is not thrown in the tiered block store implementation

commitBlock

public void commitBlock(long sessionId,
                        long blockId,
                        boolean pinOnCreate)
                 throws BlockAlreadyExistsException,
                        BlockDoesNotExistException,
                        InvalidWorkerStateException,
                        IOException,
                        WorkerOutOfSpaceException

Description copied from interface: BlockWorker

Commits a block to Alluxio managed space. The block must be temporary. The block is persisted after BlockStore.commitBlock(long, long, boolean). The block will not be accessible until BlockMasterClient.commitBlock(long, long, String, String, long, long) succeeds.

Specified by:

commitBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to commit

pinOnCreate - whether to pin block on create

Throws:

BlockAlreadyExistsException - if blockId already exists in committed blocks

BlockDoesNotExistException - if the temporary block cannot be found

InvalidWorkerStateException - if blockId does not belong to sessionId

WorkerOutOfSpaceException - if there is no more space left to hold the block

IOException

commitBlockInUfs

public void commitBlockInUfs(long blockId,
                             long length)
                      throws IOException

Description copied from interface: BlockWorker

Commits a block in UFS.

Specified by:

commitBlockInUfs in interface BlockWorker

Parameters:

blockId - the id of the block to commit

length - length of the block to commit

Throws:

IOException

createBlock

public String createBlock(long sessionId,
                          long blockId,
                          String tierAlias,
                          String medium,
                          long initialBytes)
                   throws BlockAlreadyExistsException,
                          WorkerOutOfSpaceException,
                          IOException

Description copied from interface: BlockWorker

Creates a block in Alluxio managed space for short-circuit writes. The block will be temporary until it is committed. Throws an IllegalArgumentException if the location does not belong to tiered storage.

Specified by:

createBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to create

tierAlias - the alias of the tier to place the new block in, BlockStoreLocation.ANY_TIER for any tier

medium - the name of the medium to place the new block in

initialBytes - the initial amount of bytes to be allocated

Returns:

a string representing the path to the local file

Throws:

BlockAlreadyExistsException - if blockId already exists, either temporary or committed, or block in eviction plan already exists

WorkerOutOfSpaceException - if this Store has no more space than the initialBlockSize

IOException

createBlockRemote

public void createBlockRemote(long sessionId,
                              long blockId,
                              String tierAlias,
                              String medium,
                              long initialBytes)
                       throws BlockAlreadyExistsException,
                              WorkerOutOfSpaceException,
                              IOException

Description copied from interface: BlockWorker

Creates a block for non short-circuit writes or caching requests. Calls BlockWorker.getTempBlockWriterRemote(long, long) to get a writer for writing to the block. Throws an IllegalArgumentException if the location does not belong to tiered storage.

Specified by:

createBlockRemote in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be created

tierAlias - the alias of the tier to place the new block in

medium - the name of the medium to place the new block in

initialBytes - the initial amount of bytes to be allocated

Throws:

BlockAlreadyExistsException - if blockId already exists, either temporary or committed, or block in eviction plan already exists

WorkerOutOfSpaceException - if this Store has no more space than the initialBlockSize

IOException

freeSpace

public void freeSpace(long sessionId,
                      long availableBytes,
                      String tierAlias)
               throws WorkerOutOfSpaceException,
                      BlockDoesNotExistException,
                      IOException,
                      BlockAlreadyExistsException,
                      InvalidWorkerStateException

Description copied from interface: BlockWorker

Frees space to make a specific amount of bytes available in a best-effort way in the tier. The implementation should try to free at least 1 byte from the worker, otherwise a WorkerOutOfSpaceException should be thrown if no space is available.

Specified by:

freeSpace in interface BlockWorker

Parameters:

sessionId - the session id

availableBytes - the amount of free space in bytes

tierAlias - the alias of the tier to free space

Throws:

WorkerOutOfSpaceException - if there is not enough space

BlockDoesNotExistException - if blocks can not be found

BlockAlreadyExistsException - if blocks to move already exists in destination location

InvalidWorkerStateException - if blocks to move/evict is uncommitted

IOException

getTempBlockWriterRemote

public BlockWriter getTempBlockWriterRemote(long sessionId,
                                            long blockId)
                                     throws BlockDoesNotExistException,
                                            BlockAlreadyExistsException,
                                            InvalidWorkerStateException,
                                            IOException

Description copied from interface: BlockWorker

Opens a BlockWriter for an existing temporary block for non short-circuit writes or cache requests. The temporary block must already exist with BlockWorker.createBlockRemote(long, long, String, String, long).

Specified by:

getTempBlockWriterRemote in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be opened for writing

Returns:

the block writer for the local block file

Throws:

BlockDoesNotExistException - if the block cannot be found

BlockAlreadyExistsException - if a committed block with the same ID exists

InvalidWorkerStateException - if the worker state is invalid

IOException

getReport

public BlockHeartbeatReport getReport()

Description copied from interface: BlockWorker

Gets a report for the periodic heartbeat to master. Contains the blocks added since the last heart beat and blocks removed since the last heartbeat.

Specified by:

getReport in interface BlockWorker

Returns:

a block heartbeat report

getStoreMeta

public BlockStoreMeta getStoreMeta()

Description copied from interface: BlockWorker

Gets the metadata for the entire block store. Contains the block mapping per storage dir and the total capacity and used capacity of each tier. This function is cheap.

Specified by:

getStoreMeta in interface BlockWorker

Returns:

the block store metadata

getStoreMetaFull

public BlockStoreMeta getStoreMetaFull()

Description copied from interface: BlockWorker

Similar as BlockWorker.getStoreMeta() except that this also contains full blockId list. This function is expensive.

Specified by:

getStoreMetaFull in interface BlockWorker

Returns:

the full block store metadata

getVolatileBlockMeta

public BlockMeta getVolatileBlockMeta(long blockId)
                               throws BlockDoesNotExistException

Description copied from interface: BlockWorker

Gets the metadata of a block given its blockId or throws IOException. This method does not require a lock id so the block is possible to be moved or removed after it returns.

Specified by:

getVolatileBlockMeta in interface BlockWorker

Parameters:

blockId - the block id

Returns:

metadata of the block

Throws:

BlockDoesNotExistException - if no BlockMeta for this blockId is found

getBlockMeta

public BlockMeta getBlockMeta(long sessionId,
                              long blockId,
                              long lockId)
                       throws BlockDoesNotExistException,
                              InvalidWorkerStateException

Description copied from interface: BlockWorker

Gets the metadata of a specific block from local storage.

Unlike BlockWorker.getVolatileBlockMeta(long), this method requires the lock id returned by a previously acquired BlockWorker.lockBlock(long, long).

Specified by:

getBlockMeta in interface BlockWorker

Parameters:

sessionId - the id of the session to get this file

blockId - the id of the block

lockId - the id of the lock

Returns:

metadata of the block

Throws:

BlockDoesNotExistException - if the block id can not be found in committed blocks or lockId can not be found

InvalidWorkerStateException - if session id or block id is not the same as that in the LockRecord of lockId

hasBlockMeta

public boolean hasBlockMeta(long blockId)

Description copied from interface: BlockWorker

Checks if the storage has a given block.

Specified by:

hasBlockMeta in interface BlockWorker

Parameters:

blockId - the block id

Returns:

true if the block is contained, false otherwise

lockBlock

public long lockBlock(long sessionId,
                      long blockId)
               throws BlockDoesNotExistException

Description copied from interface: BlockWorker

Obtains a read lock the block.

Specified by:

lockBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be locked

Returns:

the lock id that uniquely identifies the lock obtained

Throws:

BlockDoesNotExistException - if blockId cannot be found, for example, evicted already

lockBlockNoException

public long lockBlockNoException(long sessionId,
                                 long blockId)

Description copied from interface: BlockWorker

Obtains a read lock the block without throwing an exception. If the lock fails, return BlockLockManager.INVALID_LOCK_ID.

Specified by:

lockBlockNoException in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be locked

Returns:

the lock id that uniquely identifies the lock obtained or BlockLockManager.INVALID_LOCK_ID if it failed to lock

moveBlock

public void moveBlock(long sessionId,
                      long blockId,
                      String tierAlias)
               throws BlockDoesNotExistException,
                      BlockAlreadyExistsException,
                      InvalidWorkerStateException,
                      WorkerOutOfSpaceException,
                      IOException

Description copied from interface: BlockWorker

Moves a block from its current location to a target location, currently only tier level moves are supported. Throws an IllegalArgumentException if the tierAlias is out of range of tiered storage.

Specified by:

moveBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to move

tierAlias - the alias of the tier to move the block to

Throws:

BlockDoesNotExistException - if blockId cannot be found

BlockAlreadyExistsException - if blockId already exists in committed blocks of the newLocation

InvalidWorkerStateException - if blockId has not been committed

WorkerOutOfSpaceException - if newLocation does not have enough extra space to hold the block

IOException

moveBlockToMedium

public void moveBlockToMedium(long sessionId,
                              long blockId,
                              String mediumType)
                       throws BlockDoesNotExistException,
                              BlockAlreadyExistsException,
                              InvalidWorkerStateException,
                              WorkerOutOfSpaceException,
                              IOException

Description copied from interface: BlockWorker

Moves a block from its current location to a target location, with a specific medium type. Throws an IllegalArgumentException if the medium type is not one of the listed medium types.

Specified by:

moveBlockToMedium in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to move

mediumType - the medium type to move to

Throws:

BlockDoesNotExistException - if blockId cannot be found

BlockAlreadyExistsException - if blockId already exists in committed blocks of the newLocation

InvalidWorkerStateException - if blockId has not been committed

WorkerOutOfSpaceException - if newLocation does not have enough extra space to hold the block

IOException

readBlock

public String readBlock(long sessionId,
                        long blockId,
                        long lockId)
                 throws BlockDoesNotExistException,
                        InvalidWorkerStateException

Description copied from interface: BlockWorker

Gets the path to the block file in local storage. The block must be a permanent block, and the caller must first obtain the lock on the block.

Specified by:

readBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to read

lockId - the id of the lock on this block

Returns:

a string representing the path to this block in local storage

Throws:

BlockDoesNotExistException - if the blockId cannot be found in committed blocks or lockId cannot be found

InvalidWorkerStateException - if sessionId or blockId is not the same as that in the LockRecord of lockId

readBlockRemote

public BlockReader readBlockRemote(long sessionId,
                                   long blockId,
                                   long lockId)
                            throws BlockDoesNotExistException,
                                   InvalidWorkerStateException,
                                   IOException

Description copied from interface: BlockWorker

Gets the block reader for the block for non short-circuit reads.

Specified by:

readBlockRemote in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to read

lockId - the id of the lock on this block

Returns:

the block reader for the block

Throws:

BlockDoesNotExistException - if lockId is not found

InvalidWorkerStateException - if sessionId or blockId is not the same as that in the LockRecord of lockId

IOException

readUfsBlock

public BlockReader readUfsBlock(long sessionId,
                                long blockId,
                                long offset,
                                boolean positionShort)
                         throws BlockDoesNotExistException,
                                IOException

Description copied from interface: BlockWorker

Gets a block reader to read a UFS block for non short-circuit reads.

Specified by:

readUfsBlock in interface BlockWorker

Parameters:

sessionId - the client session ID

blockId - the ID of the UFS block to read

offset - the offset within the block

positionShort - whether the operation is using positioned read to a small buffer size

Returns:

the block reader instance

Throws:

BlockDoesNotExistException - if the block does not exist in the UFS block store

IOException

removeBlock

public void removeBlock(long sessionId,
                        long blockId)
                 throws InvalidWorkerStateException,
                        BlockDoesNotExistException,
                        IOException

Description copied from interface: BlockWorker

Frees a block from Alluxio managed space.

Specified by:

removeBlock in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to be freed

Throws:

InvalidWorkerStateException - if blockId has not been committed

BlockDoesNotExistException - if block cannot be found

IOException

requestSpace

public void requestSpace(long sessionId,
                         long blockId,
                         long additionalBytes)
                  throws BlockDoesNotExistException,
                         WorkerOutOfSpaceException,
                         IOException

Description copied from interface: BlockWorker

Request an amount of space for a block in its storage directory. The block must be a temporary block.

Specified by:

requestSpace in interface BlockWorker

Parameters:

sessionId - the id of the client

blockId - the id of the block to allocate space to

additionalBytes - the amount of bytes to allocate

Throws:

BlockDoesNotExistException - if blockId can not be found, or some block in eviction plan cannot be found

WorkerOutOfSpaceException - if requested space can not be satisfied

IOException

unlockBlock

public void unlockBlock(long lockId)
                 throws BlockDoesNotExistException

Description copied from interface: BlockWorker

Releases the lock with the specified lock id.

Specified by:

unlockBlock in interface BlockWorker

Parameters:

lockId - the id of the lock to release

Throws:

BlockDoesNotExistException - if lock id cannot be found

unlockBlock

public boolean unlockBlock(long sessionId,
                           long blockId)

Description copied from interface: BlockWorker

Releases the lock with the specified session and block id.

Specified by:

unlockBlock in interface BlockWorker

Parameters:

sessionId - the session id

blockId - the block id

Returns:

false if it fails to unlock due to the lock is not found

sessionHeartbeat

public void sessionHeartbeat(long sessionId)

Description copied from interface: BlockWorker

Handles the heartbeat from a client.

Specified by:

sessionHeartbeat in interface BlockWorker

Parameters:

sessionId - the id of the client

updatePinList

public void updatePinList(Set<Long> pinnedInodes)

Description copied from interface: BlockWorker

Sets the pinlist for the underlying block store. Typically called by PinListSync.

Specified by:

updatePinList in interface BlockWorker

Parameters:

pinnedInodes - a set of pinned inodes

getFileInfo

public FileInfo getFileInfo(long fileId)
                     throws IOException

Description copied from interface: BlockWorker

Gets the file information.

Specified by:

getFileInfo in interface BlockWorker

Parameters:

fileId - the file id

Returns:

the file info

Throws:

IOException

openUfsBlock

public boolean openUfsBlock(long sessionId,
                            long blockId,
                            Protocol.OpenUfsBlockOptions options)
                     throws BlockAlreadyExistsException

Description copied from interface: BlockWorker

Opens a UFS block. It registers the block metadata information to the UFS block store. It throws an UfsBlockAccessTokenUnavailableException if the number of concurrent readers on this block exceeds a threshold.

Specified by:

openUfsBlock in interface BlockWorker

Parameters:

sessionId - the session ID

blockId - the block ID

options - the options

Returns:

whether the UFS block is successfully opened

Throws:

BlockAlreadyExistsException - if the UFS block already exists in the UnderFileSystemBlockStore

closeUfsBlock

public void closeUfsBlock(long sessionId,
                          long blockId)
                   throws BlockAlreadyExistsException,
                          IOException,
                          WorkerOutOfSpaceException

Description copied from interface: BlockWorker

Closes a UFS block for a client session. It also commits the block to Alluxio block store if the UFS block has been cached successfully.

Specified by:

closeUfsBlock in interface BlockWorker

Parameters:

sessionId - the session ID

blockId - the block ID

Throws:

BlockAlreadyExistsException - if it fails to commit the block to Alluxio block store because the block exists in the Alluxio block store

WorkerOutOfSpaceException - the the worker does not have enough space to commit the block

IOException

clearMetrics

public void clearMetrics()

Description copied from interface: BlockWorker

Clears the worker metrics.

Specified by:

clearMetrics in interface BlockWorker

cleanupSession

public void cleanupSession(long sessionId)

Description copied from interface: SessionCleanable

Cleans up the session.

Specified by:

cleanupSession in interface SessionCleanable

Parameters:

sessionId - the session id