test/long/org/apache/cassandra/db/commitlog/ComitLogStress.java

package org.apache.cassandra.db.commitlog;
/*
 * 
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * 
 */


import java.nio.ByteBuffer;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.cassandra.Util;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;

public class ComitLogStress
{

    public static final String format = "%s,%s,%s,%s,%s,%s";

    public static void main(String[] args) throws Exception {
        int NUM_THREADS = Runtime.getRuntime().availableProcessors();
        if (args.length >= 1) {
            NUM_THREADS = Integer.parseInt(args[0]);
            System.out.println("Setting num threads to: " + NUM_THREADS);
        }
        ExecutorService executor = new JMXEnabledThreadPoolExecutor(NUM_THREADS, NUM_THREADS, 60,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10 * NUM_THREADS), new NamedThreadFactory(""), "");
        ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1);

        org.apache.cassandra.SchemaLoader.loadSchema();
        final AtomicLong count = new AtomicLong();
        final long start = System.currentTimeMillis();
        System.out.println(String.format(format, "seconds", "max_mb", "allocated_mb", "free_mb", "diffrence", "count"));
        scheduled.scheduleAtFixedRate(new Runnable() {
            long lastUpdate = 0;

            public void run() {
                Runtime runtime = Runtime.getRuntime();
                long maxMemory = mb(runtime.maxMemory());
                long allocatedMemory = mb(runtime.totalMemory());
                long freeMemory = mb(runtime.freeMemory());
                long temp = count.get();
                System.out.println(String.format(format, ((System.currentTimeMillis() - start) / 1000),
                        maxMemory, allocatedMemory, freeMemory, (temp - lastUpdate), lastUpdate));
                lastUpdate = temp;
            }
        }, 1, 1, TimeUnit.SECONDS);

        while (true) {
            executor.execute(new CommitlogExecutor());
            count.incrementAndGet();
        }
    }

    private static long mb(long maxMemory) {
        return maxMemory / (1024 * 1024);
    }

    static final String keyString = UUIDGen.getTimeUUID().toString();
    public static class CommitlogExecutor implements Runnable {
        public void run() {
            String ks = "Keyspace1";
            ByteBuffer key = ByteBufferUtil.bytes(keyString);
            Mutation mutation = new Mutation(ks, key);
            mutation.add("Standard1", Util.cellname("name"), ByteBufferUtil.bytes("value"),
                    System.currentTimeMillis());
            CommitLog.instance.add(mutation);
        }
    }
}