test: test_workload_gen: Mimic an OSD's workload.

In it's current state, the workload generator will queue a lot of
transactions onto the FileStore, and will wait if needed in case
there are too many in-flight transactions.

The workload generator will perform the transactions over a
pre-determined number of collections and objects, which may very
well be defined at runtime by using the options '-C <VAL>' and
'-O <VAL>' for collections and objects per collection, respectively.
If these are not provided, the program will default to 30 collections
and 6000 objects per collection.

3 files changed, 400 insertions, 0 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 4440d046fde..129ecc86bed 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -710,6 +710,12 @@ test_store_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS} \
 	-I$(top_srcdir)/src/leveldb/include
 bin_DEBUGPROGRAMS += test_store
 
+test_wrkldgen_SOURCES = test/test_workload_gen/workload_generator.cc
+test_wrkldgen_LDFLAGS = ${AM_LDFLAGS}
+test_wrkldgen_LDADD =  ${UNITTEST_STATIC_LDADD} libos.la $(LIBGLOBAL_LDA)
+test_wrkldgen_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS}
+bin_DEBUGPROGRAMS += test_wrkldgen
+
 xattr_bench_SOURCES = test/xattr_bench.cc
 xattr_bench_LDFLAGS = ${AM_LDFLAGS}
 xattr_bench_LDADD =  ${UNITTEST_STATIC_LDADD} libos.la leveldb/libleveldb.a $(LIBGLOBAL_LDA)
