/
LocalBlockInStream.java
117 lines (104 loc) · 3.79 KB
/
LocalBlockInStream.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
/*
* The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
* (the "License"). You may not use this work except in compliance with the License, which is
* available at www.apache.org/licenses/LICENSE-2.0
*
* This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
* either express or implied, as more fully set forth in the License.
*
* See the NOTICE file distributed with this work for information regarding copyright ownership.
*/
package alluxio.client.block;
import alluxio.client.ClientContext;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.util.io.BufferUtils;
import alluxio.wire.LockBlockResult;
import alluxio.wire.WorkerNetAddress;
import alluxio.worker.block.io.LocalFileBlockReader;
import com.google.common.io.Closer;
import java.io.IOException;
import java.nio.ByteBuffer;
import javax.annotation.concurrent.NotThreadSafe;
/**
* This class provides a streaming API to read a block in Alluxio. The data will be directly read
* from the local machine's storage.
*/
@NotThreadSafe
public final class LocalBlockInStream extends BufferedBlockInStream {
/** Helper to manage closables. */
private final Closer mCloser;
/** Client to communicate with the local worker. */
private final BlockWorkerClient mBlockWorkerClient;
/** The block store context which provides block worker clients. */
private final BlockStoreContext mContext;
/** The file reader to read a local block. */
private final LocalFileBlockReader mReader;
/**
* Creates a new local block input stream.
*
* @param blockId the block id
* @param blockSize the size of the block
* @param workerNetAddress the address of the local worker
* @param context the block store context to use for acquiring worker and master clients
* @throws IOException if I/O error occurs
*/
public LocalBlockInStream(long blockId, long blockSize, WorkerNetAddress workerNetAddress,
BlockStoreContext context) throws IOException {
super(blockId, blockSize);
mContext = context;
mCloser = Closer.create();
mBlockWorkerClient = mContext.acquireWorkerClient(workerNetAddress);
try {
LockBlockResult result = mBlockWorkerClient.lockBlock(blockId);
if (result == null) {
throw new IOException(ExceptionMessage.BLOCK_NOT_LOCALLY_AVAILABLE.getMessage(mBlockId));
}
mReader = new LocalFileBlockReader(result.getBlockPath());
mCloser.register(mReader);
} catch (IOException e) {
mContext.releaseWorkerClient(mBlockWorkerClient);
throw e;
}
}
@Override
public void close() throws IOException {
if (mClosed) {
return;
}
try {
if (mBlockIsRead) {
mBlockWorkerClient.accessBlock(mBlockId);
ClientContext.getClientMetrics().incBlocksReadLocal(1);
}
mBlockWorkerClient.unlockBlock(mBlockId);
} catch (AlluxioException e) {
throw new IOException(e);
} finally {
mContext.releaseWorkerClient(mBlockWorkerClient);
mCloser.close();
if (mBuffer != null && mBuffer.isDirect()) {
BufferUtils.cleanDirectBuffer(mBuffer);
}
}
mClosed = true;
}
@Override
protected void bufferedRead(int len) throws IOException {
if (mBuffer.isDirect()) { // Buffer may not be direct on initialization
BufferUtils.cleanDirectBuffer(mBuffer);
}
mBuffer = mReader.read(getPosition(), len);
}
@Override
public int directRead(byte[] b, int off, int len) throws IOException {
ByteBuffer buf = mReader.read(getPosition(), len);
buf.get(b, off, len);
BufferUtils.cleanDirectBuffer(buf);
return len;
}
@Override
protected void incrementBytesReadMetric(int bytes) {
ClientContext.getClientMetrics().incBytesReadLocal(bytes);
}
}