journal: some fixes, still some issues left

src/Makefile.am

@@ -50,6 +50,10 @@ fakesyn_LDADD = libmon.a libmds.a libosd.a libebofs.a \
 #newsyn_LDADD = libmon.a libmds.a libosd.a libebofs.a \
 #	libclient.a libosdc.a libcrush.a libcommon.a
 
+# tester tools
+streamtest_SOURCES = streamtest.cc
+streamtest_LDADD = libosd.a libebofs.a libcommon.a
+
 ##
 INCLUDES = 
 LDADD = -lpthread 
@@ -62,6 +66,7 @@ bin_PROGRAMS = \
 	cmon cmds cosd csyn \
 	mkmonmap monmaptool crushtool cmonctl \
 	fakesyn \
+	streamtest \
 	$(FUSEBIN) $(NEWSYN)
 noinst_LIBRARIES = \
 	libcommon.a libcrush.a \

src/osd/FakeStore.cc

@@ -151,12 +151,13 @@ int FakeStore::mkfs()
   sprintf(fn, "%s/fsid", basedir.c_str());
   int fd = ::open(fn, O_CREAT|O_TRUNC|O_WRONLY);
   ::write(fd, &fsid, sizeof(fsid));
-  ::fchmod(fd, 666);
+  ::fchmod(fd, 0644);
   ::close(fd);
+  dout(10) << "mkfs fsid is " << fsid << dendl;
 
   // journal?
   struct stat st;
