*** empty log message ***

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@604 29311d96-e01e-0410-9327-a35deaab8ce9

ceph/config.cc

@@ -114,7 +114,7 @@ md_config_t g_conf = {
   // --- osd ---
   osd_pg_bits: 6,
   osd_max_rep: 4,
-  osd_maxthreads: 10,    // 0 == no threading
+  osd_maxthreads: 2,    // 0 == no threading
   osd_mkfs: false,
   osd_fake_sync: false,
   
@@ -325,8 +325,8 @@ void parse_config_options(vector<char*>& args)
 
 	else if (strcmp(args[i], "--fakestore") == 0) {
 	  g_conf.ebofs = 0;
-	  g_conf.osd_pg_bits = 5;
-	  g_conf.osd_maxthreads = 1;   // fucking hell
+	  //g_conf.osd_pg_bits = 5;
+	  //g_conf.osd_maxthreads = 1;   // fucking hell
 	}
 	else if (strcmp(args[i], "--fakestore_fsync") == 0) 
 	  g_conf.fakestore_fsync = atoi(args[++i]);

ceph/ebofs/Allocator.cc

@@ -221,6 +221,7 @@ int Allocator::_release_loner(Extent& ex)
   assert(ex.length > 0);
   int b = pick_bucket(ex.length);
   fs->free_tab[b]->insert(ex.start, ex.length);
+  fs->free_blocks += ex.length;
   return 0;
 }
 
@@ -232,7 +233,6 @@ int Allocator::_release_merge(Extent& orig)
 {
   dout(15) << "_release_merge " << orig << endl;
   assert(orig.length > 0);
-  fs->free_blocks += orig.length;
 
   Extent newex = orig;
   
@@ -246,6 +246,7 @@ int Allocator::_release_merge(Extent& orig)
 	  newex.length += cursor.current().value;
 	  
 	  // remove it
+	  fs->free_blocks -= cursor.current().value;
 	  fs->free_tab[b]->remove( cursor.current().key );
 	  break;
 	}
@@ -260,8 +261,9 @@ int Allocator::_release_merge(Extent& orig)
 	  // merge
 	  newex.start = cursor.current().key;
 	  newex.length += cursor.current().value;
-	  
+
 	  // remove it
+	  fs->free_blocks -= cursor.current().value;
 	  fs->free_tab[b]->remove( cursor.current().key );
 	  break;
 	}

ceph/mds/MDS.cc

@@ -257,15 +257,16 @@ void MDS::handle_shutdown_finish(Message *m)
   if (did_shut_down.size() == (unsigned)mdcluster->get_num_mds()) {
 	// MDS's all ready to shut down!
 
+	dout(1) << "sending shutdown to remaining MDSs, OSDs" << endl;
 	for (int i=1; i<g_conf.num_mds; i++) {
-	  dout(1) << "sending shutdown to mds" << i << endl;
+	  dout(10) << "sending shutdown to mds" << i << endl;
 	  messenger->send_message(new MGenericMessage(MSG_SHUTDOWN),
 							  MSG_ADDR_MDS(i), 0, 0);
 	}
 
 	// shut down osd's
 	for (int i=0; i<g_conf.num_osd; i++) {
-	  dout(1) << "sending shutdown to osd" << i << endl;
+	  dout(10) << "sending shutdown to osd" << i << endl;
 	  messenger->send_message(new MGenericMessage(MSG_SHUTDOWN),
 							  MSG_ADDR_OSD(i), 0, 0);
 	}

ceph/osd/OSD.cc

@@ -243,8 +243,8 @@ void OSD::unlock_object(object_t oid)
 	// someone is in line
 	list<Cond*>& ls = object_lock_waiters[oid];
 	Cond *c = ls.front();
-	dout(15) << "unlock_object " << hex << oid << dec << " waking up next guy " << c << endl;
 	ls.pop_front();
+	dout(15) << "unlock_object " << hex << oid << dec << " waking up next guy " << c << endl;
 	if (ls.empty()) 
 	  object_lock_waiters.erase(oid);
 	c->Signal();
@@ -1996,25 +1996,27 @@ void OSD::handle_op(MOSDOp *op)
 	}  
   }
 
-  // queue op
-  object_t oid = op->get_oid();
-  op_queue[oid].push_back(op);
-  pending_ops++;
-
   if (g_conf.osd_maxthreads < 1) {
-	// do it now
-	osd_lock.Unlock();
-	{
-	  dequeue_op(oid);
-	}
-	osd_lock.Lock();
+	do_op(op); // do it now
   } else {
-	threadpool->put_op(oid);
+	enqueue_op(op->get_oid(), op); 	// queue for worker threads
   }
 }
 
 
 /*
+ * enqueue called with osd_lock held
+ */
+void OSD::enqueue_op(object_t oid, MOSDOp *op)
+{
+  op_queue[oid].push_back(op);
+  pending_ops++;
+  
+  threadpool->put_op(oid);
+}
+
+/*
+ * dequeue called in worker thread, without osd_lock
  */
 void OSD::dequeue_op(object_t oid)
 {
@@ -2030,29 +2032,46 @@ void OSD::dequeue_op(object_t oid)
 	assert(!ls.empty());
 	op = ls.front();
 	ls.pop_front();
+
+	dout(10) << "dequeue_op " << hex << oid << dec << " op " << op << ", " << ls.size() << " / " << (pending_ops-1) << " more pending" << endl;
+	
 	if (ls.empty())
 	  op_queue.erase(oid);
   }
   osd_lock.Unlock();
   
-  dout(10) << "dequeue_op " << hex << oid << dec << " op " << op << ", " << (pending_ops-1) << " others pending" << endl;
+  // do it
   do_op(op);
+
+  // unlock
+  unlock_object(oid);
+  
+  // finish
+  dout(10) << "finish op " << op << endl;
+  osd_lock.Lock();
+  {
+	assert(pending_ops > 0);
+	pending_ops--;
+	if (pending_ops == 0 && waiting_for_no_ops)
+	  no_pending_ops.Signal();
+  }
+  osd_lock.Unlock();
 }
 
 
 
 /*
- * called asynchronously by worker thread after items are dequeued
+ * do an op
+ *
+ * object lock may be held (if multithreaded)
+ * osd_lock NOT held.
  */
 void OSD::do_op(MOSDOp *op) 
 {
-  dout(10) << "do_op " << op << endl;
+  dout(10) << "do_op " << op << " " << op->get_op() << endl;
 
   logger->inc("op");
 
-  object_t oid = op->get_oid();
-  //lock_object(oid);  // done by dequeue_op now
-
   // replication ops?
   if (OSD_OP_IS_REP(op->get_op())) {
 	// replication/recovery
@@ -2096,17 +2115,6 @@ void OSD::do_op(MOSDOp *op)
 	}
   }
 
-  unlock_object(oid);
-  
-  //dout(12) << "finish op " << op << endl;
-
-  // finish
-  osd_lock.Lock();
-  assert(pending_ops > 0);
-  pending_ops--;
-  if (pending_ops == 0 && waiting_for_no_ops)
-	no_pending_ops.Signal();
-  osd_lock.Unlock();
 }
 
 void OSD::wait_for_no_ops()

ceph/osd/OSD.h

@@ -111,6 +111,7 @@ class OSD : public Dispatcher {
   void do_op(class MOSDOp *m);
 
  public:
+  void enqueue_op(object_t oid, MOSDOp *op);
   void dequeue_op(object_t oid);
   static void static_dequeueop(OSD *o, object_t oid) {
 	o->dequeue_op(oid);