Hadoop: Allow a command-line like property, and change the parameter passing

so that Java passes one string to the C++ code and that parses it.

src/client/hadoop/CephFSInterface.cc

@@ -16,37 +16,44 @@ static int path_size;
  * Method:    ceph_initializeClient
  * Signature: (Ljava/lang/String;)Z
  */
-
-/*
- * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
- * Method:    ceph_initializeClient
- * Signature: (Ljava/lang/String;Ljava/lang/String;)Z
- */
 JNIEXPORT jboolean JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1initializeClient
-  (JNIEnv *env, jobject obj, jstring j_debug_level, jstring j_mon_addr)
+  (JNIEnv *env, jobject obj, jstring j_args )
 {
   dout(3) << "CephFSInterface: Initializing Ceph client:" << dendl;
-
-  const char *c_debug_level = env->GetStringUTFChars(j_debug_level, 0);
-  if (c_debug_level == NULL) return false; //out of memory!
-  const char *c_mon_addr = env->GetStringUTFChars(j_mon_addr, 0);
-  if(c_mon_addr == NULL) {
-    env->ReleaseStringUTFChars(j_debug_level, c_debug_level);
-    return false;
-  }
+  const char *c_args = env->GetStringUTFChars(j_args, 0);
+  if (c_args == NULL) return false; //out of memory!
+  string args(c_args);
   path_size = 64; //reasonable starting point?
-  //construct an arguments array
-  const char *argv[10];
-  int argc = 0;
-  argv[argc++] = "CephFSInterface";
-  argv[argc++] = "-m";
-  argv[argc++] = c_mon_addr;
-  argv[argc++] = "--debug_client";
-  argv[argc++] = c_debug_level;
 
-  int r = ceph_initialize(argc, argv);
-  env->ReleaseStringUTFChars(j_debug_level, c_debug_level);
-  env->ReleaseStringUTFChars(j_mon_addr, c_mon_addr);
+  //construct an arguments vector
+  vector<string> args_vec;
+  size_t i = 0;
+  size_t j = 0;
+  while (1) {
+    j = args.find(' ', i);
+    if (j == string::npos) {
+      if (i == 0) { //there were no spaces? That can't happen!
+	env->ReleaseStringUTFChars(j_args, c_args);
+	return false;
+      }
+      //otherwise it's the last argument, so push it on and exit loop
+      args_vec.push_back(args.substr(i, args.size()));
+      break;
+    }
+    if (j!=i) //if there are two spaces in a row, dont' make a new arg
+      args_vec.push_back(args.substr(i, j-i));
+    i = j+1;
+  }
+
+  //convert to array
+  const char ** argv = new const char*[args_vec.size()];
+  for (size_t i = 0; i < args_vec.size(); ++i)
+    argv[i] = args_vec[i].c_str();
+
+  int r = ceph_initialize(args_vec.size(), argv);
+  env->ReleaseStringUTFChars(j_args, c_args);
+  delete argv;
+
   if (r < 0) return false;
   r = ceph_mount();
   if (r < 0) return false;

src/client/hadoop/CephFSInterface.h

@@ -17,10 +17,10 @@ extern "C" {
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
  * Method:    ceph_initializeClient
- * Signature: (Ljava/lang/String;Ljava/lang/String;)Z
+ * Signature: (Ljava/lang/String;)Z
  */
 JNIEXPORT jboolean JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1initializeClient
-  (JNIEnv *, jobject, jstring, jstring);
+  (JNIEnv *, jobject, jstring);
 
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephFileSystem

src/client/hadoop/ceph/CephFileSystem.java

@@ -37,12 +37,15 @@ import org.apache.hadoop.fs.FileStatus;
  * fs.ceph.libDir -- the directory that libceph and libhadoopceph are
  * located in. This assumes Hadoop is being run on a linux-style machine
  * with names like libceph.so.
+ * fs.ceph.commandLine -- if you prefer you can fill in this property
+ * just as you would when starting Ceph up from the command line. Specific
+ * properties override any configuration specified here.
  * <p>
  * You can also enable debugging of the CephFileSystem and Ceph itself: <br>
  * fs.ceph.debug -- if 'true' will print out method enter/exit messages,
  * plus a little more.
- * fs.ceph.debugLevel -- a number that will print out diagnostic messages
- * from Ceph of at least that importance.
+ * fs.ceph.clientDebug/fs.ceph.messengerDebug -- will print out debugging
+ * from the respective Ceph system of at least that importance.
  */
 public class CephFileSystem extends FileSystem {
 
@@ -59,7 +62,7 @@ public class CephFileSystem extends FileSystem {
   private static String monAddr;
   private static String fs_default_name;
   
-  private native boolean ceph_initializeClient(String debugLevel, String mon);
+  private native boolean ceph_initializeClient(String arguments);
   private native String  ceph_getcwd();
   private native boolean ceph_setcwd(String path);
   private native boolean ceph_rmdir(String path);
@@ -120,19 +123,36 @@ public class CephFileSystem extends FileSystem {
       System.load(conf.get("fs.ceph.libDir")+"/libhadoopcephfs.so");
       System.load(conf.get("fs.ceph.libDir")+"/libceph.so");
       super.initialize(uri, conf);
-      //store.initialize(uri, conf);
       setConf(conf);
       this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());    
       
-      conf.setIfUnset("fs.ceph.debugLevel", "0");
-      conf.setIfUnset("fs.ceph.debug", "false");
       fs_default_name = conf.get("fs.default.name");
-      monAddr = conf.get("fs.ceph.monAddr");
-      if (monAddr == null) throw new IOException("You must specify a Ceph monitor address!");
-      cephDebugLevel = conf.get("fs.ceph.debugLevel");
-      debug = ("true".equals(conf.get("fs.ceph.debug")));
-      //  Initializes the client
-      if (!ceph_initializeClient(cephDebugLevel, monAddr)) {
+      debug = ("true".equals(conf.get("fs.ceph.debug", "false")));
+      //build up the arguments for Ceph
+      String arguments = new String();
+      arguments += conf.get("fs.ceph.commandLine", "");
+      if (conf.get("fs.ceph.clientDebug") != null) {
+	arguments += " --debug_client ";
+	arguments += conf.get("fs.ceph.clientDebug");
+      }
+      if (conf.get("fs.ceph.messengerDebug") != null) {
+	arguments += " --debug_ms ";
+	arguments += conf.get("fs.ceph.messengerDebug");
+      }
+      if (conf.get("fs.ceph.monAddr") != null) {
+	arguments += " -m ";
+	arguments += conf.get("fs.ceph.monAddr");
+      }
+      //make sure they gave us a ceph monitor address or conf file
+      if ( (conf.get("fs.ceph.monAddr") == null) &&
+	   (arguments.indexOf("-m") == -1) &&
+	   (arguments.indexOf("-c") == -1) ) {
+	debug("You need to specify a Ceph monitor address.");
+	throw new IOException("You must specify a Ceph monitor address or config file!");
+      }
+      //  Initialize the client
+      if (!ceph_initializeClient(arguments)) {
+	debug("Ceph initialization failed!");
 	throw new IOException("Ceph initialization failed!");
       }
       initialized = true;

src/client/hadoop/org_apache_hadoop_fs_ceph_CephFileSystem.h

@@ -8,14 +8,16 @@
 extern "C" {
 #endif
 #undef org_apache_hadoop_fs_ceph_CephFileSystem_DEFAULT_BLOCK_SIZE
-#define org_apache_hadoop_fs_ceph_CephFileSystem_DEFAULT_BLOCK_SIZE 8388608LL
+#define org_apache_hadoop_fs_ceph_CephFileSystem_DEFAULT_BLOCK_SIZE 4194304LL
+#undef org_apache_hadoop_fs_ceph_CephFileSystem_EEXIST
+#define org_apache_hadoop_fs_ceph_CephFileSystem_EEXIST 17L
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
  * Method:    ceph_initializeClient
- * Signature: (Ljava/lang/String;Ljava/lang/String;)Z
+ * Signature: (Ljava/lang/String;)Z
  */
 JNIEXPORT jboolean JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1initializeClient
-  (JNIEnv *, jobject, jstring, jstring);
+  (JNIEnv *, jobject, jstring);
 
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
@@ -81,14 +83,6 @@ JNIEXPORT jboolean JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1e
 JNIEXPORT jlong JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1getblocksize
   (JNIEnv *, jobject, jstring);
 
-/*
- * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
- * Method:    ceph_getfilesize
- * Signature: (Ljava/lang/String;)J
- */
-JNIEXPORT jlong JNICALL Java_org_apache_hadoop_fs_ceph_CephFileSystem_ceph_1getfilesize
-  (JNIEnv *, jobject, jstring);
-
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephFileSystem
  * Method:    ceph_isdirectory