diff --git a/src/test/test_workload_gen/workload_generator.cc b/src/test/test_workload_gen/workload_generator.cc
new file mode 100644
index 00000000000..7762440badc
--- /dev/null
+++ b/src/test/test_workload_gen/workload_generator.cc
@@ -0,0 +1,276 @@
+/*
+ * workload_generator.cc
+ *
+ *  Created on: Mar 12, 2012
+ *      Author: jecluis
+ */
+#include <stdio.h>
+#include <string.h>
+#include <iostream>
+#include <assert.h>
+#include <time.h>
+#include <stdlib.h>
+#include <signal.h>
+#include "os/FileStore.h"
+#include "common/ceph_argparse.h"
+#include "global/global_init.h"
+#include "common/debug.h"
+#include <boost/scoped_ptr.hpp>
+#include "workload_generator.h"
+
+boost::scoped_ptr<WorkloadGenerator> wrkldgen;
+const coll_t WorkloadGenerator::META_COLL("meta");
+const coll_t WorkloadGenerator::TEMP_COLL("temp");
+
+void sig_handler(int sig) {
+	if (sig == SIGINT) {
+		wrkldgen->stop();
+	}
+}
+
+WorkloadGenerator::WorkloadGenerator(vector<const char*> args) :
+		NUM_COLLS(DEF_NUM_COLLS), 
+		NUM_OBJ_PER_COLL(DEF_NUM_OBJ_PER_COLL),
+		store(0), rng(time(NULL)), in_flight(0),
+		lock("State Lock"), stop_running(false)
+{
+	init_args(args);
+
+	::mkdir("workload_gen_dir", 0777);
+	ObjectStore *store_ptr = new FileStore(string("workload_gen_dir"),
+			string("workload_gen_journal"));
+	store.reset(store_ptr);
+	store->mkfs();
+	store->mount();
+
+	osr = new ObjectStore::Sequencer[NUM_COLLS];
+
+	init();
+}
+
+void WorkloadGenerator::init_args(vector<const char*> args) {
+	for (std::vector<const char*>::iterator i = args.begin();
+			i != args.end(); ) {
+		string val;
+
+		if (ceph_argparse_double_dash(args, i)) {
+			break;
+		} else if (ceph_argparse_witharg(args, i, &val,
+				"-C", "--num-collections", (char*)NULL)) {
+			NUM_COLLS = strtoll(val.c_str(), NULL, 10);
+		} else if (ceph_argparse_witharg(args, i, &val,
+				"-O", "--num-objects", (char*)NULL)) {
+			NUM_OBJ_PER_COLL = strtoll(val.c_str(), NULL, 10);
+		}
+//		else if (ceph_argparse_witharg(args, i, &val,
+//				"-o", "--outfn", (char*)NULL)) {
+//			outfn = val;
+//		}
+	}
+}
+
+void WorkloadGenerator::init() {
+
+	dout(0) << "Initializing..." << dendl;
+
+	ObjectStore::Transaction *t = new ObjectStore::Transaction;
+
+	t->create_collection(META_COLL);
+	t->create_collection(TEMP_COLL);
+//	store->queue_transaction(&osr, t);
+	store->apply_transaction(*t);
+
+	wait_for_ready();
+
+	char buf[100];
+	for (int i = 0; i < NUM_COLLS; i ++) {
+		memset(buf, 0, 100);
+		snprintf(buf, 100, "0.%d_head", i);
+		coll_t coll(buf);
+
+		dout(0) << "Creating collection " << coll.to_str() << dendl;
+
+		t = new ObjectStore::Transaction;
+
+		t->create_collection(coll);
+
+		memset(buf, 0, 100);
+		snprintf(buf, 100, "pglog_0.%d_head", i);
+		hobject_t coll_meta_obj(sobject_t(object_t(buf), CEPH_NOSNAP));
+		t->touch(META_COLL, coll_meta_obj);
+
+		store->queue_transaction(&osr[i], t,
+				new C_WorkloadGeneratorOnReadable(this, t));
+		in_flight++;
+	}
+
+	wait_for_done();
+	dout(0) << "Done initializing!" << dendl;
+}
+
+int WorkloadGenerator::get_random_collection_nr() {
+	return (rand() % NUM_COLLS);
+}
+
+int WorkloadGenerator::get_random_object_nr(int coll_nr) {
+	return ((rand() % NUM_OBJ_PER_COLL) + (coll_nr*NUM_OBJ_PER_COLL));
+}
+
+coll_t WorkloadGenerator::get_collection_by_nr(int nr) {
+
+	char buf[100];
+	memset(buf, 0, 100);
+
+	snprintf(buf, 100, "0.%d_head", nr);
+	return coll_t(buf);
+}
+
+hobject_t WorkloadGenerator::get_object_by_nr(int nr) {
+
+	char buf[100];
+	memset(buf, 0, 100);
+	snprintf(buf, 100, "%d", nr);
+
+	return hobject_t(sobject_t(object_t(buf), CEPH_NOSNAP));
+}
+
+hobject_t WorkloadGenerator::get_coll_meta_object(coll_t coll) {
+	char buf[100];
+	memset(buf, 0, 100);
+	snprintf(buf, 100, "pglog_%s", coll.c_str());
+
+	return hobject_t(sobject_t(object_t(buf), CEPH_NOSNAP));
+}
+
+/**
+ * We'll generate a random amount of bytes, ranging from a single byte up to
+ * a couple of MB.
+ */
+size_t WorkloadGenerator::get_random_byte_amount(size_t min, size_t max) {
+	size_t diff = max - min;
+
+	return (size_t) (min + (rand() % diff));
+}
+
+void WorkloadGenerator::get_filled_byte_array(bufferlist& bl, size_t size) {
+
+	static const char alphanum[] =
+			"0123456789"
+			"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+			"abcdefghijklmnopqrstuvwxyz";
+	bufferptr bp(size);
+	for (unsigned int i = 0; i < size-1; i ++) {
+		bp[i] = alphanum[rand() % sizeof(alphanum)];
+	}
+	bp[size-1] = '\0';
+	bl.append(bp);
+}
+
+int WorkloadGenerator::do_write_object(ObjectStore::Transaction *t,
+		coll_t coll, hobject_t obj) {
+
+	size_t bytes = get_random_byte_amount(MIN_WRITE_BYTES, MAX_WRITE_BYTES);
+	bufferlist bl;
+	get_filled_byte_array(bl, bytes);
+	t->write(coll, obj, 0, bl.length(), bl);
+
+	return 0;
+}
+
+int WorkloadGenerator::do_setattr_object(ObjectStore::Transaction *t,
+		coll_t coll, hobject_t obj) {
+	size_t size;
+	size = get_random_byte_amount(MIN_XATTR_OBJ_BYTES, MAX_XATTR_OBJ_BYTES);
+
+	bufferlist bl;
+	get_filled_byte_array(bl, size);
+	t->setattr(coll, obj, "objxattr", bl);
+
+	return 0;
+}
+
+int WorkloadGenerator::do_setattr_collection(ObjectStore::Transaction *t,
+		coll_t coll) {
+
+	size_t size;
+	size = get_random_byte_amount(MIN_XATTR_COLL_BYTES, MAX_XATTR_COLL_BYTES);
+
+	bufferlist bl;
+	get_filled_byte_array(bl, size);
+	t->collection_setattr(coll, "collxattr", bl);
+
+	return 0;
+}
+
+int WorkloadGenerator::do_append_log(ObjectStore::Transaction *t,
+		coll_t coll) {
+
+	bufferlist bl;
+	get_filled_byte_array(bl, LOG_APPEND_BYTES);
+	hobject_t log_obj = get_coll_meta_object(coll);
+
+	struct stat st;
+	int err = store->stat(META_COLL, log_obj, &st);
+	assert(err >= 0);
+	t->write(META_COLL, log_obj, st.st_size, bl.length(), bl);
+
+	return 0;
+}
+
+void WorkloadGenerator::run() {
+
+	do {
+		lock.Lock();
+		wait_for_ready();
+
+		int coll_nr = get_random_collection_nr();
+		int obj_nr = get_random_object_nr(coll_nr);
+		coll_t coll = get_collection_by_nr(coll_nr);
+		hobject_t obj = get_object_by_nr(obj_nr);
+
+		ObjectStore::Transaction *t = new ObjectStore::Transaction;
+		int err;
+
+		err = do_write_object(t, coll, obj);
+		assert(err == 0);
+
+		err = do_setattr_object(t, coll, obj);
+		assert(err == 0);
+
+		err = do_setattr_collection(t, coll);
+		assert(err == 0);
+
+		err = do_append_log(t, coll);
+		assert(err == 0);
+
+		store->queue_transaction(&osr[coll_nr], t,
+				new C_WorkloadGeneratorOnReadable(this, t));
+
+		in_flight++;
+
+		lock.Unlock();
+	} while (!stop_running);
+}
+
+void WorkloadGenerator::print_results() {
+
+}
+
+
+int main(int argc, char *argv[]) {
+	vector<const char*> args;
+	argv_to_vec(argc, (const char **)argv, args);
+
+	global_init(args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, 0);
+	common_init_finish(g_ceph_context);
+	g_ceph_context->_conf->set_val("osd_journal_size", "400");
+	g_ceph_context->_conf->apply_changes(NULL);
+
+
+	WorkloadGenerator *wrkldgen_ptr = new WorkloadGenerator(args);
+	wrkldgen.reset(wrkldgen_ptr);
+	wrkldgen->run();
+	wrkldgen->print_results();
+
+	return 0;
+}
diff --git a/src/test/test_workload_gen/workload_generator.h b/src/test/test_workload_gen/workload_generator.h
new file mode 100644
index 00000000000..efe2538b924
--- /dev/null
+++ b/src/test/test_workload_gen/workload_generator.h
@@ -0,0 +1,118 @@
+/*
+ * workload_generator.h
+ *
+ *  Created on: Mar 12, 2012
+ *      Author: jecluis
+ */
+
+#ifndef WORKLOAD_GENERATOR_H_
+#define WORKLOAD_GENERATOR_H_
+
+#include "os/FileStore.h"
+#include <boost/scoped_ptr.hpp>
+#include <boost/random/mersenne_twister.hpp>
+#include <boost/random/uniform_int.hpp>
+#include <pthread.h>
+
+typedef boost::mt11213b gen_type;
+
+class WorkloadGenerator
+{
+private:
+	  static const int NUM_THREADS 		= 2;
+	  static const int MAX_IN_FLIGHT	= 50;
+
+	  static const int DEF_NUM_OBJ_PER_COLL = 6000;
+	  static const int DEF_NUM_COLLS	= 30;
+
+	  static const coll_t META_COLL;
+	  static const coll_t TEMP_COLL;
+
+	  static const size_t MIN_WRITE_BYTES = 1;
+	  static const size_t MAX_WRITE_MB	  = 5;
+	  static const size_t MAX_WRITE_BYTES = (MAX_WRITE_MB * 1024 * 1024);
+
+	  static const size_t MIN_XATTR_OBJ_BYTES 	= 2;
+	  static const size_t MAX_XATTR_OBJ_BYTES 	= 300;
+	  static const size_t MIN_XATTR_COLL_BYTES 	= 4;
+	  static const size_t MAX_XATTR_COLL_BYTES 	= 600;
+//	  static const size_t XATTR_NAME_BYTES		= 30;
+
+	  static const size_t LOG_APPEND_BYTES		= 1024;
+
+	  int NUM_COLLS;
+	  int NUM_OBJ_PER_COLL;
+
+	  boost::scoped_ptr<ObjectStore> store;
+
+	  gen_type rng;
+	  int in_flight;
+	  ObjectStore::Sequencer *osr;
+
+	  Mutex lock;
+	  Cond cond;
+
+	  bool stop_running;
+
+	  void wait_for_ready() {
+		  while (in_flight >= MAX_IN_FLIGHT)
+			  cond.Wait(lock);
+	  }
+
+	  void wait_for_done() {
+		  Mutex::Locker locker(lock);
+		  while (in_flight)
+			  cond.Wait(lock);
+	  }
+
+	  void init_args(vector<const char*> args);
+	  void init();
+
+	  int get_random_collection_nr();
+	  int get_random_object_nr(int coll_nr);
+
+	  coll_t get_collection_by_nr(int nr);
+	  hobject_t get_object_by_nr(int nr);
+	  hobject_t get_coll_meta_object(coll_t coll);
+
+	  size_t get_random_byte_amount(size_t min, size_t max);
+	  void get_filled_byte_array(bufferlist& bl, size_t size);
+
+	  int do_write_object(ObjectStore::Transaction *t,
+			  coll_t coll, hobject_t obj);
+	  int do_setattr_object(ObjectStore::Transaction *t,
+	  			  coll_t coll, hobject_t obj);
+	  int do_setattr_collection(ObjectStore::Transaction *t, coll_t coll);
+	  int do_append_log(ObjectStore::Transaction *t, coll_t coll);
+
+public:
+	WorkloadGenerator(vector<const char*> args);
+	~WorkloadGenerator() {
+		store->umount();
+	}
+
+	class C_WorkloadGeneratorOnReadable : public Context {
+		WorkloadGenerator *state;
+		ObjectStore::Transaction *t;
+
+	public:
+		C_WorkloadGeneratorOnReadable(WorkloadGenerator *state,
+				ObjectStore::Transaction *t) : state(state), t(t) {}
+
+		void finish(int r) {
+			dout(0) << "Got one back!" << dendl;
+			Mutex::Locker locker(state->lock);
+			state->in_flight--;
+			state->cond.Signal();
+		}
+	};
+
+
+	void run(void);
+	void print_results(void);
+	void stop() {
+		stop_running = true;
+	}
+};
+
+#endif /* WORKLOAD_GENERATOR_H_ */