    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;
    19
    20  import java.io.File;
    21  import java.io.FileOutputStream;
    22  import java.io.IOException;
    23  import java.io.PrintWriter;
    24  import java.util.Random;
    25
    26  import org.activeio.packet.ByteArrayPacket;
    27
    28  /***
    29   * Provides the base class uses to run performance tests against a Journal.
    30   * Should be subclassed to customize for specific journal implementation.
    31   *
    32   * @version $Revision: 1.1 $
    33   */
    34  abstract public class JournalPerfToolSupport implements JournalEventListener {
    35
    36  	private JournalStatsFilter journal;
    37  	private Random random = new Random();
    38  	private byte data[];
    39  	private int workerCount=0;
    40  	private PrintWriter statWriter;
    41  	// Performance test Options
    42
    43  	// The output goes here:
    44  	protected File journalDirectory = new File("journal-logs");
    45  	protected File statCSVFile = new File("stats.csv");;
    46
    47  	// Controls how often we start a new batch of workers.
    48  	protected int workerIncrement=20;
    49  	protected long incrementDelay=1000*20;
    50  	protected boolean verbose=true;
    51
    52  	// Worker configuration.
    53  	protected int recordSize=1024;
    54  	protected int syncFrequency=15;
    55  	protected int workerThinkTime=100;
    56
    57      private final class Worker implements Runnable {
    58  		public void run() {
    59  			int i=random.nextInt()%syncFrequency;
    60  			while(true) {
    61  				boolean sync=false;
    62
    63  				if( syncFrequency>=0 && (i%syncFrequency)==0 ) {
    64  					sync=true;
    65  				}
    66  				try {
    67  					journal.write(new ByteArrayPacket(data), sync);
    68  					Thread.sleep(workerThinkTime);
    69  				} catch (Exception e) {
    70  					e.printStackTrace();
    71  					return;
    72  				}
    73  				i++;
    74  			}
    75  		}
    76  	}
    77
    78      /***
    79       * @throws IOException
    80  	 *
    81  	 */
    82  	protected void exec() throws Exception {
    83
    84  		System.out.println("Client threads write records using: Record Size: "+recordSize+", Sync Frequency: "+syncFrequency+", Worker Think Time: "+workerThinkTime);
    85
    86  		// Create the record and fill it with some values.
    87  		data = new byte[recordSize];
    88  		for (int i = 0; i < data.length; i++) {
    89  			data[i] = (byte)i;
    90  		}
    91
    92  		if( statCSVFile!=null ) {
    93  			statWriter = new PrintWriter(new FileOutputStream(statCSVFile));
    94  			statWriter.println("Threads,Throughput (k/s),Forcd write latency (ms),Throughput (records/s)");
    95  		}
    96
    97          if( journalDirectory.exists() ) {
    98          	deleteDir(journalDirectory);
    99          }
    100         journal = new JournalStatsFilter(createJournal()).enableDetailedStats(verbose);
    101         journal.setJournalEventListener(this);
    102
    103         try {
    104
    105         	// Wait a little to see the worker affect the stats.
    106         	// Increment the number of workers every few seconds.
    107         	while(true) {
    108         		System.out.println("Starting "+workerIncrement+" Workers...");
    109             	for(int i=0;i <workerIncrement;i++) {
    110                 	new Thread(new Worker()).start();
    111                 	workerCount++;
    112             	}
    113
    114             	// Wait a little to see the worker affect the stats.
    115             	System.out.println("Waiting "+(incrementDelay/1000)+" seconds before next Stat sample.");
    116             	Thread.sleep(incrementDelay);
    117             	displayStats();
    118             	journal.reset();
    119         	}
    120
    121
    122         } finally {
    123         	journal.close();
    124         }
    125 	}
    126
    127 	private void displayStats() {
    128 		System.out.println("Stats at "+workerCount+" workers.");
    129 		System.out.println(journal);
    130 		if( statWriter!= null ) {
    131 			statWriter.println(""+workerCount+","+journal.getThroughputKps()+","+journal.getAvgSyncedLatencyMs()+","+journal.getThroughputRps());
    132 			statWriter.flush();
    133 		}
    134 	}
    135
    136 	/***
    137 	 * @return
    138 	 */
    139 	abstract public Journal createJournal() throws Exception;
    140
    141 	static private void deleteDir(File f) {
    142 		File[] files = f.listFiles();
    143 		for (int i = 0; i < files.length; i++) {
    144 			File file = files[i];
    145 			file.delete();
    146 		}
    147 		f.delete();
    148 	}
    149
    150
    151 	public void overflowNotification(RecordLocation safeLocation) {
    152 		try {
    153 			System.out.println("Mark set: "+safeLocation);
    154 			journal.setMark(safeLocation, false);
    155 		} catch (InvalidRecordLocationException e) {
    156 			e.printStackTrace();
    157 		} catch (IOException e) {
    158 			e.printStackTrace();
    159 		}
    160 	}
    161 }