rough SessionMap, to replace ClientMap

src/mds/Capability.h

@@ -17,6 +17,7 @@
 #define __CAPABILITY_H
 
 #include "include/buffer.h"
+#include "include/xlist.h"
 
 #include <map>
 using namespace std;
@@ -24,7 +25,7 @@ using namespace std;
 #include "config.h"
 
 
-// definite caps
+// define caps
 #define CAP_FILE_RDCACHE   1    // client can safely cache reads
 #define CAP_FILE_RD        2    // client can read
 #define CAP_FILE_WR        4    // client can write
@@ -69,17 +70,20 @@ private:
   capseq_t last_sent, last_recv;
   
   bool suppress;
-
 public:
+  xlist<Capability*>::item xlist_item;
+
   Capability(int want=0, capseq_t s=0) :
     wanted_caps(want),
     last_sent(s),
     last_recv(s),
-    suppress(false) { 
+    suppress(false),
+    xlist_item(this) { 
   }
   Capability(Export& other) : 
     wanted_caps(other.wanted),
-    last_sent(0), last_recv(0) { 
+    last_sent(0), last_recv(0),
+    xlist_item(this) { 
     // issued vs pending
     if (other.issued & ~other.pending)
       issue(other.issued);

src/mds/Server.cc

@@ -140,9 +140,11 @@ void Server::handle_client_session(MClientSession *m)
 {
   dout(3) << "handle_client_session " << *m << " from " << m->get_source() << dendl;
   int from = m->get_source().num();
-  bool open = m->op == MClientSession::OP_REQUEST_OPEN;
+  bool open = false;
 
-  if (open) {
+  switch (m->op) {
+  case MClientSession::OP_REQUEST_OPEN;
+    open = true;
     if (mds->clientmap.have_session(from)) {
       dout(10) << "already open, dropping this req" << dendl;
       delete m;
@@ -154,7 +156,9 @@ void Server::handle_client_session(MClientSession *m)
       return;
     }
     mds->clientmap.add_opening(from);
-  } else {
+    break;
+
+  case MClientSession::OP_REQUEST_CLOSE:
     if (mds->clientmap.is_closing(from)) {
       dout(10) << "already closing, dropping this req" << dendl;
       delete m;
@@ -169,6 +173,10 @@ void Server::handle_client_session(MClientSession *m)
     assert(m->seq == mds->clientmap.get_push_seq(from));
 
     mds->clientmap.add_closing(from);
+    break;
+
+  default:
+    assert(0);
   }
 
   // journal it

src/mds/SessionMap.h

@@ -0,0 +1,121 @@
+// -*- 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) 2004-2006 Sage Weil <sage@newdream.net>
+ *
+ * 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.
+ * 
+ */
+
+#ifndef __MDS_SESSIONMAP_H
+#define __MDS_SESSIONMAP_H
+
+#include <set>
+using std::set;
+
+#include <ext/hash_map>
+using __gnu_cxx::hash_map;
+
+#include "include/Context.h"
+#include "include/xlist.h"
+#include "mdstypes.h"
+
+class CInode;
+
+class Session {
+  // -- state --
+public:
+  static const int STATE_OPENING = 1;
+  static const int STATE_OPEN = 2;
+  static const int STATE_CLOSING = 3;
+  static const int STATE_STALE = 4;   // ?
+  static const int STATE_RECONNECTING = 5;
+
+private:
+  int state;
+  utime_t last_alive;         // last alive
+public:
+  entity_inst_t inst;
+
+  // -- caps --
+private:
+  version_t cap_push_seq;     // cap push seq #
+  xlist<CInode*> cap_inodes;  // inodes with caps; front=most recently used
+
+public:
+  version_t inc_push_seq() { return ++cap_push_seq; }
+  version_t get_push_seq() const { return cap_push_seq; }
+
+  // -- completed requests --
+private:
+  set<tid_t> completed_requests;
+  map<tid_t, Context*> waiting_for_trim;
+
+public:
+  void add_completed_request(tid_t t) {
+    completed_requests.insert(t);
+  }
+  void trim_completed_requests(tid_t mintid) {
+    // trim
+    while (completed_requests.empty() && 
+	   (mintid == 0 || *completed_requests.begin() < mintid))
+      completed_requests.erase(completed_requests.begin());
+
+    // kick waiters
+    list<Context*> fls;
+    while (!waiting_for_trim.empty() &&
+	   (mintid == 0 || waiting_for_trim.begin()->first < mintid)) {
+      fls.push_back(waiting_for_trim.begin()->second);
+      waiting_for_trim.erase(waiting_for_trim.begin());
+    }
+    finish_contexts(fls);
+  }
+  void add_trim_waiter(metareqid_t ri, Context *c) {
+    waiting_for_trim[ri.tid] = c;
+  }
+  bool have_completed_request(metareqid_t ri) const {
+    return completed_requests.count(ri.tid);
+  }
+
+};
+
+class SessionMap {
+private:
+  MDS *mds;
+  hash_map<entity_name_t, Session> session_map;
+  version_t version, projected, committing, committed;
+
+public:
+  SessionMap(MDS *m) : mds(m), 
+		       version(0), projected(0), committing(0), committed(0) 
+  { }
+    
+  bool empty() { return session_map.empty(); }
+  
+  Session* get_session(entity_name_t w) {
+    if (session_map.count(w))
+      return &session_map[w];
+    return 0;
+  }
+  entity_inst_t& get_inst(entity_name_t w) {
+    assert(session_map.count(w));
+    return session_map[w].inst;
+  }
+
+  Session* add_session(entity_name_t w) {
+    return &session_map[w];
+  }
+  void remove_session(entity_name_t w) {
+    session_map.erase(w);
+  }
+  
+  
+};
+
+
+#endif