-  sprintf(fn, "%s/journal", basedir.c_str());
+  sprintf(fn, "%s.journal", basedir.c_str());
   if (::stat(fn, &st) == 0) {
     journal = new FileJournal(fsid, &finisher, fn, g_conf.journal_dio);
     if (journal->create() < 0) {
@@ -166,6 +167,8 @@ int FakeStore::mkfs()
     }
     delete journal;
     journal = 0;
+  } else {
+    dout(10) << "mkfs no journal at " << fn << dendl;
   }
 
   if (g_conf.fakestore_dev) {
@@ -226,6 +229,7 @@ int FakeStore::mount()
   fd = ::open(fn, O_RDONLY);
   ::read(fd, &fsid, sizeof(fsid));
   ::close(fd);
+  dout(10) << "mount fsid is " << fsid << dendl;
   
   // get epoch
   sprintf(fn, "%s/commit_epoch", basedir.c_str());
@@ -235,9 +239,13 @@ int FakeStore::mount()
   dout(5) << "mount epoch is " << super_epoch << dendl;
 
   // journal
-  sprintf(fn, "%s/journal", basedir.c_str());
-  if (::stat(fn, &st) == 0)
+  sprintf(fn, "%s.journal", basedir.c_str());
+  if (::stat(fn, &st) == 0) {
+    dout(10) << "mount opening journal at " << fn << dendl;
     journal = new FileJournal(fsid, &finisher, fn, g_conf.journal_dio);
+  } else {
+    dout(10) << "mount no journal at " << fn << dendl;
+  }
   r = journal_replay();
   if (r == -EINVAL) {
     dout(0) << "mount got EINVAL on journal open, not mounting" << dendl;
@@ -379,7 +387,7 @@ int FakeStore::write(pobject_t oid,
     derr(0) << "write couldn't open " << fn << " flags " << flags << " errno " << errno << " " << strerror(errno) << dendl;
     return fd;
   }
-  ::fchmod(fd, 0664);
+  ::fchmod(fd, 0644);
   ::flock(fd, LOCK_EX);    // lock for safety
   
   // seek
@@ -434,7 +442,7 @@ void FakeStore::sync_entry()
     sprintf(fn, "%s/commit_epoch", basedir.c_str());
     int fd = ::open(fn, O_CREAT|O_WRONLY);
     ::write(fd, &super_epoch, sizeof(super_epoch));
-    ::fchmod(fd, 0664);
+    ::fchmod(fd, 0644);
     ::fsync(fd);  // this should cause the fs's journal to commit.
     ::close(fd);
 

src/osd/FileJournal.cc

@@ -102,6 +102,9 @@ int FileJournal::open(epoch_t epoch)
 
   // read header?
   read_header();
+  dout(10) << "open journal header.fsid = " << header.fsid 
+    //<< " vs expected fsid = " << fsid 
+	   << dendl;
   if (header.fsid != fsid) {
     dout(2) << "open journal fsid doesn't match, invalid (someone else's?) journal" << dendl;
     err = -EINVAL;
@@ -152,12 +155,12 @@ int FileJournal::open(epoch_t epoch)
 
     if (read_pos == 0) {
       dout(0) << "no valid journal segments" << dendl;
-      return -EINVAL;
+      return 0; //hrm return -EINVAL; 
     }
 
   } else {
     dout(0) << "journal was empty" << dendl;
-    read_pos = get_top();
+    read_pos = -1;
   }
 
   return 0;
@@ -549,7 +552,7 @@ void FileJournal::make_writeable()
 {
   _open(true);
 
-  if (read_pos)
+  if (read_pos > 0)
     write_pos = read_pos;
   else
     write_pos = get_top();

src/osd/FileJournal.h

@@ -162,6 +162,9 @@ private:
   int open(epoch_t epoch);
   void close();
 
+  bool is_writeable() {
+    return read_pos == 0;
+  }
   void make_writeable();
 
   // writes

src/osd/Journal.h

@@ -34,6 +34,7 @@ public:
   virtual void close() = 0;
 
   // writes
+  virtual bool is_writeable() = 0;
   virtual void make_writeable() = 0;
   virtual void submit_entry(epoch_t epoch, bufferlist& e, Context *oncommit) = 0;
   virtual void commit_epoch_start(epoch_t) = 0;  // mark epoch boundary

src/osd/JournalingObjectStore.cc

@@ -19,7 +19,7 @@ int JournalingObjectStore::journal_replay()
     journal = 0;
     return err;
   }
-  
+
   int count = 0;
   while (1) {
     bufferlist bl;

src/osd/JournalingObjectStore.h

@@ -50,7 +50,7 @@ protected:
   }
 
   void journal_write(pobject_t oid, off_t off, size_t len, const bufferlist& bl, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.write(oid, off, len, bl);
       bufferlist tbl;
@@ -61,7 +61,7 @@ protected:
   }
   
   void journal_zero(pobject_t oid, off_t off, size_t len, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.zero(oid, off, len);
       bufferlist tbl;
@@ -72,7 +72,7 @@ protected:
   }
   
   void journal_remove(pobject_t oid, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.remove(oid);
       bufferlist bl;
@@ -83,7 +83,7 @@ protected:
   }
 
   void journal_truncate(pobject_t oid, off_t size, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.truncate(oid, size);
       bufferlist bl;
@@ -94,7 +94,7 @@ protected:
   }
 
   void journal_clone(pobject_t from, pobject_t to, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.clone(from, to);
       bufferlist bl;
@@ -105,7 +105,7 @@ protected:
   }
 
   void journal_setattr(pobject_t oid, const char *name, const void *value, size_t size, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.setattr(oid, name, value, size);
       bufferlist bl;
@@ -116,7 +116,7 @@ protected:
   }
 
   void journal_setattrs(pobject_t oid, map<string,bufferptr>& attrset, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.setattrs(oid, attrset);
       bufferlist bl;
@@ -127,7 +127,7 @@ protected:
   }
 
   void journal_rmattr(pobject_t oid, const char *name, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.rmattr(oid, name);
       bufferlist bl;
@@ -138,7 +138,7 @@ protected:
   }
 
   void journal_create_collection(coll_t cid, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.create_collection(cid);
       bufferlist bl;
@@ -149,7 +149,7 @@ protected:
   }
 
   void journal_destroy_collection(coll_t cid, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.remove_collection(cid);
       bufferlist bl;
@@ -160,7 +160,7 @@ protected:
   }
   
   void journal_collection_add(coll_t cid, pobject_t oid, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.collection_add(cid, oid);
       bufferlist bl;
@@ -171,7 +171,7 @@ protected:
   }
 
   void journal_collection_remove(coll_t cid, pobject_t oid, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.collection_remove(cid, oid);
       bufferlist bl;
@@ -182,7 +182,7 @@ protected:
   }
 
   void journal_collection_setattr(coll_t cid, const char *name, const void *value, size_t size, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.collection_setattr(cid, name, value, size);
       bufferlist bl;
@@ -193,7 +193,7 @@ protected:
   }
 
   void journal_collection_setattrs(coll_t cid, map<string,bufferptr>& aset, Context *onsafe) {
-    if (journal) {
+    if (journal && journal->is_writeable()) {
       Transaction t;
       t.collection_setattrs(cid, aset);
       bufferlist bl;

src/streamtest.cc

@@ -16,6 +16,7 @@
 
 #include <iostream>
 #include "ebofs/Ebofs.h"
+#include "osd/FakeStore.h"
 
 struct io {
   utime_t start, ack, commit;
@@ -95,12 +96,15 @@ int main(int argc, const char **argv)
   cout << "#dev " << filename
        << seconds << " seconds, " << bytes << " bytes per write" << std::endl;
 
-  Ebofs fs(filename, journal);
-  if (fs.mkfs() < 0) {
+  //ObjectStore *fs = new Ebofs(filename, journal);
+  ObjectStore *fs = new FakeStore(filename);
+
+  if (g_conf.mkfs && 
+      fs->mkfs() < 0) {
     cout << "mkfs failed" << std::endl;
     return -1;
   }
-  if (fs.mount() < 0) {
+  if (fs->mount() < 0) {
     cout << "mount failed" << std::endl;
     return -1;
   }
@@ -112,10 +116,10 @@ int main(int argc, const char **argv)
   //cout << "stop at " << end << std::endl;
   cout << "# offset\tack\tcommit" << std::endl;
   while (now < end) {
-    object_t oid(1,1);
+    pobject_t poid(0, 0, object_t(1, 1));
     utime_t start = now;
     set_start(pos, now);
-    fs.write(oid, pos, bytes, bl, new C_Commit(pos));
+    fs->write(poid, pos, bytes, bl, new C_Commit(pos));
     now = g_clock.now();
     set_ack(pos, now);
     pos += bytes;
@@ -131,7 +135,7 @@ int main(int argc, const char **argv)
     }
   }
 
-  fs.umount();
+  fs->umount();
 
 }