View Javadoc
1   /*
2   Copyright (c) 2012 James Ahlborn
3   
4   Licensed under the Apache License, Version 2.0 (the "License");
5   you may not use this file except in compliance with the License.
6   You may obtain a copy of the License at
7   
8       http://www.apache.org/licenses/LICENSE-2.0
9   
10  Unless required by applicable law or agreed to in writing, software
11  distributed under the License is distributed on an "AS IS" BASIS,
12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  See the License for the specific language governing permissions and
14  limitations under the License.
15  */
16  
17  package com.healthmarketscience.jackcess.util;
18  
19  import java.io.File;
20  import java.io.IOException;
21  import java.io.InputStream;
22  import java.io.OutputStream;
23  import java.io.RandomAccessFile;
24  import java.nio.ByteBuffer;
25  import java.nio.MappedByteBuffer;
26  import java.nio.channels.Channels;
27  import java.nio.channels.FileChannel;
28  import java.nio.channels.FileLock;
29  import java.nio.channels.NonWritableChannelException;
30  import java.nio.channels.ReadableByteChannel;
31  import java.nio.channels.WritableByteChannel;
32  
33  import com.healthmarketscience.jackcess.Database;
34  import com.healthmarketscience.jackcess.DatabaseBuilder;
35  import com.healthmarketscience.jackcess.impl.ByteUtil;
36  import com.healthmarketscience.jackcess.impl.DatabaseImpl;
37  
38  /**
39   * FileChannel implementation which maintains the entire "file" in memory.
40   * This enables working with a Database entirely in memory (for situations
41   * where disk usage may not be possible or desirable).  Obviously, this
42   * requires enough jvm heap space to fit the file data.  Use one of the
43   * {@code newChannel()} methods to construct an instance of this class.
44   * <p/>
45   * In order to use this class with a Database, you <i>must</i> use the {@link
46   * DatabaseBuilder} to open/create the Database instance, passing an instance
47   * of this class to the {@link DatabaseBuilder#setChannel} method.
48   * <p/>
49   * Implementation note: this class is optimized for use with {@link Database}.
50   * Therefore not all methods may be implemented and individual read/write
51   * operations are only supported within page boundaries.
52   *
53   * @author James Ahlborn
54   * @usage _advanced_class_
55   */
56  public class MemFileChannel extends FileChannel 
57  {
58    private static final byte[][] EMPTY_DATA = new byte[0][];
59  
60    // use largest possible Jet "page size" to ensure that reads/writes will
61    // always be within a single chunk
62    private static final int CHUNK_SIZE = 4096;
63    // this ensures that an "empty" mdb will fit in the initial chunk table
64    private static final int INIT_CHUNKS = 128;
65  
66    /** current read/write position */
67    private long _position;
68    /** current amount of actual data in the file */
69    private long _size;
70    /** chunks containing the file data.  the length of the chunk array is
71        always a power of 2 and the chunks are always CHUNK_SIZE. */ 
72    private byte[][] _data;
73  
74    private MemFileChannel() 
75    {
76      this(0L, 0L, EMPTY_DATA);
77    }
78  
79    private MemFileChannel(long position, long size, byte[][] data) {
80      _position = position;
81      _size = size;
82      _data = data;
83    }
84  
85    /**
86     * Creates a new read/write, empty MemFileChannel.
87     */
88    public static MemFileChannel newChannel() {
89      return new MemFileChannel();
90    }
91  
92    /**
93     * Creates a new read/write MemFileChannel containing the contents of the
94     * given File.  Note, modifications to the returned channel will <i>not</i>
95     * affect the original File source.
96     */
97    public static MemFileChannel newChannel(File file) throws IOException {
98      return newChannel(file, DatabaseImpl.RW_CHANNEL_MODE);
99    }
100 
101   /**
102    * Creates a new MemFileChannel containing the contents of the
103    * given File with the given mode (for mode details see
104    * {@link RandomAccessFile#RandomAccessFile(File,String)}).  Note,
105    * modifications to the returned channel will <i>not</i> affect the original
106    * File source.
107    */
108   public static MemFileChannel newChannel(File file, String mode) 
109     throws IOException 
110   {
111     FileChannel in = null;
112     try {
113       return newChannel(in = new RandomAccessFile(
114                             file, DatabaseImpl.RO_CHANNEL_MODE).getChannel(),
115                         mode);
116     } finally {
117       ByteUtil.closeQuietly(in);
118     }
119   }
120 
121   /**
122    * Creates a new read/write MemFileChannel containing the contents of the
123    * given InputStream.
124    */
125   public static MemFileChannel newChannel(InputStream in) throws IOException {
126     return newChannel(in, DatabaseImpl.RW_CHANNEL_MODE);
127   }
128 
129   /**
130    * Creates a new MemFileChannel containing the contents of the
131    * given InputStream with the given mode (for mode details see
132    * {@link RandomAccessFile#RandomAccessFile(File,String)}).
133    */
134   public static MemFileChannel newChannel(InputStream in, String mode) 
135     throws IOException 
136   {
137     return newChannel(Channels.newChannel(in), mode);
138   }
139 
140   /**
141    * Creates a new read/write MemFileChannel containing the contents of the
142    * given ReadableByteChannel.
143    */
144   public static MemFileChannel newChannel(ReadableByteChannel in) 
145     throws IOException
146   {
147     return newChannel(in, DatabaseImpl.RW_CHANNEL_MODE);
148   }
149 
150   /**
151    * Creates a new MemFileChannel containing the contents of the
152    * given ReadableByteChannel with the given mode (for mode details see
153    * {@link RandomAccessFile#RandomAccessFile(File,String)}).
154    */
155   public static MemFileChannel newChannel(ReadableByteChannel in, String mode) 
156     throws IOException
157   {
158     MemFileChannel channel = new MemFileChannel();
159     channel.transferFrom(in, 0L, Long.MAX_VALUE);
160     if(!mode.contains("w")) {
161       channel = new ReadOnlyChannel(channel);
162     }
163     return channel;
164   }
165 
166   @Override
167   public int read(ByteBuffer dst) throws IOException {
168     int bytesRead = read(dst, _position);
169     if(bytesRead > 0) {
170       _position += bytesRead;
171     }
172     return bytesRead;
173   }
174 
175   @Override
176   public int read(ByteBuffer dst, long position) throws IOException {
177     if(position >= _size) {
178       return -1;
179     }
180 
181     int numBytes = (int)Math.min(dst.remaining(), _size - position);
182     int rem = numBytes;
183 
184     while(rem > 0) {
185       byte[] chunk = _data[getChunkIndex(position)];
186       int chunkOffset = getChunkOffset(position);
187       int bytesRead = Math.min(rem, CHUNK_SIZE - chunkOffset);
188       dst.put(chunk, chunkOffset, bytesRead);
189       rem -= bytesRead;
190       position += bytesRead;
191     }
192 
193     return numBytes;
194   }
195 
196   @Override
197   public int write(ByteBuffer src) throws IOException {
198     int bytesWritten = write(src, _position);
199     _position += bytesWritten;
200     return bytesWritten;
201   }
202 
203   @Override
204   public int write(ByteBuffer src, long position) throws IOException {
205     int numBytes = src.remaining();
206     long newSize = position + numBytes;
207     ensureCapacity(newSize);
208 
209     int rem = numBytes;
210     while(rem > 0) {
211       byte[] chunk = _data[getChunkIndex(position)];
212       int chunkOffset = getChunkOffset(position);
213       int bytesWritten = Math.min(rem, CHUNK_SIZE - chunkOffset);
214       src.get(chunk, chunkOffset, bytesWritten);
215       rem -= bytesWritten;
216       position += bytesWritten;
217     }
218 
219     if(newSize > _size) {
220       _size = newSize;
221     }
222 
223     return numBytes;
224   }
225 
226   @Override
227   public long position() throws IOException {
228     return _position;
229   }
230 
231   @Override
232   public FileChannel position(long newPosition) throws IOException {
233     if(newPosition < 0L) {
234       throw new IllegalArgumentException("negative position");
235     }
236     _position = newPosition;
237     return this;
238   }
239 
240   @Override
241   public long size() throws IOException {
242     return _size;
243   }
244 
245   @Override
246   public FileChannel truncate(long newSize) throws IOException {
247     if(newSize < 0L) {
248       throw new IllegalArgumentException("negative size");
249     }
250     if(newSize < _size) {
251       // we'll optimize for memory over speed and aggressively free unused
252       // chunks
253       for(int i = getNumChunks(newSize); i < getNumChunks(_size); ++i) {
254         _data[i] = null;
255       }
256       _size = newSize;
257     }
258     _position = Math.min(newSize, _position);
259     return this;
260   }
261 
262   @Override
263   public void force(boolean metaData) throws IOException {
264     // nothing to do
265   }
266 
267   /**
268    * Convenience method for writing the entire contents of this channel to the
269    * given destination channel.
270    * @see #transferTo(long,long,WritableByteChannel)
271    */
272   public long transferTo(WritableByteChannel dst)
273     throws IOException
274   {
275     return transferTo(0L, _size, dst);
276   }
277 
278   @Override
279   public long transferTo(long position, long count, WritableByteChannel dst)
280     throws IOException
281   {
282     if(position >= _size) {
283       return 0L;
284     }
285     
286     count = Math.min(count, _size - position);
287 
288     int chunkIndex = getChunkIndex(position);
289     int chunkOffset = getChunkOffset(position);
290 
291     long numBytes = 0L;
292     while(count > 0L) {
293 
294       int chunkBytes = (int)Math.min(count, CHUNK_SIZE - chunkOffset);
295       ByteBuffer src = ByteBuffer.wrap(_data[chunkIndex], chunkOffset,
296                                        chunkBytes);
297 
298       do {
299         int bytesWritten = dst.write(src);
300         if(bytesWritten == 0L) {
301           // dst full
302           return numBytes;
303         }
304         numBytes += bytesWritten;
305         count -= bytesWritten;
306       } while(src.hasRemaining());
307       
308       ++chunkIndex;
309       chunkOffset = 0;
310     }
311 
312     return numBytes;
313   }
314 
315   /**
316    * Convenience method for writing the entire contents of this channel to the
317    * given destination stream.
318    * @see #transferTo(long,long,WritableByteChannel)
319    */
320   public long transferTo(OutputStream dst)
321     throws IOException
322   {
323     return transferTo(0L, _size, dst);
324   }
325 
326   /**
327    * Convenience method for writing the selected portion of this channel to
328    * the given destination stream.
329    * @see #transferTo(long,long,WritableByteChannel)
330    */
331   public long transferTo(long position, long count, OutputStream dst)
332     throws IOException
333   {
334     return transferTo(position, count, Channels.newChannel(dst));
335   }
336 
337   @Override
338   public long transferFrom(ReadableByteChannel src,
339                            long position, long count)
340     throws IOException
341   {
342     int chunkIndex = getChunkIndex(position);
343     int chunkOffset = getChunkOffset(position);
344 
345     long numBytes = 0L;
346     while(count > 0L) {
347 
348       ensureCapacity(position + numBytes + 1);
349 
350       int chunkBytes = (int)Math.min(count, CHUNK_SIZE - chunkOffset);
351       ByteBuffer dst = ByteBuffer.wrap(_data[chunkIndex], chunkOffset,
352                                        chunkBytes);
353       do {
354         int bytesRead = src.read(dst);
355         if(bytesRead <= 0) {
356           // src empty
357           return numBytes;
358         }
359         numBytes += bytesRead;
360         count -= bytesRead;
361         _size = Math.max(_size, position + numBytes);
362       } while(dst.hasRemaining());
363       
364       ++chunkIndex;
365       chunkOffset = 0;      
366     }
367     
368     return numBytes;
369   }
370 
371   @Override
372   protected void implCloseChannel() throws IOException {
373     // release data
374     _data = EMPTY_DATA;
375     _size = _position = 0L;
376   }
377 
378   private void ensureCapacity(long newSize)
379   {
380     if(newSize <= _size) {
381       // nothing to do
382       return;
383     }
384 
385     int newNumChunks = getNumChunks(newSize);
386     int numChunks = getNumChunks(_size);
387 
388     if(newNumChunks > _data.length) {
389 
390       // need to extend chunk array (use powers of 2)
391       int newDataLen = Math.max(_data.length, INIT_CHUNKS);
392       while(newDataLen < newNumChunks) {
393         newDataLen <<= 1;
394       }
395 
396       byte[][] newData = new byte[newDataLen][];
397 
398       // copy existing chunks
399       System.arraycopy(_data, 0, newData, 0, numChunks);
400 
401       _data = newData;
402     }
403 
404     // allocate new chunks
405     for(int i = numChunks; i < newNumChunks; ++i) {
406       _data[i] = new byte[CHUNK_SIZE];
407     }
408   }
409 
410   private static int getChunkIndex(long pos) {
411     return (int)(pos / CHUNK_SIZE);
412   }
413   
414   private static int getChunkOffset(long pos) {
415     return (int)(pos % CHUNK_SIZE);
416   }
417 
418   private static int getNumChunks(long size) {
419     return getChunkIndex(size + CHUNK_SIZE - 1);
420   }
421   
422   @Override
423   public long write(ByteBuffer[] srcs, int offset, int length)
424     throws IOException
425   {
426     long numBytes = 0L;
427     for(int i = offset; i < offset + length; ++i) {
428       numBytes += write(srcs[i]);
429     }
430     return numBytes;
431   }
432 
433   @Override
434   public long read(ByteBuffer[] dsts, int offset, int length)
435     throws IOException
436   {    
437     long numBytes = 0L;
438     for(int i = offset; i < offset + length; ++i) {
439       if(_position >= _size) {
440         return ((numBytes > 0L) ? numBytes : -1L);
441       }
442       numBytes += read(dsts[i]);
443     }
444     return numBytes;
445   }
446 
447   @Override
448   public MappedByteBuffer map(MapMode mode, long position, long size)
449     throws IOException
450   {
451     throw new UnsupportedOperationException();
452   }
453 
454   @Override
455   public FileLock lock(long position, long size, boolean shared)
456     throws IOException
457   {
458     throw new UnsupportedOperationException();
459   }
460 
461   @Override
462   public FileLock tryLock(long position, long size, boolean shared)
463     throws IOException
464   {
465     throw new UnsupportedOperationException();
466   }
467 
468   /**
469    * Subclass of MemFileChannel which is read-only.
470    */
471   private static final class ReadOnlyChannel extends MemFileChannel
472   {
473     private ReadOnlyChannel(MemFileChannel channel)
474     {
475       super(channel._position, channel._size, channel._data);
476     }
477     
478     @Override
479     public int write(ByteBuffer src, long position) throws IOException {
480       throw new NonWritableChannelException();
481     }
482 
483     @Override
484     public FileChannel truncate(long newSize) throws IOException {
485       throw new NonWritableChannelException();
486     }
487 
488     @Override
489     public long transferFrom(ReadableByteChannel src,
490                              long position, long count)
491       throws IOException
492     {
493       throw new NonWritableChannelException();
494     }    
495   }
496 }