1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.util;
20
21 import java.nio.ByteBuffer;
22 import java.util.concurrent.locks.Lock;
23 import java.util.concurrent.locks.ReentrantLock;
24
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.classification.InterfaceAudience;
28 import org.apache.hadoop.util.StringUtils;
29
30
31
32
33
34
35 @InterfaceAudience.Private
36 public final class ByteBufferArray {
37 static final Log LOG = LogFactory.getLog(ByteBufferArray.class);
38
39 static final int DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024;
40 private ByteBuffer buffers[];
41 private Lock locks[];
42 private int bufferSize;
43 private int bufferCount;
44
45
46
47
48
49
50
51
52 public ByteBufferArray(long capacity, boolean directByteBuffer) {
53 this.bufferSize = DEFAULT_BUFFER_SIZE;
54 if (this.bufferSize > (capacity / 16))
55 this.bufferSize = (int) roundUp(capacity / 16, 32768);
56 this.bufferCount = (int) (roundUp(capacity, bufferSize) / bufferSize);
57 LOG.info("Allocating buffers total=" + StringUtils.byteDesc(capacity)
58 + " , sizePerBuffer=" + StringUtils.byteDesc(bufferSize) + ", count="
59 + bufferCount);
60 buffers = new ByteBuffer[bufferCount + 1];
61 locks = new Lock[bufferCount + 1];
62 for (int i = 0; i <= bufferCount; i++) {
63 locks[i] = new ReentrantLock();
64 if (i < bufferCount) {
65 buffers[i] = directByteBuffer ? ByteBuffer.allocateDirect(bufferSize)
66 : ByteBuffer.allocate(bufferSize);
67 } else {
68 buffers[i] = ByteBuffer.allocate(0);
69 }
70
71 }
72 }
73
74 private long roundUp(long n, long to) {
75 return ((n + to - 1) / to) * to;
76 }
77
78
79
80
81
82
83
84
85 public int getMultiple(long start, int len, byte[] dstArray) {
86 return getMultiple(start, len, dstArray, 0);
87 }
88
89
90
91
92
93
94
95
96
97
98 public int getMultiple(long start, int len, byte[] dstArray, int dstOffset) {
99 multiple(start, len, dstArray, dstOffset, new Visitor() {
100 public void visit(ByteBuffer bb, byte[] array, int arrayIdx, int len) {
101 bb.get(array, arrayIdx, len);
102 }
103 });
104 return len;
105 }
106
107
108
109
110
111
112
113 public void putMultiple(long start, int len, byte[] srcArray) {
114 putMultiple(start, len, srcArray, 0);
115 }
116
117
118
119
120
121
122
123
124
125 public void putMultiple(long start, int len, byte[] srcArray, int srcOffset) {
126 multiple(start, len, srcArray, srcOffset, new Visitor() {
127 public void visit(ByteBuffer bb, byte[] array, int arrayIdx, int len) {
128 bb.put(array, arrayIdx, len);
129 }
130 });
131 }
132
133 private interface Visitor {
134
135
136
137
138
139
140
141
142
143 void visit(ByteBuffer bb, byte[] array, int arrayOffset, int len);
144 }
145
146
147
148
149
150
151
152
153
154
155
156
157 void multiple(long start, int len, byte[] array, int arrayOffset, Visitor visitor) {
158 assert len >= 0;
159 long end = start + len;
160 int startBuffer = (int) (start / bufferSize), startOffset = (int) (start % bufferSize);
161 int endBuffer = (int) (end / bufferSize), endOffset = (int) (end % bufferSize);
162 assert array.length >= len + arrayOffset;
163 assert startBuffer >= 0 && startBuffer < bufferCount;
164 assert endBuffer >= 0 && endBuffer < bufferCount
165 || (endBuffer == bufferCount && endOffset == 0);
166 if (startBuffer >= locks.length || startBuffer < 0) {
167 String msg = "Failed multiple, start=" + start + ",startBuffer="
168 + startBuffer + ",bufferSize=" + bufferSize;
169 LOG.error(msg);
170 throw new RuntimeException(msg);
171 }
172 int srcIndex = 0, cnt = -1;
173 for (int i = startBuffer; i <= endBuffer; ++i) {
174 Lock lock = locks[i];
175 lock.lock();
176 try {
177 ByteBuffer bb = buffers[i];
178 if (i == startBuffer) {
179 cnt = bufferSize - startOffset;
180 if (cnt > len) cnt = len;
181 bb.limit(startOffset + cnt).position(
182 startOffset );
183 } else if (i == endBuffer) {
184 cnt = endOffset;
185 bb.limit(cnt).position(0);
186 } else {
187 cnt = bufferSize ;
188 bb.limit(cnt).position(0);
189 }
190 visitor.visit(bb, array, srcIndex + arrayOffset, cnt);
191 srcIndex += cnt;
192 } finally {
193 lock.unlock();
194 }
195 }
196 assert srcIndex == len;
197 }
198 }