Merge branch 'wip-getdir'

.gitignore

@@ -54,8 +54,7 @@ core
 .cproject
 /build-doc
 /doc/object_store.png
-/src/test_libhadoopcephfs_build
-/src/test_filestore_idempotent
+/src/test_*
 
 # temporary directory used by e.g. "make distcheck", e.g. ceph-0.42
 /ceph-[0-9]*/

qa/workunits/libcephfs/test.sh

@@ -0,0 +1,5 @@
+#!/bin/sh -e
+
+test_libcephfs_readdir
+
+exit 0

src/Makefile.am

@@ -672,6 +672,12 @@ test_rados_api_misc_LDADD =  librados.la ${UNITTEST_STATIC_LDADD}
 test_rados_api_misc_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS}
 bin_DEBUGPROGRAMS += test_rados_api_misc
 
+test_libcephfs_readdir_SOURCES = test/libcephfs/readdir_r_cb.cc
+test_libcephfs_readdir_LDFLAGS = -pthread ${AM_LDFLAGS}
+test_libcephfs_readdir_LDADD =  ${UNITTEST_LDADD} libcephfs.la
+test_libcephfs_readdir_CXXFLAGS = $(AM_CXXFLAGS) ${UNITTEST_CXXFLAGS}
+bin_DEBUGPROGRAMS += test_libcephfs_readdir
+
 test_store_SOURCES = test/store_test.cc
 test_store_LDFLAGS = ${AM_LDFLAGS}
 test_store_LDADD =  ${UNITTEST_STATIC_LDADD} libos.la $(LIBGLOBAL_LDA)

src/client/Client.cc

@@ -4371,7 +4371,7 @@ int Client::_readdir_cache_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p)
   if (!dir) {
     ldout(cct, 10) << " dir is empty" << dendl;
     dirp->set_end();
-    return 1;
+    return 0;
   }
 
   map<string,Dentry*>::iterator pd;
@@ -4419,7 +4419,7 @@ int Client::_readdir_cache_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p)
 
   ldout(cct, 10) << "_readdir_cache_cb " << dirp << " on " << dirp->inode->ino << " at end" << dendl;
   dirp->set_end();
-  return 1;
+  return 0;
 }
 
 int Client::readdir_r_cb(dir_result_t *d, add_dirent_cb_t cb, void *p)
@@ -4552,7 +4552,7 @@ int Client::readdir_r_cb(dir_result_t *d, add_dirent_cb_t cb, void *p)
     }
 
     dirp->set_end();
-    return 1;
+    return 0;
   }
   assert(0);
   return 0;
@@ -4580,12 +4580,13 @@ struct single_readdir {
   bool full;
 };
 
