ceph/src/mds/SimpleLock.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15
  16 #ifndef CEPH_SIMPLELOCK_H
  17 #define CEPH_SIMPLELOCK_H
  18
  19 #include <boost/intrusive_ptr.hpp>
  20
  21 #include "MDSCacheObject.h"
  22 #include "MDSContext.h"
  23
  24 // -- lock types --
  25 // see CEPH_LOCK_*
  26
  27 inline const char *get_lock_type_name(int t) {
  28   switch (t) {
  29   case CEPH_LOCK_DN: return "dn";
  30   case CEPH_LOCK_DVERSION: return "dversion";
  31   case CEPH_LOCK_IVERSION: return "iversion";
  32   case CEPH_LOCK_IFILE: return "ifile";
  33   case CEPH_LOCK_IAUTH: return "iauth";
  34   case CEPH_LOCK_ILINK: return "ilink";
  35   case CEPH_LOCK_IDFT: return "idft";
  36   case CEPH_LOCK_INEST: return "inest";
  37   case CEPH_LOCK_IXATTR: return "ixattr";
  38   case CEPH_LOCK_ISNAP: return "isnap";
  39   case CEPH_LOCK_INO: return "ino";
  40   case CEPH_LOCK_IFLOCK: return "iflock";
  41   case CEPH_LOCK_IPOLICY: return "ipolicy";
  42   default: ceph_abort(); return 0;
  43   }
  44 }
  45
  46 #include "include/memory.h"
  47
  48 struct MutationImpl;
  49 typedef boost::intrusive_ptr<MutationImpl> MutationRef;
  50
  51 extern "C" {
  52 #include "locks.h"
  53 }
  54
  55
  56 #define CAP_ANY     0
  57 #define CAP_LONER   1
  58 #define CAP_XLOCKER 2
  59
  60 struct LockType {
  61   int type;
  62   const sm_t *sm;
  63
  64   explicit LockType(int t) : type(t) {
  65     switch (type) {
  66     case CEPH_LOCK_DN:
  67     case CEPH_LOCK_IAUTH:
  68     case CEPH_LOCK_ILINK:
  69     case CEPH_LOCK_IXATTR:
  70     case CEPH_LOCK_ISNAP:
  71     case CEPH_LOCK_IFLOCK:
  72     case CEPH_LOCK_IPOLICY:
  73       sm = &sm_simplelock;
  74       break;
  75     case CEPH_LOCK_IDFT:
  76     case CEPH_LOCK_INEST:
  77       sm = &sm_scatterlock;
  78       break;
  79     case CEPH_LOCK_IFILE:
  80       sm = &sm_filelock;
  81       break;
  82     case CEPH_LOCK_DVERSION:
  83     case CEPH_LOCK_IVERSION:
  84       sm = &sm_locallock;
  85       break;
  86     default:
  87       sm = 0;
  88     }
  89   }
  90
  91 };
  92
  93
  94 class SimpleLock {
  95 public:
  96   LockType *type;
  97
  98   const char *get_state_name(int n) const {
  99     switch (n) {
 100     case LOCK_UNDEF: return "UNDEF";
 101     case LOCK_SYNC: return "sync";
 102     case LOCK_LOCK: return "lock";
 103
 104     case LOCK_PREXLOCK: return "prexlock";
 105     case LOCK_XLOCK: return "xlock";
 106     case LOCK_XLOCKDONE: return "xlockdone";
 107     case LOCK_XLOCKSNAP: return "xlocksnap";
 108     case LOCK_LOCK_XLOCK: return "lock->xlock";
 109
 110     case LOCK_SYNC_LOCK: return "sync->lock";
 111     case LOCK_LOCK_SYNC: return "lock->sync";
 112     case LOCK_REMOTEXLOCK: return "remote_xlock";
 113     case LOCK_EXCL: return "excl";
 114     case LOCK_EXCL_SYNC: return "excl->sync";
 115     case LOCK_EXCL_LOCK: return "excl->lock";
 116     case LOCK_SYNC_EXCL: return "sync->excl";
 117     case LOCK_LOCK_EXCL: return "lock->excl";
 118
 119     case LOCK_XSYN: return "xsyn";
 120     case LOCK_XSYN_EXCL: return "xsyn->excl";
 121     case LOCK_EXCL_XSYN: return "excl->xsyn";
 122     case LOCK_XSYN_SYNC: return "xsyn->sync";
 123
 124     case LOCK_SYNC_MIX: return "sync->mix";
 125     case LOCK_SYNC_MIX2: return "sync->mix(2)";
 126     case LOCK_LOCK_TSYN: return "lock->tsyn";
 127
 128     case LOCK_MIX_LOCK: return "mix->lock";
 129     case LOCK_MIX_LOCK2: return "mix->lock(2)";
 130     case LOCK_MIX: return "mix";
 131     case LOCK_MIX_TSYN: return "mix->tsyn";
 132
 133     case LOCK_TSYN_MIX: return "tsyn->mix";
 134     case LOCK_TSYN_LOCK: return "tsyn->lock";
 135     case LOCK_TSYN: return "tsyn";
 136
 137     case LOCK_MIX_SYNC: return "mix->sync";
 138     case LOCK_MIX_SYNC2: return "mix->sync(2)";
 139     case LOCK_EXCL_MIX: return "excl->mix";
 140     case LOCK_MIX_EXCL: return "mix->excl";
 141
 142     case LOCK_PRE_SCAN: return "*->scan";
 143     case LOCK_SCAN: return "scan";
 144
 145     case LOCK_SNAP_SYNC: return "snap->sync";
 146
 147     default: ceph_abort(); return 0;
 148     }
 149   }
 150
 151
 152   // waiting
 153   static const uint64_t WAIT_RD          = (1<<0);  // to read
 154   static const uint64_t WAIT_WR          = (1<<1);  // to write
 155   static const uint64_t WAIT_XLOCK       = (1<<2);  // to xlock   (** dup)
 156   static const uint64_t WAIT_STABLE      = (1<<2);  // for a stable state
 157   static const uint64_t WAIT_REMOTEXLOCK = (1<<3);  // for a remote xlock
 158   static const int WAIT_BITS        = 4;
 159   static const uint64_t WAIT_ALL         = ((1<<WAIT_BITS)-1);
 160
 161
 162 protected:
 163   // parent (what i lock)
 164   MDSCacheObject *parent;
 165
 166   // lock state
 167   __s16 state;
 168
 169 private:
 170   __s16 num_rdlock;
 171   __s32 num_client_lease;
 172
 173   struct unstable_bits_t {
 174     set<__s32> gather_set;  // auth+rep.  >= 0 is mds, < 0 is client
 175
 176     // local state
 177     int num_wrlock, num_xlock;
 178     MutationRef xlock_by;
 179     client_t xlock_by_client;
 180     client_t excl_client;
 181
 182     bool empty() {
 183       return
 184         gather_set.empty() &&
 185         num_wrlock == 0 &&
 186         num_xlock == 0 &&
 187         xlock_by.get() == NULL &&
 188         xlock_by_client == -1 &&
 189         excl_client == -1;
 190     }
 191
 192     unstable_bits_t() : num_wrlock(0),
 193                         num_xlock(0),
 194                         xlock_by(),
 195                         xlock_by_client(-1),
 196                         excl_client(-1) {}
 197   };
 198
 199   mutable std::unique_ptr<unstable_bits_t> _unstable;
 200
 201   bool have_more() const { return _unstable ? true : false; }
 202   unstable_bits_t *more() const {
 203     if (!_unstable)
 204       _unstable.reset(new unstable_bits_t);
 205     return _unstable.get();
 206   }
 207   void try_clear_more() {
 208     if (_unstable && _unstable->empty()) {
 209       _unstable.reset();
 210     }
 211   }
 212
 213 public:
 214
 215   client_t get_excl_client() const {
 216     return have_more() ? more()->excl_client : -1;
 217   }
 218   void set_excl_client(client_t c) {
 219     if (c < 0 && !have_more())
 220       return;  // default is -1
 221     more()->excl_client = c;
 222   }
 223
 224   SimpleLock(MDSCacheObject *o, LockType *lt) :
 225     type(lt),
 226     parent(o),
 227     state(LOCK_SYNC),
 228     num_rdlock(0),
 229     num_client_lease(0)
 230   {}
 231   virtual ~SimpleLock() {}
 232
 233   virtual bool is_scatterlock() const {
 234     return false;
 235   }
 236   virtual bool is_locallock() const {
 237     return false;
 238   }
 239
 240   // parent
 241   MDSCacheObject *get_parent() { return parent; }
 242   int get_type() const { return type->type; }
 243   const sm_t* get_sm() const { return type->sm; }
 244
 245   int get_wait_shift() const {
 246     switch (get_type()) {
 247     case CEPH_LOCK_DN:       return 8;
 248     case CEPH_LOCK_DVERSION: return 8 + 1*SimpleLock::WAIT_BITS;
 249     case CEPH_LOCK_IAUTH:    return 8 + 2*SimpleLock::WAIT_BITS;
 250     case CEPH_LOCK_ILINK:    return 8 + 3*SimpleLock::WAIT_BITS;
 251     case CEPH_LOCK_IDFT:     return 8 + 4*SimpleLock::WAIT_BITS;
 252     case CEPH_LOCK_IFILE:    return 8 + 5*SimpleLock::WAIT_BITS;
 253     case CEPH_LOCK_IVERSION: return 8 + 6*SimpleLock::WAIT_BITS;
 254     case CEPH_LOCK_IXATTR:   return 8 + 7*SimpleLock::WAIT_BITS;
 255     case CEPH_LOCK_ISNAP:    return 8 + 8*SimpleLock::WAIT_BITS;
 256     case CEPH_LOCK_INEST:    return 8 + 9*SimpleLock::WAIT_BITS;
 257     case CEPH_LOCK_IFLOCK:   return 8 +10*SimpleLock::WAIT_BITS;
 258     case CEPH_LOCK_IPOLICY:  return 8 +11*SimpleLock::WAIT_BITS;
 259     default:
 260       ceph_abort();
 261     }
 262   }
 263
 264   int get_cap_shift() const {
 265     switch (get_type()) {
 266     case CEPH_LOCK_IAUTH: return CEPH_CAP_SAUTH;
 267     case CEPH_LOCK_ILINK: return CEPH_CAP_SLINK;
 268     case CEPH_LOCK_IFILE: return CEPH_CAP_SFILE;
 269     case CEPH_LOCK_IXATTR: return CEPH_CAP_SXATTR;
 270     default: return 0;
 271     }
 272   }
 273   int get_cap_mask() const {
 274     switch (get_type()) {
 275     case CEPH_LOCK_IFILE: return (1 << CEPH_CAP_FILE_BITS) - 1;
 276     default: return (1 << CEPH_CAP_SIMPLE_BITS) - 1;
 277     }
 278   }
 279
 280   struct ptr_lt {
 281     bool operator()(const SimpleLock* l, const SimpleLock* r) const {
 282       // first sort by object type (dn < inode)
 283       if (!(l->type->type > CEPH_LOCK_DN) && (r->type->type > CEPH_LOCK_DN)) return true;
 284       if ((l->type->type > CEPH_LOCK_DN) == (r->type->type > CEPH_LOCK_DN)) {
 285         // then sort by object
 286         if (l->parent->is_lt(r->parent)) return true;
 287         if (l->parent == r->parent) {
 288           // then sort by (inode) lock type
 289           if (l->type->type < r->type->type) return true;
 290         }
 291       }
 292       return false;
 293     }
 294   };
 295
 296   void decode_locked_state(bufferlist& bl) {
 297     parent->decode_lock_state(type->type, bl);
 298   }
 299   void encode_locked_state(bufferlist& bl) {
 300     parent->encode_lock_state(type->type, bl);
 301   }
 302   void finish_waiters(uint64_t mask, int r=0) {
 303     parent->finish_waiting(mask << get_wait_shift(), r);
 304   }
 305   void take_waiting(uint64_t mask, list<MDSInternalContextBase*>& ls) {
 306     parent->take_waiting(mask << get_wait_shift(), ls);
 307   }
 308   void add_waiter(uint64_t mask, MDSInternalContextBase *c) {
 309     parent->add_waiter((mask << get_wait_shift()) | MDSCacheObject::WAIT_ORDERED, c);
 310   }
 311   bool is_waiter_for(uint64_t mask) const {
 312     return parent->is_waiter_for(mask << get_wait_shift());
 313   }
 314
 315
 316
 317   // state
 318   int get_state() const { return state; }
 319   int set_state(int s) {
 320     state = s;
 321     //assert(!is_stable() || gather_set.size() == 0);  // gather should be empty in stable states.
 322     return s;
 323   }
 324   void set_state_rejoin(int s, list<MDSInternalContextBase*>& waiters) {
 325     if (!is_stable() && get_parent()->is_auth()) {
 326       state = s;
 327       get_parent()->auth_unpin(this);
 328     } else {
 329       state = s;
 330     }
 331     if (is_stable())
 332       take_waiting(SimpleLock::WAIT_ALL, waiters);
 333   }
 334
 335   bool is_stable() const {
 336     return get_sm()->states[state].next == 0;
 337   }
 338   bool is_unstable_and_locked() const {
 339     if (is_stable())
 340       return false;
 341     return is_rdlocked() || is_wrlocked() || is_xlocked();
 342   }
 343   int get_next_state() {
 344     return get_sm()->states[state].next;
 345   }
 346
 347
 348   bool is_sync_and_unlocked() const {
 349     return
 350       get_state() == LOCK_SYNC &&
 351       !is_rdlocked() &&
 352       !is_leased() &&
 353       !is_wrlocked() &&
 354       !is_xlocked();
 355   }
 356
 357
 358   /*
 359   bool fw_rdlock_to_auth() {
 360     return get_sm()->states[state].can_rdlock == FW;
 361   }
 362   */
 363   bool req_rdlock_from_auth() {
 364     return get_sm()->states[state].can_rdlock == REQ;
 365   }
 366
 367   // gather set
 368   static set<int32_t> empty_gather_set;
 369
 370   // int32_t: <0 is client, >=0 is MDS rank
 371   const set<int32_t>& get_gather_set() const {
 372     return have_more() ? more()->gather_set : empty_gather_set;
 373   }
 374
 375   void init_gather() {
 376     for (compact_map<mds_rank_t,unsigned>::iterator p = parent->replicas_begin();
 377          p != parent->replicas_end();
 378          ++p)
 379       more()->gather_set.insert(p->first);
 380   }
 381   bool is_gathering() const {
 382     return have_more() && !more()->gather_set.empty();
 383   }
 384   bool is_gathering(int32_t i) const {
 385     return have_more() && more()->gather_set.count(i);
 386   }
 387   void clear_gather() {
 388     if (have_more())
 389       more()->gather_set.clear();
 390   }
 391   void remove_gather(int32_t i) {
 392     if (have_more())
 393       more()->gather_set.erase(i);
 394   }
 395
 396
 397
 398   virtual bool is_dirty() const { return false; }
 399   virtual bool is_stale() const { return false; }
 400   virtual bool is_flushing() const { return false; }
 401   virtual bool is_flushed() const { return false; }
 402   virtual void clear_flushed() { }
 403
 404   // can_*
 405   bool can_lease(client_t client) const {
 406     return get_sm()->states[state].can_lease == ANY ||
 407       (get_sm()->states[state].can_lease == AUTH && parent->is_auth()) ||
 408       (get_sm()->states[state].can_lease == XCL && client >= 0 && get_xlock_by_client() == client);
 409   }
 410   bool can_read(client_t client) const {
 411     return get_sm()->states[state].can_read == ANY ||
 412       (get_sm()->states[state].can_read == AUTH && parent->is_auth()) ||
 413       (get_sm()->states[state].can_read == XCL && client >= 0 && get_xlock_by_client() == client);
 414   }
 415   bool can_read_projected(client_t client) const {
 416     return get_sm()->states[state].can_read_projected == ANY ||
 417       (get_sm()->states[state].can_read_projected == AUTH && parent->is_auth()) ||
 418       (get_sm()->states[state].can_read_projected == XCL && client >= 0 && get_xlock_by_client() == client);
 419   }
 420   bool can_rdlock(client_t client) const {
 421     return get_sm()->states[state].can_rdlock == ANY ||
 422       (get_sm()->states[state].can_rdlock == AUTH && parent->is_auth()) ||
 423       (get_sm()->states[state].can_rdlock == XCL && client >= 0 && get_xlock_by_client() == client);
 424   }
 425   bool can_wrlock(client_t client) const {
 426     return get_sm()->states[state].can_wrlock == ANY ||
 427       (get_sm()->states[state].can_wrlock == AUTH && parent->is_auth()) ||
 428       (get_sm()->states[state].can_wrlock == XCL && client >= 0 && (get_xlock_by_client() == client ||
 429                                                                     get_excl_client() == client));
 430   }
 431   bool can_force_wrlock(client_t client) const {
 432     return get_sm()->states[state].can_force_wrlock == ANY ||
 433       (get_sm()->states[state].can_force_wrlock == AUTH && parent->is_auth()) ||
 434       (get_sm()->states[state].can_force_wrlock == XCL && client >= 0 && (get_xlock_by_client() == client ||
 435                                                                           get_excl_client() == client));
 436   }
 437   bool can_xlock(client_t client) const {
 438     return get_sm()->states[state].can_xlock == ANY ||
 439       (get_sm()->states[state].can_xlock == AUTH && parent->is_auth()) ||
 440       (get_sm()->states[state].can_xlock == XCL && client >= 0 && get_xlock_by_client() == client);
 441   }
 442
 443   // rdlock
 444   bool is_rdlocked() const { return num_rdlock > 0; }
 445   int get_rdlock() {
 446     if (!num_rdlock)
 447       parent->get(MDSCacheObject::PIN_LOCK);
 448     return ++num_rdlock;
 449   }
 450   int put_rdlock() {
 451     assert(num_rdlock>0);
 452     --num_rdlock;
 453     if (num_rdlock == 0)
 454       parent->put(MDSCacheObject::PIN_LOCK);
 455     return num_rdlock;
 456   }
 457   int get_num_rdlocks() const {
 458     return num_rdlock;
 459   }
 460
 461   // wrlock
 462   void get_wrlock(bool force=false) {
 463     //assert(can_wrlock() || force);
 464     if (more()->num_wrlock == 0)
 465       parent->get(MDSCacheObject::PIN_LOCK);
 466     ++more()->num_wrlock;
 467   }
 468   void put_wrlock() {
 469     --more()->num_wrlock;
 470     if (more()->num_wrlock == 0) {
 471       parent->put(MDSCacheObject::PIN_LOCK);
 472       try_clear_more();
 473     }
 474   }
 475   bool is_wrlocked() const {
 476     return have_more() && more()->num_wrlock > 0;
 477   }
 478   int get_num_wrlocks() const {
 479     return have_more() ? more()->num_wrlock : 0;
 480   }
 481
 482   // xlock
 483   void get_xlock(MutationRef who, client_t client) {
 484     assert(get_xlock_by() == MutationRef());
 485     assert(state == LOCK_XLOCK || is_locallock() ||
 486            state == LOCK_LOCK /* if we are a slave */);
 487     parent->get(MDSCacheObject::PIN_LOCK);
 488     more()->num_xlock++;
 489     more()->xlock_by = who;
 490     more()->xlock_by_client = client;
 491   }
 492   void set_xlock_done() {
 493     assert(more()->xlock_by);
 494     assert(state == LOCK_XLOCK || is_locallock() ||
 495            state == LOCK_LOCK /* if we are a slave */);
 496     if (!is_locallock())
 497       state = LOCK_XLOCKDONE;
 498     more()->xlock_by.reset();
 499   }
 500   void put_xlock() {
 501     assert(state == LOCK_XLOCK || state == LOCK_XLOCKDONE ||
 502            state == LOCK_XLOCKSNAP || is_locallock() ||
 503            state == LOCK_LOCK /* if we are a master of a slave */);
 504     --more()->num_xlock;
 505     parent->put(MDSCacheObject::PIN_LOCK);
 506     if (more()->num_xlock == 0) {
 507       more()->xlock_by.reset();
 508       more()->xlock_by_client = -1;
 509       try_clear_more();
 510     }
 511   }
 512   bool is_xlocked() const {
 513     return have_more() && more()->num_xlock > 0;
 514   }
 515   int get_num_xlocks() const {
 516     return have_more() ? more()->num_xlock : 0;
 517   }
 518   client_t get_xlock_by_client() const {
 519     return have_more() ? more()->xlock_by_client : -1;
 520   }
 521   bool is_xlocked_by_client(client_t c) const {
 522     return have_more() ? more()->xlock_by_client == c : false;
 523   }
 524   MutationRef get_xlock_by() const {
 525     return have_more() ? more()->xlock_by : MutationRef();
 526   }
 527
 528   // lease
 529   void get_client_lease() {
 530     num_client_lease++;
 531   }
 532   void put_client_lease() {
 533     assert(num_client_lease > 0);
 534     num_client_lease--;
 535     if (num_client_lease == 0) {
 536       try_clear_more();
 537     }
 538   }
 539   bool is_leased() const {
 540     return num_client_lease > 0;
 541   }
 542   int get_num_client_lease() const {
 543     return num_client_lease;
 544   }
 545
 546   bool is_used() const {
 547     return is_xlocked() || is_rdlocked() || is_wrlocked() || num_client_lease;
 548   }
 549
 550   // encode/decode
 551   void encode(bufferlist& bl) const {
 552     ENCODE_START(2, 2, bl);
 553     ::encode(state, bl);
 554     if (have_more())
 555       ::encode(more()->gather_set, bl);
 556     else
 557       ::encode(empty_gather_set, bl);
 558     ENCODE_FINISH(bl);
 559   }
 560   void decode(bufferlist::iterator& p) {
 561     DECODE_START(2, p);
 562     ::decode(state, p);
 563     set<__s32> g;
 564     ::decode(g, p);
 565     if (!g.empty())
 566       more()->gather_set.swap(g);
 567     DECODE_FINISH(p);
 568   }
 569   void encode_state_for_replica(bufferlist& bl) const {
 570     __s16 s = get_replica_state();
 571     ::encode(s, bl);
 572   }
 573   void decode_state(bufferlist::iterator& p, bool is_new=true) {
 574     __s16 s;
 575     ::decode(s, p);
 576     if (is_new)
 577       state = s;
 578   }
 579   void decode_state_rejoin(bufferlist::iterator& p, list<MDSInternalContextBase*>& waiters) {
 580     __s16 s;
 581     ::decode(s, p);
 582     set_state_rejoin(s, waiters);
 583   }
 584
 585
 586   // caps
 587   bool is_loner_mode() const {
 588     return get_sm()->states[state].loner;
 589   }
 590   int gcaps_allowed_ever() const {
 591     return parent->is_auth() ? get_sm()->allowed_ever_auth : get_sm()->allowed_ever_replica;
 592   }
 593   int gcaps_allowed(int who, int s=-1) const {
 594     if (s < 0) s = state;
 595     if (parent->is_auth()) {
 596       if (get_xlock_by_client() >= 0 && who == CAP_XLOCKER)
 597         return get_sm()->states[s].xlocker_caps | get_sm()->states[s].caps; // xlocker always gets more
 598       else if (is_loner_mode() && who == CAP_ANY)
 599         return get_sm()->states[s].caps;
 600       else
 601         return get_sm()->states[s].loner_caps | get_sm()->states[s].caps;  // loner always gets more
 602     } else
 603       return get_sm()->states[s].replica_caps;
 604   }
 605   int gcaps_careful() const {
 606     if (get_num_wrlocks())
 607       return get_sm()->careful;
 608     return 0;
 609   }
 610
 611
 612   int gcaps_xlocker_mask(client_t client) const {
 613     if (client == get_xlock_by_client())
 614       return type->type == CEPH_LOCK_IFILE ? 0xf : (CEPH_CAP_GSHARED|CEPH_CAP_GEXCL);
 615     return 0;
 616   }
 617
 618   // simplelock specifics
 619   int get_replica_state() const {
 620     return get_sm()->states[state].replica_state;
 621   }
 622   void export_twiddle() {
 623     clear_gather();
 624     state = get_replica_state();
 625   }
 626
 627   /** replicate_relax
 628    * called on first replica creation.
 629    */
 630   void replicate_relax() {
 631     assert(parent->is_auth());
 632     assert(!parent->is_replicated());
 633     if (state == LOCK_LOCK && !is_used())
 634       state = LOCK_SYNC;
 635   }
 636   bool remove_replica(int from) {
 637     if (is_gathering(from)) {
 638       remove_gather(from);
 639       if (!is_gathering())
 640         return true;
 641     }
 642     return false;
 643   }
 644   bool do_import(int from, int to) {
 645     if (!is_stable()) {
 646       remove_gather(from);
 647       remove_gather(to);
 648       if (!is_gathering())
 649         return true;
 650     }
 651     if (!is_stable() && !is_gathering())
 652       return true;
 653     return false;
 654   }
 655
 656   void _print(ostream& out) const {
 657     out << get_lock_type_name(get_type()) << " ";
 658     out << get_state_name(get_state());
 659     if (!get_gather_set().empty())
 660       out << " g=" << get_gather_set();
 661     if (num_client_lease)
 662       out << " l=" << num_client_lease;
 663     if (is_rdlocked())
 664       out << " r=" << get_num_rdlocks();
 665     if (is_wrlocked())
 666       out << " w=" << get_num_wrlocks();
 667     if (is_xlocked()) {
 668       out << " x=" << get_num_xlocks();
 669       if (get_xlock_by())
 670         out << " by " << get_xlock_by();
 671     }
 672     /*if (is_stable())
 673       out << " stable";
 674     else
 675       out << " unstable";
 676     */
 677   }
 678
 679   /**
 680    * Write bare values (caller must be in an object section)
 681    * to formatter, or nothing if is_sync_and_unlocked.
 682    */
 683   void dump(Formatter *f) const;
 684
 685   virtual void print(ostream& out) const {
 686     out << "(";
 687     _print(out);
 688     out << ")";
 689   }
 690 };
 691 WRITE_CLASS_ENCODER(SimpleLock)
 692
 693 inline ostream& operator<<(ostream& out, const SimpleLock& l)
 694 {
 695   l.print(out);
 696   return out;
 697 }
 698
 699
 700 #endif