FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public AutoCloseableLock acquireDatasetReadLock() { return datasetReadLock.acquire(); } FsDatasetImpl(DataNode datanode, DataStorage storage, Configuration conf
      ); @Override // FsDatasetSpi FsVolumeReferences getFsVolumeReferences(); @Override DatanodeStorage getStorage(final String storageUuid); @Override // FsDatasetSpi StorageReport[] getStorageReports(String bpid); @Override FsVolumeImpl getVolume(final ExtendedBlock b); @Override // FsDatasetSpi Block getStoredBlock(String bpid, long blkid); @Override Set<? extends Replica> deepCopyReplica(String bpid); @Override // FsDatasetSpi LengthInputStream getMetaDataInputStream(ExtendedBlock b); @Override AutoCloseableLock acquireDatasetLock(); @Override AutoCloseableLock acquireDatasetReadLock(); @VisibleForTesting FsVolumeImpl createFsVolume(String storageUuid,
      Storage.StorageDirectory sd,
      final StorageLocation location); @Override void addVolume(final StorageLocation location,
      final List<NamespaceInfo> nsInfos); @Override void removeVolumes(
      final Collection<StorageLocation> storageLocsToRemove,
      boolean clearFailure); @Override // FSDatasetMBean long getDfsUsed(); @Override // FSDatasetMBean long getBlockPoolUsed(String bpid); @Override // FsDatasetSpi boolean hasEnoughResource(); @Override // FSDatasetMBean long getCapacity(); @Override // FSDatasetMBean long getRemaining(); @Override // FSDatasetMBean int getNumFailedVolumes(); @Override // FSDatasetMBean String[] getFailedStorageLocations(); @Override // FSDatasetMBean long getLastVolumeFailureDate(); @Override // FSDatasetMBean long getEstimatedCapacityLostTotal(); @Override // FsDatasetSpi VolumeFailureSummary getVolumeFailureSummary(); @Override // FSDatasetMBean long getCacheUsed(); @Override // FSDatasetMBean long getCacheCapacity(); @Override // FSDatasetMBean long getNumBlocksFailedToCache(); @Override // FSDatasetMBean long getNumBlocksFailedToUncache(); @Override void getMetrics(MetricsCollector collector, boolean all); @Override // FSDatasetMBean long getNumBlocksCached(); @Override // FsDatasetSpi long getLength(ExtendedBlock b); @Override // FsDatasetSpi InputStream getBlockInputStream(ExtendedBlock b,
      long seekOffset); @Override // FsDatasetSpi ReplicaInputStreams getTmpInputStreams(ExtendedBlock b,
      long blkOffset, long metaOffset); @Override ReplicaInfo moveBlockAcrossStorage(ExtendedBlock block,
      StorageType targetStorageType, String targetStorageId); @Override ReplicaInfo moveBlockAcrossVolumes(ExtendedBlock block, FsVolumeSpi
      destination); @Override // FsDatasetSpi ReplicaHandler append(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi ReplicaHandler recoverAppend(
      ExtendedBlock b, long newGS, long expectedBlockLen); @Override // FsDatasetSpi Replica recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi ReplicaHandler createRbw(
      StorageType storageType, String storageId, ExtendedBlock b,
      boolean allowLazyPersist); @Override // FsDatasetSpi ReplicaHandler recoverRbw(
      ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi ReplicaHandler createTemporary(StorageType storageType,
      String storageId, ExtendedBlock b, boolean isTransfer); @Override // FsDatasetSpi void adjustCrcChannelPosition(ExtendedBlock b, ReplicaOutputStreams streams, 
      int checksumSize); @Override // FsDatasetSpi void finalizeBlock(ExtendedBlock b, boolean fsyncDir); @Override // FsDatasetSpi void unfinalizeBlock(ExtendedBlock b); @Override // FsDatasetSpi List<Long> getCacheReport(String bpid); @Override Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid); @Override List<ReplicaInfo> getSortedFinalizedBlocks(String bpid); @Override // FsDatasetSpi void checkBlock(ExtendedBlock b, long minLength, ReplicaState state); @Override // FsDatasetSpi boolean isValidBlock(ExtendedBlock b); @Override // {@link FsDatasetSpi} boolean isValidRbw(final ExtendedBlock b); @Override // FsDatasetSpi void invalidate(String bpid, Block invalidBlks[]); void invalidate(String bpid, ReplicaInfo block); @Override // FsDatasetSpi void cache(String bpid, long[] blockIds); @Override // FsDatasetSpi void uncache(String bpid, long[] blockIds); @Override boolean isCached(String bpid, long blockId); @Override // FsDatasetSpi boolean contains(final ExtendedBlock block); @Override // FsDatasetSpi void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes); @Override // FsDatasetSpi String toString(); @Override // FsDatasetSpi void shutdown(); @Override // FSDatasetMBean String getStorageInfo(); @Override void checkAndUpdate(String bpid, ScanInfo scanInfo); @Override // FsDatasetSpi @Deprecated ReplicaInfo getReplica(String bpid, long blockId); @Override String getReplicaString(String bpid, long blockId); @Override // FsDatasetSpi ReplicaRecoveryInfo initReplicaRecovery(RecoveringBlock rBlock); @Override // FsDatasetSpi Replica updateReplicaUnderRecovery(
                                    final ExtendedBlock oldBlock,
                                    final long recoveryId,
                                    final long newBlockId,
                                    final long newlength); @Override // FsDatasetSpi long getReplicaVisibleLength(final ExtendedBlock block); @Override void addBlockPool(String bpid, Configuration conf); @VisibleForTesting static void setBlockPoolId(String bpid); @Override void shutdownBlockPool(String bpid); @Override Map<String, Object> getVolumeInfoMap(); @Override //FsDatasetSpi void deleteBlockPool(String bpid, boolean force); @Override // FsDatasetSpi BlockLocalPathInfo getBlockLocalPathInfo(ExtendedBlock block); @Override void enableTrash(String bpid); @Override void clearTrash(String bpid); @Override boolean trashEnabled(String bpid); @Override void setRollingUpgradeMarker(String bpid); @Override void clearRollingUpgradeMarker(String bpid); @Override void onCompleteLazyPersist(String bpId, long blockId,
      long creationTime, File[] savedFiles, FsVolumeImpl targetVolume); @Override void onFailLazyPersist(String bpId, long blockId); @Override void submitBackgroundSyncFileRangeRequest(ExtendedBlock block,
      ReplicaOutputStreams outs, long offset, long nbytes, int flags); @Override void setPinning(ExtendedBlock block); @Override boolean getPinning(ExtendedBlock block); @Override boolean isDeletingBlock(String bpid, long blockId); void removeDeletedBlocks(String bpid, Set<Long> blockIds); @VisibleForTesting void evictLazyPersistBlocks(long bytesNeeded); @VisibleForTesting int getNonPersistentReplicas(); @VisibleForTesting void setTimer(Timer newTimer); int getBPServiceCount(); int getVolumeCount();  }
@Test public void testReadLockEnabledByDefault() throws IOException, InterruptedException { final FsDatasetSpi ds = dataset; AtomicBoolean accessed = new AtomicBoolean(false); CountDownLatch latch = new CountDownLatch(1); CountDownLatch waiterLatch = new CountDownLatch(1); Thread holder = new Thread() { public void run() { try (AutoCloseableLock l = ds.acquireDatasetReadLock()) { latch.countDown(); sleep(10000); } catch (Exception e) { } } }; Thread waiter = new Thread() { public void run() { try (AutoCloseableLock l = ds.acquireDatasetReadLock()) { waiterLatch.countDown(); accessed.getAndSet(true); } catch (Exception e) { } } }; holder.start(); latch.await(); waiter.start(); waiterLatch.await(); assertEquals(true, accessed.get()); holder.interrupt(); holder.join(); waiter.join(); }