-static int _readdir_single_dirent_cb(void *p, struct dirent *de, struct stat *st, int stmask, off_t off)
+static int _readdir_single_dirent_cb(void *p, struct dirent *de, struct stat *st,
+				     int stmask, off_t off)
 {
   single_readdir *c = (single_readdir *)p;
 
   if (c->full)
-    return -1;
+    return -1;  // already filled this dirent
 
   *c->de = *de;
   if (c->st)
@@ -4596,7 +4597,7 @@ static int _readdir_single_dirent_cb(void *p, struct dirent *de, struct stat *st
   return 0;  
 }
 
-struct dirent * Client::readdir(dir_result_t *d)
+struct dirent *Client::readdir(dir_result_t *d)
 {
   int ret;
   static int stmask;
@@ -4608,19 +4609,11 @@ struct dirent * Client::readdir(dir_result_t *d)
   sr.stmask = &stmask;
   sr.full = false;
 
-  /*
-   * Return mechanisms are non-obvious (callback appears intended for multi-read mechanism like ceph-fuse)
-   * readdir_r_cb=0 end of directory reached on prior call
-   * readdir_r_cb=0 entry filled and offset now at end of the directory
-   * readdir_r_cb=-1 entry is filled successfully, not end of dir
-   * readdir_r_cb=-(other) on error
-   * callback leaves sr.full=false when 'offset is at end of directory'
-   * callback may leave sr.full=false on error
-   * callback sets sr.full=true on 'successfully read dirent'
-   */
+  // our callback fills the dirent and sets sr.full=true on first
+  // call, and returns -1 the second time around.
   ret = readdir_r_cb(d, _readdir_single_dirent_cb, (void *)&sr);
   if (ret < -1) {
-    errno = -ret;
+    errno = -ret;  // this sucks.
     return (dirent *) NULL;
   }
   if (sr.full) {
@@ -4637,8 +4630,10 @@ int Client::readdirplus_r(dir_result_t *d, struct dirent *de, struct stat *st, i
   sr.stmask = stmask;
   sr.full = false;
 
+  // our callback fills the dirent and sets sr.full=true on first
+  // call, and returns -1 the second time around.
   int r = readdir_r_cb(d, _readdir_single_dirent_cb, (void *)&sr);
-  if (r < 0)
+  if (r < -1)
     return r;
   if (sr.full)
     return 1;
@@ -4685,12 +4680,17 @@ int Client::_getdents(dir_result_t *dir, char *buf, int buflen, bool fullent)
   gr.pos = 0;
 
   int r = readdir_r_cb(dir, _readdir_getdent_cb, (void *)&gr);
-  if (r < 0) 
-    return r;
-
-  if (gr.pos == 0)
-    return -ERANGE;
 
+  if (r < 0) { // some error
+    if (r == -1) { // buffer ran out of space
+      if (gr.pos) { // but we got some entries already!
+        return gr.pos;
+      } // or we need a larger buffer
+      return -ERANGE;
+    } else { // actual error, return it
+      return r;
+    }
+  }
   return gr.pos;
 }
 

src/client/Client.h

@@ -518,6 +518,14 @@ public:
   int opendir(const char *name, dir_result_t **dirpp);
   int closedir(dir_result_t *dirp);
 
+  /**
+   * Fill a directory listing from dirp, invoking cb for each entry
+   * with the given pointer, the dirent, the struct stat, the stmask,
+   * and the offset.
+   *
+   * Returns 0 if it reached the end of the directory.
+   * If @cb returns a negative error code, stop and return that.
+   */
   int readdir_r_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p);
 
   struct dirent * readdir(dir_result_t *d);
@@ -526,6 +534,11 @@ public:
 
   int getdir(const char *relpath, list<string>& names);  // get the whole dir at once.
 
+  /**
+   * Returns the length of the buffer that got filled in, or -errno.
+   * If it returns -ERANGE you just need to increase the size of the
+   * buffer and try again.
+   */
   int _getdents(dir_result_t *dirp, char *buf, int buflen, bool ful);  // get a bunch of dentries at once
   int getdents(dir_result_t *dirp, char *buf, int buflen) {
     return _getdents(dirp, buf, buflen, true);

src/client/fuse_ll.cc

@@ -421,9 +421,7 @@ static void ceph_ll_readdir(fuse_req_t req, fuse_ino_t ino, size_t size,
   rc.snap = fino_snap(ino);
 
   int r;
-  do {
-    r = client->readdir_r_cb(dirp, ceph_ll_add_dirent, &rc);
-  } while (r > 0);
+  r = client->readdir_r_cb(dirp, ceph_ll_add_dirent, &rc);
 
   fuse_reply_buf(req, rc.buf, rc.pos);
   delete[] rc.buf;

src/client/hadoop/CephFSInterface.cc

@@ -741,10 +741,18 @@ JNIEXPORT jint JNICALL Java_org_apache_hadoop_fs_ceph_CephTalker_ceph_1statfs
  * Returns: an int containing the number of times replicated.
  */
 JNIEXPORT jint JNICALL Java_org_apache_hadoop_fs_ceph_CephTalker_ceph_1replication
-(JNIEnv *env, jobject obj, jint fh)
+(JNIEnv *env, jobject obj, jstring j_path)
 {
-  struct ceph_mount_info *cmount = get_ceph_mount_t(env, obj);
-  int replication = ceph_get_file_replication(cmount, (int)fh);
+  //get c-string of path, send off to libceph, release c-string, return
+  const char *c_path = env->GetStringUTFChars(j_path, 0);
+  if (c_path == NULL)
+    return -ENOMEM;
+  ceph_mount_info *cmount = get_ceph_mount_t(env, obj);
+  int fh = 0;
+  fh = ceph_open(cmount, c_path, O_RDONLY, 0);
+  int replication = ceph_get_file_replication(cmount, fh);
+  ceph_close(cmount, fh);
+  env->ReleaseStringUTFChars(j_path, c_path);
   return replication;
 }
 

src/client/hadoop/org_apache_hadoop_fs_ceph_CephTalker.h

@@ -170,10 +170,10 @@ JNIEXPORT jint JNICALL Java_org_apache_hadoop_fs_ceph_CephTalker_ceph_1statfs
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephTalker
  * Method:    ceph_replication
- * Signature: (I)I
+ * Signature: (Ljava/lang/String;)I
  */
 JNIEXPORT jint JNICALL Java_org_apache_hadoop_fs_ceph_CephTalker_ceph_1replication
-  (JNIEnv *, jobject, jint);
+  (JNIEnv *, jobject, jstring);
 
 /*
  * Class:     org_apache_hadoop_fs_ceph_CephTalker

src/include/cephfs/libcephfs.h

@@ -91,6 +91,10 @@ int ceph_readdir_r(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp,
 int ceph_readdirplus_r(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp, struct dirent *de,
 		       struct stat *st, int *stmask);
 int ceph_getdents(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp, char *name, int buflen);
+/**
+ * This returns the used buffer space on success, -ERANGE if the buffer
+ * is not large enough to hold a name, or -errno on other issues.
+ */
 int ceph_getdnames(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp, char *name, int buflen);
 void ceph_rewinddir(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp);
 loff_t ceph_telldir(struct ceph_mount_info *cmount, struct ceph_dir_result *dirp);

src/test/libcephfs/readdir_r_cb.cc

@@ -0,0 +1,55 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <sys/fcntl.h>
+
+TEST(LibCephFS, Readdir_r_cb) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  const char *c_dir = "/readdir_r_cb_tests";
+  struct ceph_dir_result *dirp;
+  ASSERT_EQ(0, ceph_mkdirs(cmount, c_dir, 0777));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+
+  // dir is empty, check that it only contains . and ..
+  int buflen = 100;
+  char *buf = new char[buflen];
+  // . is 2, .. is 3 (for null terminators)
+  ASSERT_EQ(5, ceph_getdnames(cmount, dirp, buf, buflen));
+  const char *c_file = "/readdir_r_cb_tests/foo";
+  int fd = ceph_open(cmount, c_file, O_CREAT, 0777);
+  ASSERT_LT(0, fd);
+
+  // check correctness with one entry
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(9, ceph_getdnames(cmount, dirp, buf, buflen)); // ., .., foo
+
+  // check correctness if buffer is too small
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_GE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(-ERANGE, ceph_getdnames(cmount, dirp, buf, 1));
+
+  //check correctness if it needs to split listing
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(5, ceph_getdnames(cmount, dirp, buf, 6));
+  ASSERT_EQ(4, ceph_getdnames(cmount, dirp, buf, 6));
+}