View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18  package org.activeio.journal.active;
    19
    20  import java.io.File;
    21  import java.io.IOException;
    22  import java.io.InterruptedIOException;
    23  import java.lang.reflect.InvocationTargetException;
    24
    25  import org.activeio.Disposable;
    26  import org.activeio.Packet;
    27  import org.activeio.journal.InvalidRecordLocationException;
    28  import org.activeio.journal.Journal;
    29  import org.activeio.journal.JournalEventListener;
    30  import org.activeio.journal.RecordLocation;
    31  import org.activeio.packet.ByteArrayPacket;
    32  import org.activeio.packet.ByteBufferPacketPool;
    33
    34  import EDU.oswego.cs.dl.util.concurrent.FutureResult;
    35  import EDU.oswego.cs.dl.util.concurrent.QueuedExecutor;
    36  import EDU.oswego.cs.dl.util.concurrent.ThreadFactory;
    37
    38  /***
    39   * A high speed Journal implementation. Inspired by the ideas of the <a
    40   * href="http://howl.objectweb.org/">Howl </a> project but tailored to the needs
    41   * of ActiveMQ. <p/>This Journal provides the following features:
    42   * <ul>
    43   * <li>Conncurrent writes are batched into a single write/force done by a
    44   * background thread.</li>
    45   * <li>Uses preallocated logs to avoid disk fragmentation and performance
    46   * degregation.</li>
    47   * <li>The number and size of the preallocated logs are configurable.</li>
    48   * <li>Uses direct ByteBuffers to write data to log files.</li>
    49   * <li>Allows logs to grow in case of an overflow condition so that overflow
    50   * exceptions are not not thrown. Grown logs that are inactivated (due to a new
    51   * mark) are resized to thier original size.</li>
    52   * <li>No limit on the size of the record written to the journal</li>
    53   * <li>Should be possible to extend so that multiple physical disk are used
    54   * concurrently to increase throughput and decrease latency.</li>
    55   * </ul>
    56   * <p/>
    57   *
    58   * @version $Revision: 1.1 $
    59   */
    60  final public class JournalImpl implements Journal, Disposable {
    61
    62      public static final int DEFAULT_POOL_SIZE = Integer.parseInt(System.getProperty("org.activeio.journal.active.DefaultPoolSize", ""+(5)));
    63      public static final int DEFAULT_PACKET_SIZE = Integer.parseInt(System.getProperty("org.activeio.journal.active.DefaultPacketSize", ""+(1024*1024*4)));
    64
    65      static final private int OVERFLOW_RENOTIFICATION_DELAY = 500;
    66
    67      private boolean disposed = false;
    68
    69      // The id of the current log file that is being filled.
    70      private int appendLogFileId = 0;
    71
    72      // The offset in the current log file that is being filled.
    73      private int appendLogFileOffset = 0;
    74
    75      // Used to batch writes together.
    76      private BatchedWrite pendingBatchWrite;
    77
    78      private Location lastMarkedLocation;
    79      private LogFileManager file;
    80      private QueuedExecutor executor;
    81      private int rolloverFence;
    82      private JournalEventListener eventListener;
    83      private ByteBufferPacketPool packetPool;
    84      private long overflowNotificationTime = System.currentTimeMillis();
    85      private Packet markPacket = new ByteArrayPacket(new byte[Location.SERIALIZED_SIZE]);
    86
    87      public JournalImpl(File logDirectory) throws IOException {
    88          this(new LogFileManager(logDirectory));
    89      }
    90
    91      public JournalImpl(File logDirectory, int logFileCount, int logFileSize) throws IOException {
    92          this(new LogFileManager(logDirectory, logFileCount, logFileSize));
    93      }
    94
    95      public JournalImpl(LogFileManager logFile) {
    96          this.file = logFile;
    97          this.packetPool = new ByteBufferPacketPool(DEFAULT_POOL_SIZE, DEFAULT_PACKET_SIZE);
    98          this.executor = new QueuedExecutor();
    99          this.executor.setThreadFactory(new ThreadFactory() {
    100             public Thread newThread(Runnable runnable) {
    101                 Thread answer = new Thread(runnable, "Journal Writter");
    102                 answer.setPriority(Thread.MAX_PRIORITY);
    103                 answer.setDaemon(true);
    104                 return answer;
    105             }
    106         });
    107
    108         lastMarkedLocation = file.getLastMarkedRecordLocation();
    109         Location nextAppendLocation = file.getNextAppendLocation();
    110         appendLogFileId = nextAppendLocation.getLogFileId();
    111         appendLogFileOffset = nextAppendLocation.getLogFileOffset();
    112
    113         rolloverFence = (file.getInitialLogFileSize() / 10) * 9;
    114     }
    115
    116     public RecordLocation write(Packet data, boolean sync) throws IOException {
    117         return write(LogFileManager.DATA_RECORD_TYPE, data, sync, null);
    118     }
    119
    120     private Location write(byte recordType, Packet data, boolean sync, Location mark) throws IOException {
    121         try {
    122             Location location;
    123             BatchedWrite writeCommand;
    124
    125             Record record = new Record(recordType, data, mark);
    126
    127             // The following synchronized block is the bottle neck of the journal.  Make this
    128             // code faster and the journal should speed up.
    129             synchronized (this) {
    130                 if (disposed) {
    131                     throw new IOException("Journal has been closed.");
    132                 }
    133
    134                 // Create our record
    135                 location = new Location(appendLogFileId, appendLogFileOffset);
    136                 record.setLocation(location);
    137
    138                 // Piggy back the packet on the pending write batch.
    139                 writeCommand = addToPendingWriteBatch(record, mark, sync);
    140
    141                 // Update where the next record will land.
    142                 appendLogFileOffset += data.limit() + Record.RECORD_BASE_SIZE;
    143                 rolloverCheck();
    144             }
    145
    146             if (sync) {
    147                 writeCommand.waitForForce();
    148             }
    149
    150             return location;
    151         } catch (IOException e) {
    152             throw e;
    153         } catch (InterruptedException e) {
    154             throw (IOException) new InterruptedIOException().initCause(e);
    155         } catch (Throwable e) {
    156             throw (IOException) new IOException("Write failed: " + e).initCause(e);
    157         }
    158     }
    159
    160     /***
    161      * @param record
    162      * @return
    163      * @throws InterruptedException
    164      */
    165     private BatchedWrite addToPendingWriteBatch(Record record, Location mark, boolean force) throws InterruptedException {
    166
    167         // Load the write batch up with data from our record.
    168         // it may take more than one write batch if the record is large.
    169         BatchedWrite answer = null;
    170         while (record.hasRemaining()) {
    171
    172             // Do we need another BatchWrite?
    173             boolean queueTheWrite=false;
    174             if (pendingBatchWrite == null) {
    175                 pendingBatchWrite =  new BatchedWrite(packetPool.getPacket());
    176                 queueTheWrite = true;
    177             }
    178             answer = pendingBatchWrite;
    179
    180             // Can we continue to use the pendingBatchWrite?
    181             boolean full = !pendingBatchWrite.append(record, mark, force);
    182
    183             if( queueTheWrite ) {
    184                 final BatchedWrite queuedWrite = pendingBatchWrite;
    185                 executor.execute(new Runnable() {
    186                     public void run() {
    187                         try {
    188                             queuedWrite(queuedWrite);
    189                         } catch (InterruptedException e) {
    190                         }
    191                     }
    192                 });
    193             }
    194
    195             if( full )
    196                 pendingBatchWrite = null;
    197         }
    198         return answer;
    199
    200     }
    201
    202     /***
    203      * This is a blocking call
    204      *
    205      * @param write
    206      * @throws InterruptedException
    207      */
    208     private void queuedWrite(BatchedWrite write) throws InterruptedException {
    209
    210         // Stop other threads from appending more pendingBatchWrite.
    211         write.flip();
    212
    213         // Do the write.
    214         try {
    215             file.append(write);
    216             write.forced();
    217         } catch (Throwable e) {
    218             write.writeFailed(e);
    219         } finally {
    220             write.getPacket().dispose();
    221         }
    222     }
    223
    224     /***
    225      *
    226      */
    227     private void rolloverCheck() throws IOException {
    228
    229         // See if we need to issue an overflow notification.
    230         if (eventListener != null && file.isPastHalfActive()
    231                 && overflowNotificationTime + OVERFLOW_RENOTIFICATION_DELAY < System.currentTimeMillis()) {
    232
    233             // We need to send an overflow notification to free up
    234             // some logFiles.
    235             Location safeSpot = file.getFirstRecordLocationOfSecondActiveLogFile();
    236             eventListener.overflowNotification(safeSpot);
    237             overflowNotificationTime = System.currentTimeMillis();
    238         }
    239
    240         // Is it time to roll over?
    241         if (appendLogFileOffset > rolloverFence ) {
    242
    243             // Can we roll over?
    244             if ( !file.canActivateNextLogFile() ) {
    245                 // don't delay the next overflow notification.
    246                 overflowNotificationTime -= OVERFLOW_RENOTIFICATION_DELAY;
    247
    248             } else {
    249
    250                 final FutureResult result = new FutureResult();
    251                 try {
    252                     executor.execute(new Runnable() {
    253                         public void run() {
    254                             try {
    255                                 result.set(queuedActivateNextLogFile());
    256                             } catch (Throwable e) {
    257                                 result.setException(e);
    258                             }
    259                         }
    260                     });
    261
    262                     Location location = (Location) result.get();
    263                     appendLogFileId = location.getLogFileId();
    264                     appendLogFileOffset = location.getLogFileOffset();
    265
    266                 } catch (InterruptedException e) {
    267                     throw (IOException) new IOException("Interrupted.").initCause(e);
    268                 } catch (InvocationTargetException e) {
    269                     if (e.getTargetException() instanceof IOException)
    270                         throw (IOException) new IOException(e.getTargetException().getMessage()).initCause(e
    271                                 .getTargetException());
    272                     throw (IOException) new IOException("Unexpected Exception: ").initCause(e.getTargetException());
    273                 }
    274             }
    275         }
    276     }
    277
    278     /***
    279      * This is a blocking call
    280      */
    281     private Location queuedActivateNextLogFile() throws IOException {
    282         file.activateNextLogFile();
    283         return file.getNextAppendLocation();
    284     }
    285
    286
    287
    288     /***
    289      * @param recordLocator
    290      * @param force
    291      * @return
    292      * @throws InvalidRecordLocationException
    293      * @throws IOException
    294      * @throws InterruptedException
    295      */
    296     synchronized public void setMark(RecordLocation l, boolean force) throws InvalidRecordLocationException,
    297             IOException {
    298
    299         Location location = (Location) l;
    300         if (location == null)
    301             throw new InvalidRecordLocationException("The location cannot be null.");
    302         if (lastMarkedLocation != null && location.compareTo(lastMarkedLocation) < 0)
    303             throw new InvalidRecordLocationException("The location is less than the last mark.");
    304
    305         markPacket.clear();
    306         location.writeToPacket(markPacket);
    307         markPacket.flip();
    308         write(LogFileManager.MARK_RECORD_TYPE, markPacket, force, location);
    309
    310         lastMarkedLocation = location;
    311     }
    312
    313     /***
    314      * @return
    315      */
    316     public RecordLocation getMark() {
    317         return lastMarkedLocation;
    318     }
    319
    320     /***
    321      * @param lastLocation
    322      * @return
    323      * @throws IOException
    324      * @throws InvalidRecordLocationException
    325      */
    326     public RecordLocation getNextRecordLocation(final RecordLocation lastLocation) throws IOException,
    327             InvalidRecordLocationException {
    328
    329         if (lastLocation == null) {
    330             if (lastMarkedLocation != null) {
    331                 return lastMarkedLocation;
    332             } else {
    333                 return file.getFirstActiveLogLocation();
    334             }
    335         }
    336
    337         // Run this in the queued executor thread.
    338         final FutureResult result = new FutureResult();
    339         try {
    340             executor.execute(new Runnable() {
    341                 public void run() {
    342                     try {
    343                         result.set(queuedGetNextRecordLocation((Location) lastLocation));
    344                     } catch (Throwable e) {
    345                         result.setException(e);
    346                     }
    347                 }
    348             });
    349             return (Location) result.get();
    350         } catch (InterruptedException e) {
    351             throw (IOException) new IOException("Interrupted.").initCause(e);
    352         } catch (InvocationTargetException e) {
    353             return (RecordLocation) unwrapException(e);
    354
    355         }
    356     }
    357
    358     private Object unwrapException(InvocationTargetException e) throws InvalidRecordLocationException, IOException {
    359         if (e.getTargetException() instanceof InvalidRecordLocationException)
    360             throw new InvalidRecordLocationException(e.getTargetException().getMessage(), e.getTargetException());
    361         if (e.getTargetException() instanceof IOException)
    362             throw (IOException) new IOException(e.getTargetException().getMessage()).initCause(e
    363                     .getTargetException());
    364         throw (IOException) new IOException("Unexpected Exception: ").initCause(e.getTargetException());
    365     }
    366
    367     private Location queuedGetNextRecordLocation(Location location) throws IOException, InvalidRecordLocationException {
    368         return file.getNextDataRecordLocation(location);
    369     }
    370
    371     /***
    372      * @param location
    373      * @return
    374      * @throws InvalidRecordLocationException
    375      * @throws IOException
    376      */
    377     public Packet read(final RecordLocation l) throws IOException, InvalidRecordLocationException {
    378         final Location location = (Location) l;
    379         // Run this in the queued executor thread.
    380         final FutureResult result = new FutureResult();
    381         try {
    382             executor.execute(new Runnable() {
    383                 public void run() {
    384                     try {
    385                         result.set(file.readPacket(location));
    386                     } catch (Throwable e) {
    387                         result.setException(e);
    388                     }
    389                 }
    390             });
    391             return (Packet) result.get();
    392         } catch (InterruptedException e) {
    393             throw (IOException) new IOException("Interrupted.").initCause(e);
    394         } catch (InvocationTargetException e) {
    395             if (e.getTargetException() instanceof InvalidRecordLocationException)
    396                 throw new InvalidRecordLocationException(e.getTargetException().getMessage(), e.getTargetException());
    397             if (e.getTargetException() instanceof IOException)
    398                 throw (IOException) new IOException(e.getTargetException().getMessage()).initCause(e
    399                         .getTargetException());
    400             throw (IOException) new IOException("Unexpected Exception: ").initCause(e.getTargetException());
    401         }
    402     }
    403
    404     public void setJournalEventListener(JournalEventListener eventListener) {
    405         this.eventListener = eventListener;
    406     }
    407
    408     /***
    409      * @deprecated @see #dispose()
    410      */
    411     public void close() throws IOException {
    412     	dispose();
    413     }
    414
    415     /***
    416      */
    417     public void dispose() {
    418         if (disposed)
    419             return;
    420         disposed=true;
    421         executor.shutdownAfterProcessingCurrentlyQueuedTasks();
    422         file.dispose();
    423     }
    424
    425     /***
    426      * @return
    427      */
    428     public File getLogDirectory() {
    429         return file.getLogDirectory();
    430     }
    431
    432     public int getInitialLogFileSize() {
    433         return file.getInitialLogFileSize();
    434     }
    435
    436     public String toString() {
    437         return "Active Journal: using "+file.getOnlineLogFileCount()+" x " + (file.getInitialLogFileSize()/(1024*1024f)) + " Megs at: " + getLogDirectory();
    438     }
    439
    440 }