File D:\maven-qcr-plugin-testsunitaires\qcr4java\h2-1.3.170-sources\target\site\qc\lecode\mvstore\StreamStore.java.html

1 /*
2 * Copyright 2004-2011 H2 Group. Multiple-Licensed under the H2 License,
3 * Version 1.0, and under the Eclipse Public License, Version 1.0
4 * (http://h2database.com/html/license.html).
5 * Initial Developer: H2 Group
6 */
7 package org.h2.mvstore;
8
9 import java.io.ByteArrayInputStream;
10 import java.io.ByteArrayOutputStream;
11 import java.io.IOException;
12 import java.io.InputStream;
13 import java.nio.ByteBuffer;
14 import java.util.Map;
15 import java.util.concurrent.atomic.AtomicLong;
16 import org.h2.util.IOUtils;
17
18 /**
19 * A facility to store streams in a map. Streams are split into blocks, which
20 * are stored in a map. Very small streams are inlined in the stream id.
21 * <p>
22 * The key of the map is a long (incremented for each stored block). The default
23 * initial value is 0. Before storing blocks into the map, the stream store
24 * checks if there is already a block with the next key, and if necessary
25 * searches the next free entry using a binary search (0 to Long.MAX_VALUE).
26 * <p>
27 * The format of the binary id is: An empty id represents 0 bytes of data.
28 * In-place data is encoded as 0, the size (a variable size int), then the data.
29 * A stored block is encoded as 1, the length of the block (a variable size
30 * int), then the key (a variable size long). Multiple ids can be concatenated
31 * to concatenate the data. If the id is large, it is stored itself, which is
32 * encoded as 2, the total length (a variable size long), and the key of the
33 * block that contains the id (a variable size long).
34 */
35 public class StreamStore {
36
37 private final Map<Long, byte[]> map;
38 private int minBlockSize = 256;
39 private int maxBlockSize = 256 * 1024;
40 private final AtomicLong nextKey = new AtomicLong();
41
42 /**
43 * Create a stream store instance.
44 *
45 * @param map the map to store blocks of data
46 */
47 public StreamStore(Map<Long, byte[]> map) {
48 this.map = map;
49 }
50
51 public Map<Long, byte[]> getMap() {
52 return map;
53 }
54
55 public void setNextKey(long nextKey) {
56 this.nextKey.set(nextKey);
57 }
58
59 public long getNextKey() {
60 return nextKey.get();
61 }
62
63 public void setMinBlockSize(int minBlockSize) {
64 this.minBlockSize = minBlockSize;
65 }
66
67 public int getMinBlockSize() {
68 return minBlockSize;
69 }
70
71 public void setMaxBlockSize(int maxBlockSize) {
72 this.maxBlockSize = maxBlockSize;
73 }
74
75 public long getMaxBlockSize() {
76 return maxBlockSize;
77 }
78
79 /**
80 * Store the stream, and return the id.
81 *
82 * @param in the stream
83 * @return the id (potentially an empty array)
84 */
85 public byte[] put(InputStream in) throws IOException {
86 ByteArrayOutputStream id = new ByteArrayOutputStream();
87 int level = 0;
88 while (true) {
89 if (put(id, in, level)) {
90 break;
91 }
92 if (id.size() > maxBlockSize / 2) {
93 id = putIndirectId(id);
94 level++;
95 }
96 }
97 if (id.size() > minBlockSize * 2) {
98 id = putIndirectId(id);
99 }
100 return id.toByteArray();
101 }
102
103 private boolean put(ByteArrayOutputStream id, InputStream in, int level) throws IOException {
104 if (level > 0) {
105 ByteArrayOutputStream id2 = new ByteArrayOutputStream();
106 while (true) {
107 boolean eof = put(id2, in, level - 1);
108 if (id2.size() > maxBlockSize / 2) {
109 id2 = putIndirectId(id2);
110 id2.writeTo(id);
111 return eof;
112 } else if (eof) {
113 id2.writeTo(id);
114 return true;
115 }
116 }
117 }
118 ByteArrayOutputStream buffer = new ByteArrayOutputStream();
119 int len = (int) IOUtils.copy(in, buffer, maxBlockSize);
120 if (len == 0) {
121 return true;
122 }
123 boolean eof = len < maxBlockSize;
124 byte[] data = buffer.toByteArray();
125 if (len < minBlockSize) {
126 id.write(0);
127 DataUtils.writeVarInt(id, len);
128 id.write(data);
129 } else {
130 id.write(1);
131 DataUtils.writeVarInt(id, len);
132 DataUtils.writeVarLong(id, writeBlock(data));
133 }
134 return eof;
135 }
136
137 private ByteArrayOutputStream putIndirectId(ByteArrayOutputStream id) throws IOException {
138 byte[] data = id.toByteArray();
139 id = new ByteArrayOutputStream();
140 id.write(2);
141 DataUtils.writeVarLong(id, length(data));
142 DataUtils.writeVarLong(id, writeBlock(data));
143 return id;
144 }
145
146 private long writeBlock(byte[] data) {
147 long key = getAndIncrementNextKey();
148 map.put(key, data);
149 return key;
150 }
151
152 private long getAndIncrementNextKey() {
153 long key = nextKey.getAndIncrement();
154 if (!map.containsKey(key)) {
155 return key;
156 }
157 // search the next free id using binary search
158 synchronized (this) {
159 long low = key, high = Long.MAX_VALUE;
160 while (low < high) {
161 long x = (low + high) >>> 1;
162 if (map.containsKey(x)) {
163 low = x + 1;
164 } else {
165 high = x;
166 }
167 }
168 key = low;
169 nextKey.set(key + 1);
170 return key;
171 }
172 }
173
174 /**
175 * Remove all stored blocks for the given id.
176 *
177 * @param id the id
178 */
179 public void remove(byte[] id) {
180 ByteBuffer idBuffer = ByteBuffer.wrap(id);
181 while (idBuffer.hasRemaining()) {
182 switch (idBuffer.get()) {
183 case 0:
184 int len = DataUtils.readVarInt(idBuffer);
185 idBuffer.position(idBuffer.position() + len);
186 break;
187 case 1:
188 DataUtils.readVarInt(idBuffer);
189 long k = DataUtils.readVarLong(idBuffer);
190 map.remove(k);
191 break;
192 case 2:
193 DataUtils.readVarLong(idBuffer);
194 long k2 = DataUtils.readVarLong(idBuffer);
195 // recurse
196 remove(map.get(k2));
197 map.remove(k2);
198 break;
199 default:
200 throw new IllegalArgumentException("Unsupported id");
201 }
202 }
203 }
204
205 /**
206 * Calculate the number of data bytes for the given id. As the length is
207 * encoded in the id, this operation does not cause any reads in the map.
208 *
209 * @param id the id
210 * @return the length
211 */
212 public long length(byte[] id) {
213 ByteBuffer idBuffer = ByteBuffer.wrap(id);
214 long length = 0;
215 while (idBuffer.hasRemaining()) {
216 switch (idBuffer.get()) {
217 case 0:
218 int len = DataUtils.readVarInt(idBuffer);
219 idBuffer.position(idBuffer.position() + len);
220 length += len;
221 break;
222 case 1:
223 length += DataUtils.readVarInt(idBuffer);
224 DataUtils.readVarLong(idBuffer);
225 break;
226 case 2:
227 length += DataUtils.readVarLong(idBuffer);
228 DataUtils.readVarLong(idBuffer);
229 break;
230 default:
231 throw new IllegalArgumentException("Unsupported id");
232 }
233 }
234 return length;
235 }
236
237 /**
238 * Check whether the id itself contains all the data. This operation does
239 * not cause any reads in the map.
240 *
241 * @param id the id
242 * @return if the id contains the data
243 */
244 public boolean isInPlace(byte[] id) {
245 ByteBuffer idBuffer = ByteBuffer.wrap(id);
246 while (idBuffer.hasRemaining()) {
247 if (idBuffer.get() != 0) {
248 return false;
249 }
250 int len = DataUtils.readVarInt(idBuffer);
251 idBuffer.position(idBuffer.position() + len);
252 }
253 return true;
254 }
255
256 /**
257 * Open an input stream to read data.
258 *
259 * @param id the id
260 * @return the stream
261 */
262 public InputStream get(byte[] id) {
263 return new Stream(this, id);
264 }
265
266 /**
267 * Get the block.
268 *
269 * @param key the key
270 * @return the block
271 */
272 byte[] getBlock(long key) {
273 return map.get(key);
274 }
275
276 /**
277 * A stream backed by a map.
278 */
279 static class Stream extends InputStream {
280
281 private final StreamStore store;
282 private byte[] oneByteBuffer;
283 private ByteBuffer idBuffer;
284 private ByteArrayInputStream buffer;
285 private long skip;
286 private final long length;
287 private long pos;
288
289 Stream(StreamStore store, byte[] id) {
290 this.store = store;
291 this.length = store.length(id);
292 this.idBuffer = ByteBuffer.wrap(id);
293 }
294
295 @Override
296 public int read() {
297 byte[] buffer = oneByteBuffer;
298 if (buffer == null) {
299 buffer = oneByteBuffer = new byte[1];
300 }
301 int len = read(buffer, 0, 1);
302 return len == -1 ? -1 : (buffer[0] & 255);
303 }
304
305 @Override
306 public long skip(long n) {
307 n = Math.min(length - pos, n);
308 if (n == 0) {
309 return 0;
310 }
311 if (buffer != null) {
312 long s = buffer.skip(n);
313 if (s > 0) {
314 n = s;
315 } else {
316 buffer = null;
317 skip += n;
318 }
319 } else {
320 skip += n;
321 }
322 pos += n;
323 return n;
324 }
325
326 @Override
327 public void close() {
328 buffer = null;
329 idBuffer.position(idBuffer.limit());
330 pos = length;
331 }
332
333 @Override
334 public int read(byte[] b, int off, int len) {
335 while (true) {
336 if (buffer == null) {
337 buffer = nextBuffer();
338 if (buffer == null) {
339 return -1;
340 }
341 }
342 int result = buffer.read(b, off, len);
343 if (result > 0) {
344 pos += result;
345 return result;
346 }
347 buffer = null;
348 }
349 }
350
351 private ByteArrayInputStream nextBuffer() {
352 while (idBuffer.hasRemaining()) {
353 switch (idBuffer.get()) {
354 case 0: {
355 int len = DataUtils.readVarInt(idBuffer);
356 if (skip >= len) {
357 skip -= len;
358 idBuffer.position(idBuffer.position() + len);
359 continue;
360 }
361 int p = (int) (idBuffer.position() + skip);
362 int l = (int) (len - skip);
363 idBuffer.position(p + l);
364 return new ByteArrayInputStream(idBuffer.array(), p, l);
365 }
366 case 1: {
367 int len = DataUtils.readVarInt(idBuffer);
368 long key = DataUtils.readVarLong(idBuffer);
369 if (skip >= len) {
370 skip -= len;
371 continue;
372 }
373 byte[] data = store.getBlock(key);
374 int s = (int) skip;
375 skip = 0;
376 return new ByteArrayInputStream(data, s, data.length - s);
377 }
378 case 2: {
379 long len = DataUtils.readVarInt(idBuffer);
380 long key = DataUtils.readVarLong(idBuffer);
381 if (skip >= len) {
382 skip -= len;
383 continue;
384 }
385 byte[] k = store.getBlock(key);
386 ByteBuffer newBuffer = ByteBuffer.allocate(k.length + idBuffer.limit() - idBuffer.position());
387 newBuffer.put(k);
388 newBuffer.put(idBuffer);
389 newBuffer.flip();
390 idBuffer = newBuffer;
391 return nextBuffer();
392 }
393 default:
394 throw new IllegalArgumentException("Unsupported id");
395 }
396 }
397 return null;
398 }
399
400 }
401
402 }
403