Back to home page

OSCL-LXR
 
 

     

0001 // SPDX-License-Identifier: GPL-2.0-or-later
0002 /*
0003  * dlmdomain.c
0004  *
0005  * defines domain join / leave apis
0006  *
0007  * Copyright (C) 2004 Oracle.  All rights reserved.
0008  */
0009 
0010 #include <linux/module.h>
0011 #include <linux/types.h>
0012 #include <linux/slab.h>
0013 #include <linux/highmem.h>
0014 #include <linux/init.h>
0015 #include <linux/spinlock.h>
0016 #include <linux/delay.h>
0017 #include <linux/err.h>
0018 #include <linux/debugfs.h>
0019 #include <linux/sched/signal.h>
0020 
0021 #include "../cluster/heartbeat.h"
0022 #include "../cluster/nodemanager.h"
0023 #include "../cluster/tcp.h"
0024 
0025 #include "dlmapi.h"
0026 #include "dlmcommon.h"
0027 #include "dlmdomain.h"
0028 #include "dlmdebug.h"
0029 
0030 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_DOMAIN)
0031 #include "../cluster/masklog.h"
0032 
0033 /*
0034  * ocfs2 node maps are array of long int, which limits to send them freely
0035  * across the wire due to endianness issues. To workaround this, we convert
0036  * long ints to byte arrays. Following 3 routines are helper functions to
0037  * set/test/copy bits within those array of bytes
0038  */
0039 static inline void byte_set_bit(u8 nr, u8 map[])
0040 {
0041     map[nr >> 3] |= (1UL << (nr & 7));
0042 }
0043 
0044 static inline int byte_test_bit(u8 nr, u8 map[])
0045 {
0046     return ((1UL << (nr & 7)) & (map[nr >> 3])) != 0;
0047 }
0048 
0049 static inline void byte_copymap(u8 dmap[], unsigned long smap[],
0050             unsigned int sz)
0051 {
0052     unsigned int nn;
0053 
0054     if (!sz)
0055         return;
0056 
0057     memset(dmap, 0, ((sz + 7) >> 3));
0058     for (nn = 0 ; nn < sz; nn++)
0059         if (test_bit(nn, smap))
0060             byte_set_bit(nn, dmap);
0061 }
0062 
0063 static void dlm_free_pagevec(void **vec, int pages)
0064 {
0065     while (pages--)
0066         free_page((unsigned long)vec[pages]);
0067     kfree(vec);
0068 }
0069 
0070 static void **dlm_alloc_pagevec(int pages)
0071 {
0072     void **vec = kmalloc_array(pages, sizeof(void *), GFP_KERNEL);
0073     int i;
0074 
0075     if (!vec)
0076         return NULL;
0077 
0078     for (i = 0; i < pages; i++)
0079         if (!(vec[i] = (void *)__get_free_page(GFP_KERNEL)))
0080             goto out_free;
0081 
0082     mlog(0, "Allocated DLM hash pagevec; %d pages (%lu expected), %lu buckets per page\n",
0083          pages, (unsigned long)DLM_HASH_PAGES,
0084          (unsigned long)DLM_BUCKETS_PER_PAGE);
0085     return vec;
0086 out_free:
0087     dlm_free_pagevec(vec, i);
0088     return NULL;
0089 }
0090 
0091 /*
0092  *
0093  * spinlock lock ordering: if multiple locks are needed, obey this ordering:
0094  *    dlm_domain_lock
0095  *    struct dlm_ctxt->spinlock
0096  *    struct dlm_lock_resource->spinlock
0097  *    struct dlm_ctxt->master_lock
0098  *    struct dlm_ctxt->ast_lock
0099  *    dlm_master_list_entry->spinlock
0100  *    dlm_lock->spinlock
0101  *
0102  */
0103 
0104 DEFINE_SPINLOCK(dlm_domain_lock);
0105 LIST_HEAD(dlm_domains);
0106 static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);
0107 
0108 /*
0109  * The supported protocol version for DLM communication.  Running domains
0110  * will have a negotiated version with the same major number and a minor
0111  * number equal or smaller.  The dlm_ctxt->dlm_locking_proto field should
0112  * be used to determine what a running domain is actually using.
0113  *
0114  * New in version 1.1:
0115  *  - Message DLM_QUERY_REGION added to support global heartbeat
0116  *  - Message DLM_QUERY_NODEINFO added to allow online node removes
0117  * New in version 1.2:
0118  *  - Message DLM_BEGIN_EXIT_DOMAIN_MSG added to mark start of exit domain
0119  * New in version 1.3:
0120  *  - Message DLM_DEREF_LOCKRES_DONE added to inform non-master that the
0121  *    refmap is cleared
0122  */
0123 static const struct dlm_protocol_version dlm_protocol = {
0124     .pv_major = 1,
0125     .pv_minor = 3,
0126 };
0127 
0128 #define DLM_DOMAIN_BACKOFF_MS 200
0129 
0130 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
0131                   void **ret_data);
0132 static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
0133                      void **ret_data);
0134 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
0135                    void **ret_data);
0136 static int dlm_query_region_handler(struct o2net_msg *msg, u32 len,
0137                     void *data, void **ret_data);
0138 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
0139                    void **ret_data);
0140 static int dlm_protocol_compare(struct dlm_protocol_version *existing,
0141                 struct dlm_protocol_version *request);
0142 
0143 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);
0144 
0145 void __dlm_unhash_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
0146 {
0147     if (hlist_unhashed(&res->hash_node))
0148         return;
0149 
0150     mlog(0, "%s: Unhash res %.*s\n", dlm->name, res->lockname.len,
0151          res->lockname.name);
0152     hlist_del_init(&res->hash_node);
0153     dlm_lockres_put(res);
0154 }
0155 
0156 void __dlm_insert_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
0157 {
0158     struct hlist_head *bucket;
0159 
0160     assert_spin_locked(&dlm->spinlock);
0161 
0162     bucket = dlm_lockres_hash(dlm, res->lockname.hash);
0163 
0164     /* get a reference for our hashtable */
0165     dlm_lockres_get(res);
0166 
0167     hlist_add_head(&res->hash_node, bucket);
0168 
0169     mlog(0, "%s: Hash res %.*s\n", dlm->name, res->lockname.len,
0170          res->lockname.name);
0171 }
0172 
0173 struct dlm_lock_resource * __dlm_lookup_lockres_full(struct dlm_ctxt *dlm,
0174                              const char *name,
0175                              unsigned int len,
0176                              unsigned int hash)
0177 {
0178     struct hlist_head *bucket;
0179     struct dlm_lock_resource *res;
0180 
0181     mlog(0, "%.*s\n", len, name);
0182 
0183     assert_spin_locked(&dlm->spinlock);
0184 
0185     bucket = dlm_lockres_hash(dlm, hash);
0186 
0187     hlist_for_each_entry(res, bucket, hash_node) {
0188         if (res->lockname.name[0] != name[0])
0189             continue;
0190         if (unlikely(res->lockname.len != len))
0191             continue;
0192         if (memcmp(res->lockname.name + 1, name + 1, len - 1))
0193             continue;
0194         dlm_lockres_get(res);
0195         return res;
0196     }
0197     return NULL;
0198 }
0199 
0200 /* intended to be called by functions which do not care about lock
0201  * resources which are being purged (most net _handler functions).
0202  * this will return NULL for any lock resource which is found but
0203  * currently in the process of dropping its mastery reference.
0204  * use __dlm_lookup_lockres_full when you need the lock resource
0205  * regardless (e.g. dlm_get_lock_resource) */
0206 struct dlm_lock_resource * __dlm_lookup_lockres(struct dlm_ctxt *dlm,
0207                         const char *name,
0208                         unsigned int len,
0209                         unsigned int hash)
0210 {
0211     struct dlm_lock_resource *res = NULL;
0212 
0213     mlog(0, "%.*s\n", len, name);
0214 
0215     assert_spin_locked(&dlm->spinlock);
0216 
0217     res = __dlm_lookup_lockres_full(dlm, name, len, hash);
0218     if (res) {
0219         spin_lock(&res->spinlock);
0220         if (res->state & DLM_LOCK_RES_DROPPING_REF) {
0221             spin_unlock(&res->spinlock);
0222             dlm_lockres_put(res);
0223             return NULL;
0224         }
0225         spin_unlock(&res->spinlock);
0226     }
0227 
0228     return res;
0229 }
0230 
0231 struct dlm_lock_resource * dlm_lookup_lockres(struct dlm_ctxt *dlm,
0232                     const char *name,
0233                     unsigned int len)
0234 {
0235     struct dlm_lock_resource *res;
0236     unsigned int hash = dlm_lockid_hash(name, len);
0237 
0238     spin_lock(&dlm->spinlock);
0239     res = __dlm_lookup_lockres(dlm, name, len, hash);
0240     spin_unlock(&dlm->spinlock);
0241     return res;
0242 }
0243 
0244 static struct dlm_ctxt * __dlm_lookup_domain_full(const char *domain, int len)
0245 {
0246     struct dlm_ctxt *tmp;
0247 
0248     assert_spin_locked(&dlm_domain_lock);
0249 
0250     /* tmp->name here is always NULL terminated,
0251      * but domain may not be! */
0252     list_for_each_entry(tmp, &dlm_domains, list) {
0253         if (strlen(tmp->name) == len &&
0254             memcmp(tmp->name, domain, len)==0)
0255             return tmp;
0256     }
0257 
0258     return NULL;
0259 }
0260 
0261 /* For null terminated domain strings ONLY */
0262 static struct dlm_ctxt * __dlm_lookup_domain(const char *domain)
0263 {
0264     assert_spin_locked(&dlm_domain_lock);
0265 
0266     return __dlm_lookup_domain_full(domain, strlen(domain));
0267 }
0268 
0269 
0270 /* returns true on one of two conditions:
0271  * 1) the domain does not exist
0272  * 2) the domain exists and it's state is "joined" */
0273 static int dlm_wait_on_domain_helper(const char *domain)
0274 {
0275     int ret = 0;
0276     struct dlm_ctxt *tmp = NULL;
0277 
0278     spin_lock(&dlm_domain_lock);
0279 
0280     tmp = __dlm_lookup_domain(domain);
0281     if (!tmp)
0282         ret = 1;
0283     else if (tmp->dlm_state == DLM_CTXT_JOINED)
0284         ret = 1;
0285 
0286     spin_unlock(&dlm_domain_lock);
0287     return ret;
0288 }
0289 
0290 static void dlm_free_ctxt_mem(struct dlm_ctxt *dlm)
0291 {
0292     dlm_destroy_debugfs_subroot(dlm);
0293 
0294     if (dlm->lockres_hash)
0295         dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES);
0296 
0297     if (dlm->master_hash)
0298         dlm_free_pagevec((void **)dlm->master_hash, DLM_HASH_PAGES);
0299 
0300     kfree(dlm->name);
0301     kfree(dlm);
0302 }
0303 
0304 /* A little strange - this function will be called while holding
0305  * dlm_domain_lock and is expected to be holding it on the way out. We
0306  * will however drop and reacquire it multiple times */
0307 static void dlm_ctxt_release(struct kref *kref)
0308 {
0309     struct dlm_ctxt *dlm;
0310 
0311     dlm = container_of(kref, struct dlm_ctxt, dlm_refs);
0312 
0313     BUG_ON(dlm->num_joins);
0314     BUG_ON(dlm->dlm_state == DLM_CTXT_JOINED);
0315 
0316     /* we may still be in the list if we hit an error during join. */
0317     list_del_init(&dlm->list);
0318 
0319     spin_unlock(&dlm_domain_lock);
0320 
0321     mlog(0, "freeing memory from domain %s\n", dlm->name);
0322 
0323     wake_up(&dlm_domain_events);
0324 
0325     dlm_free_ctxt_mem(dlm);
0326 
0327     spin_lock(&dlm_domain_lock);
0328 }
0329 
0330 void dlm_put(struct dlm_ctxt *dlm)
0331 {
0332     spin_lock(&dlm_domain_lock);
0333     kref_put(&dlm->dlm_refs, dlm_ctxt_release);
0334     spin_unlock(&dlm_domain_lock);
0335 }
0336 
0337 static void __dlm_get(struct dlm_ctxt *dlm)
0338 {
0339     kref_get(&dlm->dlm_refs);
0340 }
0341 
0342 /* given a questionable reference to a dlm object, gets a reference if
0343  * it can find it in the list, otherwise returns NULL in which case
0344  * you shouldn't trust your pointer. */
0345 struct dlm_ctxt *dlm_grab(struct dlm_ctxt *dlm)
0346 {
0347     struct dlm_ctxt *target;
0348     struct dlm_ctxt *ret = NULL;
0349 
0350     spin_lock(&dlm_domain_lock);
0351 
0352     list_for_each_entry(target, &dlm_domains, list) {
0353         if (target == dlm) {
0354             __dlm_get(target);
0355             ret = target;
0356             break;
0357         }
0358     }
0359 
0360     spin_unlock(&dlm_domain_lock);
0361 
0362     return ret;
0363 }
0364 
0365 int dlm_domain_fully_joined(struct dlm_ctxt *dlm)
0366 {
0367     int ret;
0368 
0369     spin_lock(&dlm_domain_lock);
0370     ret = (dlm->dlm_state == DLM_CTXT_JOINED) ||
0371         (dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN);
0372     spin_unlock(&dlm_domain_lock);
0373 
0374     return ret;
0375 }
0376 
0377 static void dlm_destroy_dlm_worker(struct dlm_ctxt *dlm)
0378 {
0379     if (dlm->dlm_worker) {
0380         destroy_workqueue(dlm->dlm_worker);
0381         dlm->dlm_worker = NULL;
0382     }
0383 }
0384 
0385 static void dlm_complete_dlm_shutdown(struct dlm_ctxt *dlm)
0386 {
0387     dlm_unregister_domain_handlers(dlm);
0388     dlm_complete_thread(dlm);
0389     dlm_complete_recovery_thread(dlm);
0390     dlm_destroy_dlm_worker(dlm);
0391 
0392     /* We've left the domain. Now we can take ourselves out of the
0393      * list and allow the kref stuff to help us free the
0394      * memory. */
0395     spin_lock(&dlm_domain_lock);
0396     list_del_init(&dlm->list);
0397     spin_unlock(&dlm_domain_lock);
0398 
0399     /* Wake up anyone waiting for us to remove this domain */
0400     wake_up(&dlm_domain_events);
0401 }
0402 
0403 static int dlm_migrate_all_locks(struct dlm_ctxt *dlm)
0404 {
0405     int i, num, n, ret = 0;
0406     struct dlm_lock_resource *res;
0407     struct hlist_node *iter;
0408     struct hlist_head *bucket;
0409     int dropped;
0410 
0411     mlog(0, "Migrating locks from domain %s\n", dlm->name);
0412 
0413     num = 0;
0414     spin_lock(&dlm->spinlock);
0415     for (i = 0; i < DLM_HASH_BUCKETS; i++) {
0416 redo_bucket:
0417         n = 0;
0418         bucket = dlm_lockres_hash(dlm, i);
0419         iter = bucket->first;
0420         while (iter) {
0421             n++;
0422             res = hlist_entry(iter, struct dlm_lock_resource,
0423                       hash_node);
0424             dlm_lockres_get(res);
0425             /* migrate, if necessary.  this will drop the dlm
0426              * spinlock and retake it if it does migration. */
0427             dropped = dlm_empty_lockres(dlm, res);
0428 
0429             spin_lock(&res->spinlock);
0430             if (dropped)
0431                 __dlm_lockres_calc_usage(dlm, res);
0432             else
0433                 iter = res->hash_node.next;
0434             spin_unlock(&res->spinlock);
0435 
0436             dlm_lockres_put(res);
0437 
0438             if (dropped) {
0439                 cond_resched_lock(&dlm->spinlock);
0440                 goto redo_bucket;
0441             }
0442         }
0443         cond_resched_lock(&dlm->spinlock);
0444         num += n;
0445     }
0446 
0447     if (!num) {
0448         if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
0449             mlog(0, "%s: perhaps there are more lock resources "
0450                  "need to be migrated after dlm recovery\n", dlm->name);
0451             ret = -EAGAIN;
0452         } else {
0453             mlog(0, "%s: we won't do dlm recovery after migrating "
0454                  "all lock resources\n", dlm->name);
0455             dlm->migrate_done = 1;
0456         }
0457     }
0458 
0459     spin_unlock(&dlm->spinlock);
0460     wake_up(&dlm->dlm_thread_wq);
0461 
0462     /* let the dlm thread take care of purging, keep scanning until
0463      * nothing remains in the hash */
0464     if (num) {
0465         mlog(0, "%s: %d lock resources in hash last pass\n",
0466              dlm->name, num);
0467         ret = -EAGAIN;
0468     }
0469     mlog(0, "DONE Migrating locks from domain %s\n", dlm->name);
0470     return ret;
0471 }
0472 
0473 static int dlm_no_joining_node(struct dlm_ctxt *dlm)
0474 {
0475     int ret;
0476 
0477     spin_lock(&dlm->spinlock);
0478     ret = dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN;
0479     spin_unlock(&dlm->spinlock);
0480 
0481     return ret;
0482 }
0483 
0484 static int dlm_begin_exit_domain_handler(struct o2net_msg *msg, u32 len,
0485                      void *data, void **ret_data)
0486 {
0487     struct dlm_ctxt *dlm = data;
0488     unsigned int node;
0489     struct dlm_exit_domain *exit_msg = (struct dlm_exit_domain *) msg->buf;
0490 
0491     if (!dlm_grab(dlm))
0492         return 0;
0493 
0494     node = exit_msg->node_idx;
0495     mlog(0, "%s: Node %u sent a begin exit domain message\n", dlm->name, node);
0496 
0497     spin_lock(&dlm->spinlock);
0498     set_bit(node, dlm->exit_domain_map);
0499     spin_unlock(&dlm->spinlock);
0500 
0501     dlm_put(dlm);
0502 
0503     return 0;
0504 }
0505 
0506 static void dlm_mark_domain_leaving(struct dlm_ctxt *dlm)
0507 {
0508     /* Yikes, a double spinlock! I need domain_lock for the dlm
0509      * state and the dlm spinlock for join state... Sorry! */
0510 again:
0511     spin_lock(&dlm_domain_lock);
0512     spin_lock(&dlm->spinlock);
0513 
0514     if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
0515         mlog(0, "Node %d is joining, we wait on it.\n",
0516               dlm->joining_node);
0517         spin_unlock(&dlm->spinlock);
0518         spin_unlock(&dlm_domain_lock);
0519 
0520         wait_event(dlm->dlm_join_events, dlm_no_joining_node(dlm));
0521         goto again;
0522     }
0523 
0524     dlm->dlm_state = DLM_CTXT_LEAVING;
0525     spin_unlock(&dlm->spinlock);
0526     spin_unlock(&dlm_domain_lock);
0527 }
0528 
0529 static void __dlm_print_nodes(struct dlm_ctxt *dlm)
0530 {
0531     int node = -1, num = 0;
0532 
0533     assert_spin_locked(&dlm->spinlock);
0534 
0535     printk("( ");
0536     while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES,
0537                      node + 1)) < O2NM_MAX_NODES) {
0538         printk("%d ", node);
0539         ++num;
0540     }
0541     printk(") %u nodes\n", num);
0542 }
0543 
0544 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
0545                    void **ret_data)
0546 {
0547     struct dlm_ctxt *dlm = data;
0548     unsigned int node;
0549     struct dlm_exit_domain *exit_msg = (struct dlm_exit_domain *) msg->buf;
0550 
0551     mlog(0, "%p %u %p", msg, len, data);
0552 
0553     if (!dlm_grab(dlm))
0554         return 0;
0555 
0556     node = exit_msg->node_idx;
0557 
0558     spin_lock(&dlm->spinlock);
0559     clear_bit(node, dlm->domain_map);
0560     clear_bit(node, dlm->exit_domain_map);
0561     printk(KERN_NOTICE "o2dlm: Node %u leaves domain %s ", node, dlm->name);
0562     __dlm_print_nodes(dlm);
0563 
0564     /* notify anything attached to the heartbeat events */
0565     dlm_hb_event_notify_attached(dlm, node, 0);
0566 
0567     spin_unlock(&dlm->spinlock);
0568 
0569     dlm_put(dlm);
0570 
0571     return 0;
0572 }
0573 
0574 static int dlm_send_one_domain_exit(struct dlm_ctxt *dlm, u32 msg_type,
0575                     unsigned int node)
0576 {
0577     int status;
0578     struct dlm_exit_domain leave_msg;
0579 
0580     mlog(0, "%s: Sending domain exit message %u to node %u\n", dlm->name,
0581          msg_type, node);
0582 
0583     memset(&leave_msg, 0, sizeof(leave_msg));
0584     leave_msg.node_idx = dlm->node_num;
0585 
0586     status = o2net_send_message(msg_type, dlm->key, &leave_msg,
0587                     sizeof(leave_msg), node, NULL);
0588     if (status < 0)
0589         mlog(ML_ERROR, "Error %d sending domain exit message %u "
0590              "to node %u on domain %s\n", status, msg_type, node,
0591              dlm->name);
0592 
0593     return status;
0594 }
0595 
0596 static void dlm_begin_exit_domain(struct dlm_ctxt *dlm)
0597 {
0598     int node = -1;
0599 
0600     /* Support for begin exit domain was added in 1.2 */
0601     if (dlm->dlm_locking_proto.pv_major == 1 &&
0602         dlm->dlm_locking_proto.pv_minor < 2)
0603         return;
0604 
0605     /*
0606      * Unlike DLM_EXIT_DOMAIN_MSG, DLM_BEGIN_EXIT_DOMAIN_MSG is purely
0607      * informational. Meaning if a node does not receive the message,
0608      * so be it.
0609      */
0610     spin_lock(&dlm->spinlock);
0611     while (1) {
0612         node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES, node + 1);
0613         if (node >= O2NM_MAX_NODES)
0614             break;
0615         if (node == dlm->node_num)
0616             continue;
0617 
0618         spin_unlock(&dlm->spinlock);
0619         dlm_send_one_domain_exit(dlm, DLM_BEGIN_EXIT_DOMAIN_MSG, node);
0620         spin_lock(&dlm->spinlock);
0621     }
0622     spin_unlock(&dlm->spinlock);
0623 }
0624 
0625 static void dlm_leave_domain(struct dlm_ctxt *dlm)
0626 {
0627     int node, clear_node, status;
0628 
0629     /* At this point we've migrated away all our locks and won't
0630      * accept mastership of new ones. The dlm is responsible for
0631      * almost nothing now. We make sure not to confuse any joining
0632      * nodes and then commence shutdown procedure. */
0633 
0634     spin_lock(&dlm->spinlock);
0635     /* Clear ourselves from the domain map */
0636     clear_bit(dlm->node_num, dlm->domain_map);
0637     while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES,
0638                      0)) < O2NM_MAX_NODES) {
0639         /* Drop the dlm spinlock. This is safe wrt the domain_map.
0640          * -nodes cannot be added now as the
0641          *   query_join_handlers knows to respond with OK_NO_MAP
0642          * -we catch the right network errors if a node is
0643          *   removed from the map while we're sending him the
0644          *   exit message. */
0645         spin_unlock(&dlm->spinlock);
0646 
0647         clear_node = 1;
0648 
0649         status = dlm_send_one_domain_exit(dlm, DLM_EXIT_DOMAIN_MSG,
0650                           node);
0651         if (status < 0 &&
0652             status != -ENOPROTOOPT &&
0653             status != -ENOTCONN) {
0654             mlog(ML_NOTICE, "Error %d sending domain exit message "
0655                  "to node %d\n", status, node);
0656 
0657             /* Not sure what to do here but lets sleep for
0658              * a bit in case this was a transient
0659              * error... */
0660             msleep(DLM_DOMAIN_BACKOFF_MS);
0661             clear_node = 0;
0662         }
0663 
0664         spin_lock(&dlm->spinlock);
0665         /* If we're not clearing the node bit then we intend
0666          * to loop back around to try again. */
0667         if (clear_node)
0668             clear_bit(node, dlm->domain_map);
0669     }
0670     spin_unlock(&dlm->spinlock);
0671 }
0672 
0673 void dlm_unregister_domain(struct dlm_ctxt *dlm)
0674 {
0675     int leave = 0;
0676     struct dlm_lock_resource *res;
0677 
0678     spin_lock(&dlm_domain_lock);
0679     BUG_ON(dlm->dlm_state != DLM_CTXT_JOINED);
0680     BUG_ON(!dlm->num_joins);
0681 
0682     dlm->num_joins--;
0683     if (!dlm->num_joins) {
0684         /* We mark it "in shutdown" now so new register
0685          * requests wait until we've completely left the
0686          * domain. Don't use DLM_CTXT_LEAVING yet as we still
0687          * want new domain joins to communicate with us at
0688          * least until we've completed migration of our
0689          * resources. */
0690         dlm->dlm_state = DLM_CTXT_IN_SHUTDOWN;
0691         leave = 1;
0692     }
0693     spin_unlock(&dlm_domain_lock);
0694 
0695     if (leave) {
0696         mlog(0, "shutting down domain %s\n", dlm->name);
0697         dlm_begin_exit_domain(dlm);
0698 
0699         /* We changed dlm state, notify the thread */
0700         dlm_kick_thread(dlm, NULL);
0701 
0702         while (dlm_migrate_all_locks(dlm)) {
0703             /* Give dlm_thread time to purge the lockres' */
0704             msleep(500);
0705             mlog(0, "%s: more migration to do\n", dlm->name);
0706         }
0707 
0708         /* This list should be empty. If not, print remaining lockres */
0709         if (!list_empty(&dlm->tracking_list)) {
0710             mlog(ML_ERROR, "Following lockres' are still on the "
0711                  "tracking list:\n");
0712             list_for_each_entry(res, &dlm->tracking_list, tracking)
0713                 dlm_print_one_lock_resource(res);
0714         }
0715 
0716         dlm_mark_domain_leaving(dlm);
0717         dlm_leave_domain(dlm);
0718         printk(KERN_NOTICE "o2dlm: Leaving domain %s\n", dlm->name);
0719         dlm_force_free_mles(dlm);
0720         dlm_complete_dlm_shutdown(dlm);
0721     }
0722     dlm_put(dlm);
0723 }
0724 EXPORT_SYMBOL_GPL(dlm_unregister_domain);
0725 
0726 static int dlm_query_join_proto_check(char *proto_type, int node,
0727                       struct dlm_protocol_version *ours,
0728                       struct dlm_protocol_version *request)
0729 {
0730     int rc;
0731     struct dlm_protocol_version proto = *request;
0732 
0733     if (!dlm_protocol_compare(ours, &proto)) {
0734         mlog(0,
0735              "node %u wanted to join with %s locking protocol "
0736              "%u.%u, we respond with %u.%u\n",
0737              node, proto_type,
0738              request->pv_major,
0739              request->pv_minor,
0740              proto.pv_major, proto.pv_minor);
0741         request->pv_minor = proto.pv_minor;
0742         rc = 0;
0743     } else {
0744         mlog(ML_NOTICE,
0745              "Node %u wanted to join with %s locking "
0746              "protocol %u.%u, but we have %u.%u, disallowing\n",
0747              node, proto_type,
0748              request->pv_major,
0749              request->pv_minor,
0750              ours->pv_major,
0751              ours->pv_minor);
0752         rc = 1;
0753     }
0754 
0755     return rc;
0756 }
0757 
0758 /*
0759  * struct dlm_query_join_packet is made up of four one-byte fields.  They
0760  * are effectively in big-endian order already.  However, little-endian
0761  * machines swap them before putting the packet on the wire (because
0762  * query_join's response is a status, and that status is treated as a u32
0763  * on the wire).  Thus, a big-endian and little-endian machines will treat
0764  * this structure differently.
0765  *
0766  * The solution is to have little-endian machines swap the structure when
0767  * converting from the structure to the u32 representation.  This will
0768  * result in the structure having the correct format on the wire no matter
0769  * the host endian format.
0770  */
0771 static void dlm_query_join_packet_to_wire(struct dlm_query_join_packet *packet,
0772                       u32 *wire)
0773 {
0774     union dlm_query_join_response response;
0775 
0776     response.packet = *packet;
0777     *wire = be32_to_cpu(response.intval);
0778 }
0779 
0780 static void dlm_query_join_wire_to_packet(u32 wire,
0781                       struct dlm_query_join_packet *packet)
0782 {
0783     union dlm_query_join_response response;
0784 
0785     response.intval = cpu_to_be32(wire);
0786     *packet = response.packet;
0787 }
0788 
0789 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
0790                   void **ret_data)
0791 {
0792     struct dlm_query_join_request *query;
0793     struct dlm_query_join_packet packet = {
0794         .code = JOIN_DISALLOW,
0795     };
0796     struct dlm_ctxt *dlm = NULL;
0797     u32 response;
0798     u8 nodenum;
0799 
0800     query = (struct dlm_query_join_request *) msg->buf;
0801 
0802     mlog(0, "node %u wants to join domain %s\n", query->node_idx,
0803           query->domain);
0804 
0805     /*
0806      * If heartbeat doesn't consider the node live, tell it
0807      * to back off and try again.  This gives heartbeat a chance
0808      * to catch up.
0809      */
0810     if (!o2hb_check_node_heartbeating_no_sem(query->node_idx)) {
0811         mlog(0, "node %u is not in our live map yet\n",
0812              query->node_idx);
0813 
0814         packet.code = JOIN_DISALLOW;
0815         goto respond;
0816     }
0817 
0818     packet.code = JOIN_OK_NO_MAP;
0819 
0820     spin_lock(&dlm_domain_lock);
0821     dlm = __dlm_lookup_domain_full(query->domain, query->name_len);
0822     if (!dlm)
0823         goto unlock_respond;
0824 
0825     /*
0826      * There is a small window where the joining node may not see the
0827      * node(s) that just left but still part of the cluster. DISALLOW
0828      * join request if joining node has different node map.
0829      */
0830     nodenum=0;
0831     while (nodenum < O2NM_MAX_NODES) {
0832         if (test_bit(nodenum, dlm->domain_map)) {
0833             if (!byte_test_bit(nodenum, query->node_map)) {
0834                 mlog(0, "disallow join as node %u does not "
0835                      "have node %u in its nodemap\n",
0836                      query->node_idx, nodenum);
0837                 packet.code = JOIN_DISALLOW;
0838                 goto unlock_respond;
0839             }
0840         }
0841         nodenum++;
0842     }
0843 
0844     /* Once the dlm ctxt is marked as leaving then we don't want
0845      * to be put in someone's domain map.
0846      * Also, explicitly disallow joining at certain troublesome
0847      * times (ie. during recovery). */
0848     if (dlm->dlm_state != DLM_CTXT_LEAVING) {
0849         int bit = query->node_idx;
0850         spin_lock(&dlm->spinlock);
0851 
0852         if (dlm->dlm_state == DLM_CTXT_NEW &&
0853             dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN) {
0854             /*If this is a brand new context and we
0855              * haven't started our join process yet, then
0856              * the other node won the race. */
0857             packet.code = JOIN_OK_NO_MAP;
0858         } else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
0859             /* Disallow parallel joins. */
0860             packet.code = JOIN_DISALLOW;
0861         } else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
0862             mlog(0, "node %u trying to join, but recovery "
0863                  "is ongoing.\n", bit);
0864             packet.code = JOIN_DISALLOW;
0865         } else if (test_bit(bit, dlm->recovery_map)) {
0866             mlog(0, "node %u trying to join, but it "
0867                  "still needs recovery.\n", bit);
0868             packet.code = JOIN_DISALLOW;
0869         } else if (test_bit(bit, dlm->domain_map)) {
0870             mlog(0, "node %u trying to join, but it "
0871                  "is still in the domain! needs recovery?\n",
0872                  bit);
0873             packet.code = JOIN_DISALLOW;
0874         } else {
0875             /* Alright we're fully a part of this domain
0876              * so we keep some state as to who's joining
0877              * and indicate to him that needs to be fixed
0878              * up. */
0879 
0880             /* Make sure we speak compatible locking protocols.  */
0881             if (dlm_query_join_proto_check("DLM", bit,
0882                                &dlm->dlm_locking_proto,
0883                                &query->dlm_proto)) {
0884                 packet.code = JOIN_PROTOCOL_MISMATCH;
0885             } else if (dlm_query_join_proto_check("fs", bit,
0886                                   &dlm->fs_locking_proto,
0887                                   &query->fs_proto)) {
0888                 packet.code = JOIN_PROTOCOL_MISMATCH;
0889             } else {
0890                 packet.dlm_minor = query->dlm_proto.pv_minor;
0891                 packet.fs_minor = query->fs_proto.pv_minor;
0892                 packet.code = JOIN_OK;
0893                 __dlm_set_joining_node(dlm, query->node_idx);
0894             }
0895         }
0896 
0897         spin_unlock(&dlm->spinlock);
0898     }
0899 unlock_respond:
0900     spin_unlock(&dlm_domain_lock);
0901 
0902 respond:
0903     mlog(0, "We respond with %u\n", packet.code);
0904 
0905     dlm_query_join_packet_to_wire(&packet, &response);
0906     return response;
0907 }
0908 
0909 static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
0910                      void **ret_data)
0911 {
0912     struct dlm_assert_joined *assert;
0913     struct dlm_ctxt *dlm = NULL;
0914 
0915     assert = (struct dlm_assert_joined *) msg->buf;
0916 
0917     mlog(0, "node %u asserts join on domain %s\n", assert->node_idx,
0918           assert->domain);
0919 
0920     spin_lock(&dlm_domain_lock);
0921     dlm = __dlm_lookup_domain_full(assert->domain, assert->name_len);
0922     /* XXX should we consider no dlm ctxt an error? */
0923     if (dlm) {
0924         spin_lock(&dlm->spinlock);
0925 
0926         /* Alright, this node has officially joined our
0927          * domain. Set him in the map and clean up our
0928          * leftover join state. */
0929         BUG_ON(dlm->joining_node != assert->node_idx);
0930 
0931         if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
0932             mlog(0, "dlm recovery is ongoing, disallow join\n");
0933             spin_unlock(&dlm->spinlock);
0934             spin_unlock(&dlm_domain_lock);
0935             return -EAGAIN;
0936         }
0937 
0938         set_bit(assert->node_idx, dlm->domain_map);
0939         clear_bit(assert->node_idx, dlm->exit_domain_map);
0940         __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
0941 
0942         printk(KERN_NOTICE "o2dlm: Node %u joins domain %s ",
0943                assert->node_idx, dlm->name);
0944         __dlm_print_nodes(dlm);
0945 
0946         /* notify anything attached to the heartbeat events */
0947         dlm_hb_event_notify_attached(dlm, assert->node_idx, 1);
0948 
0949         spin_unlock(&dlm->spinlock);
0950     }
0951     spin_unlock(&dlm_domain_lock);
0952 
0953     return 0;
0954 }
0955 
0956 static int dlm_match_regions(struct dlm_ctxt *dlm,
0957                  struct dlm_query_region *qr,
0958                  char *local, int locallen)
0959 {
0960     char *remote = qr->qr_regions;
0961     char *l, *r;
0962     int localnr, i, j, foundit;
0963     int status = 0;
0964 
0965     if (!o2hb_global_heartbeat_active()) {
0966         if (qr->qr_numregions) {
0967             mlog(ML_ERROR, "Domain %s: Joining node %d has global "
0968                  "heartbeat enabled but local node %d does not\n",
0969                  qr->qr_domain, qr->qr_node, dlm->node_num);
0970             status = -EINVAL;
0971         }
0972         goto bail;
0973     }
0974 
0975     if (o2hb_global_heartbeat_active() && !qr->qr_numregions) {
0976         mlog(ML_ERROR, "Domain %s: Local node %d has global "
0977              "heartbeat enabled but joining node %d does not\n",
0978              qr->qr_domain, dlm->node_num, qr->qr_node);
0979         status = -EINVAL;
0980         goto bail;
0981     }
0982 
0983     r = remote;
0984     for (i = 0; i < qr->qr_numregions; ++i) {
0985         mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, r);
0986         r += O2HB_MAX_REGION_NAME_LEN;
0987     }
0988 
0989     localnr = min(O2NM_MAX_REGIONS, locallen/O2HB_MAX_REGION_NAME_LEN);
0990     localnr = o2hb_get_all_regions(local, (u8)localnr);
0991 
0992     /* compare local regions with remote */
0993     l = local;
0994     for (i = 0; i < localnr; ++i) {
0995         foundit = 0;
0996         r = remote;
0997         for (j = 0; j <= qr->qr_numregions; ++j) {
0998             if (!memcmp(l, r, O2HB_MAX_REGION_NAME_LEN)) {
0999                 foundit = 1;
1000                 break;
1001             }
1002             r += O2HB_MAX_REGION_NAME_LEN;
1003         }
1004         if (!foundit) {
1005             status = -EINVAL;
1006             mlog(ML_ERROR, "Domain %s: Region '%.*s' registered "
1007                  "in local node %d but not in joining node %d\n",
1008                  qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, l,
1009                  dlm->node_num, qr->qr_node);
1010             goto bail;
1011         }
1012         l += O2HB_MAX_REGION_NAME_LEN;
1013     }
1014 
1015     /* compare remote with local regions */
1016     r = remote;
1017     for (i = 0; i < qr->qr_numregions; ++i) {
1018         foundit = 0;
1019         l = local;
1020         for (j = 0; j < localnr; ++j) {
1021             if (!memcmp(r, l, O2HB_MAX_REGION_NAME_LEN)) {
1022                 foundit = 1;
1023                 break;
1024             }
1025             l += O2HB_MAX_REGION_NAME_LEN;
1026         }
1027         if (!foundit) {
1028             status = -EINVAL;
1029             mlog(ML_ERROR, "Domain %s: Region '%.*s' registered "
1030                  "in joining node %d but not in local node %d\n",
1031                  qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, r,
1032                  qr->qr_node, dlm->node_num);
1033             goto bail;
1034         }
1035         r += O2HB_MAX_REGION_NAME_LEN;
1036     }
1037 
1038 bail:
1039     return status;
1040 }
1041 
1042 static int dlm_send_regions(struct dlm_ctxt *dlm, unsigned long *node_map)
1043 {
1044     struct dlm_query_region *qr = NULL;
1045     int status, ret = 0, i;
1046     char *p;
1047 
1048     if (find_first_bit(node_map, O2NM_MAX_NODES) >= O2NM_MAX_NODES)
1049         goto bail;
1050 
1051     qr = kzalloc(sizeof(struct dlm_query_region), GFP_KERNEL);
1052     if (!qr) {
1053         ret = -ENOMEM;
1054         mlog_errno(ret);
1055         goto bail;
1056     }
1057 
1058     qr->qr_node = dlm->node_num;
1059     qr->qr_namelen = strlen(dlm->name);
1060     memcpy(qr->qr_domain, dlm->name, qr->qr_namelen);
1061     /* if local hb, the numregions will be zero */
1062     if (o2hb_global_heartbeat_active())
1063         qr->qr_numregions = o2hb_get_all_regions(qr->qr_regions,
1064                              O2NM_MAX_REGIONS);
1065 
1066     p = qr->qr_regions;
1067     for (i = 0; i < qr->qr_numregions; ++i, p += O2HB_MAX_REGION_NAME_LEN)
1068         mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, p);
1069 
1070     i = -1;
1071     while ((i = find_next_bit(node_map, O2NM_MAX_NODES,
1072                   i + 1)) < O2NM_MAX_NODES) {
1073         if (i == dlm->node_num)
1074             continue;
1075 
1076         mlog(0, "Sending regions to node %d\n", i);
1077 
1078         ret = o2net_send_message(DLM_QUERY_REGION, DLM_MOD_KEY, qr,
1079                      sizeof(struct dlm_query_region),
1080                      i, &status);
1081         if (ret >= 0)
1082             ret = status;
1083         if (ret) {
1084             mlog(ML_ERROR, "Region mismatch %d, node %d\n",
1085                  ret, i);
1086             break;
1087         }
1088     }
1089 
1090 bail:
1091     kfree(qr);
1092     return ret;
1093 }
1094 
1095 static int dlm_query_region_handler(struct o2net_msg *msg, u32 len,
1096                     void *data, void **ret_data)
1097 {
1098     struct dlm_query_region *qr;
1099     struct dlm_ctxt *dlm = NULL;
1100     char *local = NULL;
1101     int status = 0;
1102 
1103     qr = (struct dlm_query_region *) msg->buf;
1104 
1105     mlog(0, "Node %u queries hb regions on domain %s\n", qr->qr_node,
1106          qr->qr_domain);
1107 
1108     /* buffer used in dlm_mast_regions() */
1109     local = kmalloc(sizeof(qr->qr_regions), GFP_KERNEL);
1110     if (!local)
1111         return -ENOMEM;
1112 
1113     status = -EINVAL;
1114 
1115     spin_lock(&dlm_domain_lock);
1116     dlm = __dlm_lookup_domain_full(qr->qr_domain, qr->qr_namelen);
1117     if (!dlm) {
1118         mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
1119              "before join domain\n", qr->qr_node, qr->qr_domain);
1120         goto out_domain_lock;
1121     }
1122 
1123     spin_lock(&dlm->spinlock);
1124     if (dlm->joining_node != qr->qr_node) {
1125         mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
1126              "but joining node is %d\n", qr->qr_node, qr->qr_domain,
1127              dlm->joining_node);
1128         goto out_dlm_lock;
1129     }
1130 
1131     /* Support for global heartbeat was added in 1.1 */
1132     if (dlm->dlm_locking_proto.pv_major == 1 &&
1133         dlm->dlm_locking_proto.pv_minor == 0) {
1134         mlog(ML_ERROR, "Node %d queried hb regions on domain %s "
1135              "but active dlm protocol is %d.%d\n", qr->qr_node,
1136              qr->qr_domain, dlm->dlm_locking_proto.pv_major,
1137              dlm->dlm_locking_proto.pv_minor);
1138         goto out_dlm_lock;
1139     }
1140 
1141     status = dlm_match_regions(dlm, qr, local, sizeof(qr->qr_regions));
1142 
1143 out_dlm_lock:
1144     spin_unlock(&dlm->spinlock);
1145 
1146 out_domain_lock:
1147     spin_unlock(&dlm_domain_lock);
1148 
1149     kfree(local);
1150 
1151     return status;
1152 }
1153 
1154 static int dlm_match_nodes(struct dlm_ctxt *dlm, struct dlm_query_nodeinfo *qn)
1155 {
1156     struct o2nm_node *local;
1157     struct dlm_node_info *remote;
1158     int i, j;
1159     int status = 0;
1160 
1161     for (j = 0; j < qn->qn_numnodes; ++j)
1162         mlog(0, "Node %3d, %pI4:%u\n", qn->qn_nodes[j].ni_nodenum,
1163              &(qn->qn_nodes[j].ni_ipv4_address),
1164              ntohs(qn->qn_nodes[j].ni_ipv4_port));
1165 
1166     for (i = 0; i < O2NM_MAX_NODES && !status; ++i) {
1167         local = o2nm_get_node_by_num(i);
1168         remote = NULL;
1169         for (j = 0; j < qn->qn_numnodes; ++j) {
1170             if (qn->qn_nodes[j].ni_nodenum == i) {
1171                 remote = &(qn->qn_nodes[j]);
1172                 break;
1173             }
1174         }
1175 
1176         if (!local && !remote)
1177             continue;
1178 
1179         if ((local && !remote) || (!local && remote))
1180             status = -EINVAL;
1181 
1182         if (!status &&
1183             ((remote->ni_nodenum != local->nd_num) ||
1184              (remote->ni_ipv4_port != local->nd_ipv4_port) ||
1185              (remote->ni_ipv4_address != local->nd_ipv4_address)))
1186             status = -EINVAL;
1187 
1188         if (status) {
1189             if (remote && !local)
1190                 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) "
1191                      "registered in joining node %d but not in "
1192                      "local node %d\n", qn->qn_domain,
1193                      remote->ni_nodenum,
1194                      &(remote->ni_ipv4_address),
1195                      ntohs(remote->ni_ipv4_port),
1196                      qn->qn_nodenum, dlm->node_num);
1197             if (local && !remote)
1198                 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) "
1199                      "registered in local node %d but not in "
1200                      "joining node %d\n", qn->qn_domain,
1201                      local->nd_num, &(local->nd_ipv4_address),
1202                      ntohs(local->nd_ipv4_port),
1203                      dlm->node_num, qn->qn_nodenum);
1204             BUG_ON((!local && !remote));
1205         }
1206 
1207         if (local)
1208             o2nm_node_put(local);
1209     }
1210 
1211     return status;
1212 }
1213 
1214 static int dlm_send_nodeinfo(struct dlm_ctxt *dlm, unsigned long *node_map)
1215 {
1216     struct dlm_query_nodeinfo *qn = NULL;
1217     struct o2nm_node *node;
1218     int ret = 0, status, count, i;
1219 
1220     if (find_first_bit(node_map, O2NM_MAX_NODES) >= O2NM_MAX_NODES)
1221         goto bail;
1222 
1223     qn = kzalloc(sizeof(struct dlm_query_nodeinfo), GFP_KERNEL);
1224     if (!qn) {
1225         ret = -ENOMEM;
1226         mlog_errno(ret);
1227         goto bail;
1228     }
1229 
1230     for (i = 0, count = 0; i < O2NM_MAX_NODES; ++i) {
1231         node = o2nm_get_node_by_num(i);
1232         if (!node)
1233             continue;
1234         qn->qn_nodes[count].ni_nodenum = node->nd_num;
1235         qn->qn_nodes[count].ni_ipv4_port = node->nd_ipv4_port;
1236         qn->qn_nodes[count].ni_ipv4_address = node->nd_ipv4_address;
1237         mlog(0, "Node %3d, %pI4:%u\n", node->nd_num,
1238              &(node->nd_ipv4_address), ntohs(node->nd_ipv4_port));
1239         ++count;
1240         o2nm_node_put(node);
1241     }
1242 
1243     qn->qn_nodenum = dlm->node_num;
1244     qn->qn_numnodes = count;
1245     qn->qn_namelen = strlen(dlm->name);
1246     memcpy(qn->qn_domain, dlm->name, qn->qn_namelen);
1247 
1248     i = -1;
1249     while ((i = find_next_bit(node_map, O2NM_MAX_NODES,
1250                   i + 1)) < O2NM_MAX_NODES) {
1251         if (i == dlm->node_num)
1252             continue;
1253 
1254         mlog(0, "Sending nodeinfo to node %d\n", i);
1255 
1256         ret = o2net_send_message(DLM_QUERY_NODEINFO, DLM_MOD_KEY,
1257                      qn, sizeof(struct dlm_query_nodeinfo),
1258                      i, &status);
1259         if (ret >= 0)
1260             ret = status;
1261         if (ret) {
1262             mlog(ML_ERROR, "node mismatch %d, node %d\n", ret, i);
1263             break;
1264         }
1265     }
1266 
1267 bail:
1268     kfree(qn);
1269     return ret;
1270 }
1271 
1272 static int dlm_query_nodeinfo_handler(struct o2net_msg *msg, u32 len,
1273                       void *data, void **ret_data)
1274 {
1275     struct dlm_query_nodeinfo *qn;
1276     struct dlm_ctxt *dlm = NULL;
1277     int locked = 0, status = -EINVAL;
1278 
1279     qn = (struct dlm_query_nodeinfo *) msg->buf;
1280 
1281     mlog(0, "Node %u queries nodes on domain %s\n", qn->qn_nodenum,
1282          qn->qn_domain);
1283 
1284     spin_lock(&dlm_domain_lock);
1285     dlm = __dlm_lookup_domain_full(qn->qn_domain, qn->qn_namelen);
1286     if (!dlm) {
1287         mlog(ML_ERROR, "Node %d queried nodes on domain %s before "
1288              "join domain\n", qn->qn_nodenum, qn->qn_domain);
1289         goto bail;
1290     }
1291 
1292     spin_lock(&dlm->spinlock);
1293     locked = 1;
1294     if (dlm->joining_node != qn->qn_nodenum) {
1295         mlog(ML_ERROR, "Node %d queried nodes on domain %s but "
1296              "joining node is %d\n", qn->qn_nodenum, qn->qn_domain,
1297              dlm->joining_node);
1298         goto bail;
1299     }
1300 
1301     /* Support for node query was added in 1.1 */
1302     if (dlm->dlm_locking_proto.pv_major == 1 &&
1303         dlm->dlm_locking_proto.pv_minor == 0) {
1304         mlog(ML_ERROR, "Node %d queried nodes on domain %s "
1305              "but active dlm protocol is %d.%d\n", qn->qn_nodenum,
1306              qn->qn_domain, dlm->dlm_locking_proto.pv_major,
1307              dlm->dlm_locking_proto.pv_minor);
1308         goto bail;
1309     }
1310 
1311     status = dlm_match_nodes(dlm, qn);
1312 
1313 bail:
1314     if (locked)
1315         spin_unlock(&dlm->spinlock);
1316     spin_unlock(&dlm_domain_lock);
1317 
1318     return status;
1319 }
1320 
1321 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
1322                    void **ret_data)
1323 {
1324     struct dlm_cancel_join *cancel;
1325     struct dlm_ctxt *dlm = NULL;
1326 
1327     cancel = (struct dlm_cancel_join *) msg->buf;
1328 
1329     mlog(0, "node %u cancels join on domain %s\n", cancel->node_idx,
1330           cancel->domain);
1331 
1332     spin_lock(&dlm_domain_lock);
1333     dlm = __dlm_lookup_domain_full(cancel->domain, cancel->name_len);
1334 
1335     if (dlm) {
1336         spin_lock(&dlm->spinlock);
1337 
1338         /* Yikes, this guy wants to cancel his join. No
1339          * problem, we simply cleanup our join state. */
1340         BUG_ON(dlm->joining_node != cancel->node_idx);
1341         __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
1342 
1343         spin_unlock(&dlm->spinlock);
1344     }
1345     spin_unlock(&dlm_domain_lock);
1346 
1347     return 0;
1348 }
1349 
1350 static int dlm_send_one_join_cancel(struct dlm_ctxt *dlm,
1351                     unsigned int node)
1352 {
1353     int status;
1354     struct dlm_cancel_join cancel_msg;
1355 
1356     memset(&cancel_msg, 0, sizeof(cancel_msg));
1357     cancel_msg.node_idx = dlm->node_num;
1358     cancel_msg.name_len = strlen(dlm->name);
1359     memcpy(cancel_msg.domain, dlm->name, cancel_msg.name_len);
1360 
1361     status = o2net_send_message(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
1362                     &cancel_msg, sizeof(cancel_msg), node,
1363                     NULL);
1364     if (status < 0) {
1365         mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
1366              "node %u\n", status, DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
1367              node);
1368         goto bail;
1369     }
1370 
1371 bail:
1372     return status;
1373 }
1374 
1375 /* map_size should be in bytes. */
1376 static int dlm_send_join_cancels(struct dlm_ctxt *dlm,
1377                  unsigned long *node_map,
1378                  unsigned int map_size)
1379 {
1380     int status, tmpstat;
1381     int node;
1382 
1383     if (map_size != (BITS_TO_LONGS(O2NM_MAX_NODES) *
1384              sizeof(unsigned long))) {
1385         mlog(ML_ERROR,
1386              "map_size %u != BITS_TO_LONGS(O2NM_MAX_NODES) %u\n",
1387              map_size, (unsigned)BITS_TO_LONGS(O2NM_MAX_NODES));
1388         return -EINVAL;
1389     }
1390 
1391     status = 0;
1392     node = -1;
1393     while ((node = find_next_bit(node_map, O2NM_MAX_NODES,
1394                      node + 1)) < O2NM_MAX_NODES) {
1395         if (node == dlm->node_num)
1396             continue;
1397 
1398         tmpstat = dlm_send_one_join_cancel(dlm, node);
1399         if (tmpstat) {
1400             mlog(ML_ERROR, "Error return %d cancelling join on "
1401                  "node %d\n", tmpstat, node);
1402             if (!status)
1403                 status = tmpstat;
1404         }
1405     }
1406 
1407     if (status)
1408         mlog_errno(status);
1409     return status;
1410 }
1411 
1412 static int dlm_request_join(struct dlm_ctxt *dlm,
1413                 int node,
1414                 enum dlm_query_join_response_code *response)
1415 {
1416     int status;
1417     struct dlm_query_join_request join_msg;
1418     struct dlm_query_join_packet packet;
1419     u32 join_resp;
1420 
1421     mlog(0, "querying node %d\n", node);
1422 
1423     memset(&join_msg, 0, sizeof(join_msg));
1424     join_msg.node_idx = dlm->node_num;
1425     join_msg.name_len = strlen(dlm->name);
1426     memcpy(join_msg.domain, dlm->name, join_msg.name_len);
1427     join_msg.dlm_proto = dlm->dlm_locking_proto;
1428     join_msg.fs_proto = dlm->fs_locking_proto;
1429 
1430     /* copy live node map to join message */
1431     byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES);
1432 
1433     status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg,
1434                     sizeof(join_msg), node, &join_resp);
1435     if (status < 0 && status != -ENOPROTOOPT) {
1436         mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
1437              "node %u\n", status, DLM_QUERY_JOIN_MSG, DLM_MOD_KEY,
1438              node);
1439         goto bail;
1440     }
1441     dlm_query_join_wire_to_packet(join_resp, &packet);
1442 
1443     /* -ENOPROTOOPT from the net code means the other side isn't
1444         listening for our message type -- that's fine, it means
1445         his dlm isn't up, so we can consider him a 'yes' but not
1446         joined into the domain.  */
1447     if (status == -ENOPROTOOPT) {
1448         status = 0;
1449         *response = JOIN_OK_NO_MAP;
1450     } else {
1451         *response = packet.code;
1452         switch (packet.code) {
1453         case JOIN_DISALLOW:
1454         case JOIN_OK_NO_MAP:
1455             break;
1456         case JOIN_PROTOCOL_MISMATCH:
1457             mlog(ML_NOTICE,
1458                  "This node requested DLM locking protocol %u.%u and "
1459                  "filesystem locking protocol %u.%u.  At least one of "
1460                  "the protocol versions on node %d is not compatible, "
1461                  "disconnecting\n",
1462                  dlm->dlm_locking_proto.pv_major,
1463                  dlm->dlm_locking_proto.pv_minor,
1464                  dlm->fs_locking_proto.pv_major,
1465                  dlm->fs_locking_proto.pv_minor,
1466                  node);
1467             status = -EPROTO;
1468             break;
1469         case JOIN_OK:
1470             /* Use the same locking protocol as the remote node */
1471             dlm->dlm_locking_proto.pv_minor = packet.dlm_minor;
1472             dlm->fs_locking_proto.pv_minor = packet.fs_minor;
1473             mlog(0,
1474                  "Node %d responds JOIN_OK with DLM locking protocol "
1475                  "%u.%u and fs locking protocol %u.%u\n",
1476                  node,
1477                  dlm->dlm_locking_proto.pv_major,
1478                  dlm->dlm_locking_proto.pv_minor,
1479                  dlm->fs_locking_proto.pv_major,
1480                  dlm->fs_locking_proto.pv_minor);
1481             break;
1482         default:
1483             status = -EINVAL;
1484             mlog(ML_ERROR, "invalid response %d from node %u\n",
1485                  packet.code, node);
1486             /* Reset response to JOIN_DISALLOW */
1487             *response = JOIN_DISALLOW;
1488             break;
1489         }
1490     }
1491 
1492     mlog(0, "status %d, node %d response is %d\n", status, node,
1493          *response);
1494 
1495 bail:
1496     return status;
1497 }
1498 
1499 static int dlm_send_one_join_assert(struct dlm_ctxt *dlm,
1500                     unsigned int node)
1501 {
1502     int status;
1503     int ret;
1504     struct dlm_assert_joined assert_msg;
1505 
1506     mlog(0, "Sending join assert to node %u\n", node);
1507 
1508     memset(&assert_msg, 0, sizeof(assert_msg));
1509     assert_msg.node_idx = dlm->node_num;
1510     assert_msg.name_len = strlen(dlm->name);
1511     memcpy(assert_msg.domain, dlm->name, assert_msg.name_len);
1512 
1513     status = o2net_send_message(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
1514                     &assert_msg, sizeof(assert_msg), node,
1515                     &ret);
1516     if (status < 0)
1517         mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
1518              "node %u\n", status, DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
1519              node);
1520     else
1521         status = ret;
1522 
1523     return status;
1524 }
1525 
1526 static void dlm_send_join_asserts(struct dlm_ctxt *dlm,
1527                   unsigned long *node_map)
1528 {
1529     int status, node, live;
1530 
1531     status = 0;
1532     node = -1;
1533     while ((node = find_next_bit(node_map, O2NM_MAX_NODES,
1534                      node + 1)) < O2NM_MAX_NODES) {
1535         if (node == dlm->node_num)
1536             continue;
1537 
1538         do {
1539             /* It is very important that this message be
1540              * received so we spin until either the node
1541              * has died or it gets the message. */
1542             status = dlm_send_one_join_assert(dlm, node);
1543 
1544             spin_lock(&dlm->spinlock);
1545             live = test_bit(node, dlm->live_nodes_map);
1546             spin_unlock(&dlm->spinlock);
1547 
1548             if (status) {
1549                 mlog(ML_ERROR, "Error return %d asserting "
1550                      "join on node %d\n", status, node);
1551 
1552                 /* give us some time between errors... */
1553                 if (live)
1554                     msleep(DLM_DOMAIN_BACKOFF_MS);
1555             }
1556         } while (status && live);
1557     }
1558 }
1559 
1560 struct domain_join_ctxt {
1561     unsigned long live_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
1562     unsigned long yes_resp_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
1563 };
1564 
1565 static int dlm_should_restart_join(struct dlm_ctxt *dlm,
1566                    struct domain_join_ctxt *ctxt,
1567                    enum dlm_query_join_response_code response)
1568 {
1569     int ret;
1570 
1571     if (response == JOIN_DISALLOW) {
1572         mlog(0, "Latest response of disallow -- should restart\n");
1573         return 1;
1574     }
1575 
1576     spin_lock(&dlm->spinlock);
1577     /* For now, we restart the process if the node maps have
1578      * changed at all */
1579     ret = memcmp(ctxt->live_map, dlm->live_nodes_map,
1580              sizeof(dlm->live_nodes_map));
1581     spin_unlock(&dlm->spinlock);
1582 
1583     if (ret)
1584         mlog(0, "Node maps changed -- should restart\n");
1585 
1586     return ret;
1587 }
1588 
1589 static int dlm_try_to_join_domain(struct dlm_ctxt *dlm)
1590 {
1591     int status = 0, tmpstat, node;
1592     struct domain_join_ctxt *ctxt;
1593     enum dlm_query_join_response_code response = JOIN_DISALLOW;
1594 
1595     mlog(0, "%p", dlm);
1596 
1597     ctxt = kzalloc(sizeof(*ctxt), GFP_KERNEL);
1598     if (!ctxt) {
1599         status = -ENOMEM;
1600         mlog_errno(status);
1601         goto bail;
1602     }
1603 
1604     /* group sem locking should work for us here -- we're already
1605      * registered for heartbeat events so filling this should be
1606      * atomic wrt getting those handlers called. */
1607     o2hb_fill_node_map(dlm->live_nodes_map, sizeof(dlm->live_nodes_map));
1608 
1609     spin_lock(&dlm->spinlock);
1610     memcpy(ctxt->live_map, dlm->live_nodes_map, sizeof(ctxt->live_map));
1611 
1612     __dlm_set_joining_node(dlm, dlm->node_num);
1613 
1614     spin_unlock(&dlm->spinlock);
1615 
1616     node = -1;
1617     while ((node = find_next_bit(ctxt->live_map, O2NM_MAX_NODES,
1618                      node + 1)) < O2NM_MAX_NODES) {
1619         if (node == dlm->node_num)
1620             continue;
1621 
1622         status = dlm_request_join(dlm, node, &response);
1623         if (status < 0) {
1624             mlog_errno(status);
1625             goto bail;
1626         }
1627 
1628         /* Ok, either we got a response or the node doesn't have a
1629          * dlm up. */
1630         if (response == JOIN_OK)
1631             set_bit(node, ctxt->yes_resp_map);
1632 
1633         if (dlm_should_restart_join(dlm, ctxt, response)) {
1634             status = -EAGAIN;
1635             goto bail;
1636         }
1637     }
1638 
1639     mlog(0, "Yay, done querying nodes!\n");
1640 
1641     /* Yay, everyone agree's we can join the domain. My domain is
1642      * comprised of all nodes who were put in the
1643      * yes_resp_map. Copy that into our domain map and send a join
1644      * assert message to clean up everyone elses state. */
1645     spin_lock(&dlm->spinlock);
1646     memcpy(dlm->domain_map, ctxt->yes_resp_map,
1647            sizeof(ctxt->yes_resp_map));
1648     set_bit(dlm->node_num, dlm->domain_map);
1649     spin_unlock(&dlm->spinlock);
1650 
1651     /* Support for global heartbeat and node info was added in 1.1 */
1652     if (dlm->dlm_locking_proto.pv_major > 1 ||
1653         dlm->dlm_locking_proto.pv_minor > 0) {
1654         status = dlm_send_nodeinfo(dlm, ctxt->yes_resp_map);
1655         if (status) {
1656             mlog_errno(status);
1657             goto bail;
1658         }
1659         status = dlm_send_regions(dlm, ctxt->yes_resp_map);
1660         if (status) {
1661             mlog_errno(status);
1662             goto bail;
1663         }
1664     }
1665 
1666     dlm_send_join_asserts(dlm, ctxt->yes_resp_map);
1667 
1668     /* Joined state *must* be set before the joining node
1669      * information, otherwise the query_join handler may read no
1670      * current joiner but a state of NEW and tell joining nodes
1671      * we're not in the domain. */
1672     spin_lock(&dlm_domain_lock);
1673     dlm->dlm_state = DLM_CTXT_JOINED;
1674     dlm->num_joins++;
1675     spin_unlock(&dlm_domain_lock);
1676 
1677 bail:
1678     spin_lock(&dlm->spinlock);
1679     __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
1680     if (!status) {
1681         printk(KERN_NOTICE "o2dlm: Joining domain %s ", dlm->name);
1682         __dlm_print_nodes(dlm);
1683     }
1684     spin_unlock(&dlm->spinlock);
1685 
1686     if (ctxt) {
1687         /* Do we need to send a cancel message to any nodes? */
1688         if (status < 0) {
1689             tmpstat = dlm_send_join_cancels(dlm,
1690                             ctxt->yes_resp_map,
1691                             sizeof(ctxt->yes_resp_map));
1692             if (tmpstat < 0)
1693                 mlog_errno(tmpstat);
1694         }
1695         kfree(ctxt);
1696     }
1697 
1698     mlog(0, "returning %d\n", status);
1699     return status;
1700 }
1701 
1702 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm)
1703 {
1704     o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_up);
1705     o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_down);
1706     o2net_unregister_handler_list(&dlm->dlm_domain_handlers);
1707 }
1708 
1709 static int dlm_register_domain_handlers(struct dlm_ctxt *dlm)
1710 {
1711     int status;
1712 
1713     mlog(0, "registering handlers.\n");
1714 
1715     o2hb_setup_callback(&dlm->dlm_hb_down, O2HB_NODE_DOWN_CB,
1716                 dlm_hb_node_down_cb, dlm, DLM_HB_NODE_DOWN_PRI);
1717     o2hb_setup_callback(&dlm->dlm_hb_up, O2HB_NODE_UP_CB,
1718                 dlm_hb_node_up_cb, dlm, DLM_HB_NODE_UP_PRI);
1719 
1720     status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_down);
1721     if (status)
1722         goto bail;
1723 
1724     status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_up);
1725     if (status)
1726         goto bail;
1727 
1728     status = o2net_register_handler(DLM_MASTER_REQUEST_MSG, dlm->key,
1729                     sizeof(struct dlm_master_request),
1730                     dlm_master_request_handler,
1731                     dlm, NULL, &dlm->dlm_domain_handlers);
1732     if (status)
1733         goto bail;
1734 
1735     status = o2net_register_handler(DLM_ASSERT_MASTER_MSG, dlm->key,
1736                     sizeof(struct dlm_assert_master),
1737                     dlm_assert_master_handler,
1738                     dlm, dlm_assert_master_post_handler,
1739                     &dlm->dlm_domain_handlers);
1740     if (status)
1741         goto bail;
1742 
1743     status = o2net_register_handler(DLM_CREATE_LOCK_MSG, dlm->key,
1744                     sizeof(struct dlm_create_lock),
1745                     dlm_create_lock_handler,
1746                     dlm, NULL, &dlm->dlm_domain_handlers);
1747     if (status)
1748         goto bail;
1749 
1750     status = o2net_register_handler(DLM_CONVERT_LOCK_MSG, dlm->key,
1751                     DLM_CONVERT_LOCK_MAX_LEN,
1752                     dlm_convert_lock_handler,
1753                     dlm, NULL, &dlm->dlm_domain_handlers);
1754     if (status)
1755         goto bail;
1756 
1757     status = o2net_register_handler(DLM_UNLOCK_LOCK_MSG, dlm->key,
1758                     DLM_UNLOCK_LOCK_MAX_LEN,
1759                     dlm_unlock_lock_handler,
1760                     dlm, NULL, &dlm->dlm_domain_handlers);
1761     if (status)
1762         goto bail;
1763 
1764     status = o2net_register_handler(DLM_PROXY_AST_MSG, dlm->key,
1765                     DLM_PROXY_AST_MAX_LEN,
1766                     dlm_proxy_ast_handler,
1767                     dlm, NULL, &dlm->dlm_domain_handlers);
1768     if (status)
1769         goto bail;
1770 
1771     status = o2net_register_handler(DLM_EXIT_DOMAIN_MSG, dlm->key,
1772                     sizeof(struct dlm_exit_domain),
1773                     dlm_exit_domain_handler,
1774                     dlm, NULL, &dlm->dlm_domain_handlers);
1775     if (status)
1776         goto bail;
1777 
1778     status = o2net_register_handler(DLM_DEREF_LOCKRES_MSG, dlm->key,
1779                     sizeof(struct dlm_deref_lockres),
1780                     dlm_deref_lockres_handler,
1781                     dlm, NULL, &dlm->dlm_domain_handlers);
1782     if (status)
1783         goto bail;
1784 
1785     status = o2net_register_handler(DLM_MIGRATE_REQUEST_MSG, dlm->key,
1786                     sizeof(struct dlm_migrate_request),
1787                     dlm_migrate_request_handler,
1788                     dlm, NULL, &dlm->dlm_domain_handlers);
1789     if (status)
1790         goto bail;
1791 
1792     status = o2net_register_handler(DLM_MIG_LOCKRES_MSG, dlm->key,
1793                     DLM_MIG_LOCKRES_MAX_LEN,
1794                     dlm_mig_lockres_handler,
1795                     dlm, NULL, &dlm->dlm_domain_handlers);
1796     if (status)
1797         goto bail;
1798 
1799     status = o2net_register_handler(DLM_MASTER_REQUERY_MSG, dlm->key,
1800                     sizeof(struct dlm_master_requery),
1801                     dlm_master_requery_handler,
1802                     dlm, NULL, &dlm->dlm_domain_handlers);
1803     if (status)
1804         goto bail;
1805 
1806     status = o2net_register_handler(DLM_LOCK_REQUEST_MSG, dlm->key,
1807                     sizeof(struct dlm_lock_request),
1808                     dlm_request_all_locks_handler,
1809                     dlm, NULL, &dlm->dlm_domain_handlers);
1810     if (status)
1811         goto bail;
1812 
1813     status = o2net_register_handler(DLM_RECO_DATA_DONE_MSG, dlm->key,
1814                     sizeof(struct dlm_reco_data_done),
1815                     dlm_reco_data_done_handler,
1816                     dlm, NULL, &dlm->dlm_domain_handlers);
1817     if (status)
1818         goto bail;
1819 
1820     status = o2net_register_handler(DLM_BEGIN_RECO_MSG, dlm->key,
1821                     sizeof(struct dlm_begin_reco),
1822                     dlm_begin_reco_handler,
1823                     dlm, NULL, &dlm->dlm_domain_handlers);
1824     if (status)
1825         goto bail;
1826 
1827     status = o2net_register_handler(DLM_FINALIZE_RECO_MSG, dlm->key,
1828                     sizeof(struct dlm_finalize_reco),
1829                     dlm_finalize_reco_handler,
1830                     dlm, NULL, &dlm->dlm_domain_handlers);
1831     if (status)
1832         goto bail;
1833 
1834     status = o2net_register_handler(DLM_BEGIN_EXIT_DOMAIN_MSG, dlm->key,
1835                     sizeof(struct dlm_exit_domain),
1836                     dlm_begin_exit_domain_handler,
1837                     dlm, NULL, &dlm->dlm_domain_handlers);
1838     if (status)
1839         goto bail;
1840 
1841     status = o2net_register_handler(DLM_DEREF_LOCKRES_DONE, dlm->key,
1842                     sizeof(struct dlm_deref_lockres_done),
1843                     dlm_deref_lockres_done_handler,
1844                     dlm, NULL, &dlm->dlm_domain_handlers);
1845 bail:
1846     if (status)
1847         dlm_unregister_domain_handlers(dlm);
1848 
1849     return status;
1850 }
1851 
1852 static int dlm_join_domain(struct dlm_ctxt *dlm)
1853 {
1854     int status;
1855     unsigned int backoff;
1856     unsigned int total_backoff = 0;
1857     char wq_name[O2NM_MAX_NAME_LEN];
1858 
1859     BUG_ON(!dlm);
1860 
1861     mlog(0, "Join domain %s\n", dlm->name);
1862 
1863     status = dlm_register_domain_handlers(dlm);
1864     if (status) {
1865         mlog_errno(status);
1866         goto bail;
1867     }
1868 
1869     status = dlm_launch_thread(dlm);
1870     if (status < 0) {
1871         mlog_errno(status);
1872         goto bail;
1873     }
1874 
1875     status = dlm_launch_recovery_thread(dlm);
1876     if (status < 0) {
1877         mlog_errno(status);
1878         goto bail;
1879     }
1880 
1881     dlm_debug_init(dlm);
1882 
1883     snprintf(wq_name, O2NM_MAX_NAME_LEN, "dlm_wq-%s", dlm->name);
1884     dlm->dlm_worker = alloc_workqueue(wq_name, WQ_MEM_RECLAIM, 0);
1885     if (!dlm->dlm_worker) {
1886         status = -ENOMEM;
1887         mlog_errno(status);
1888         goto bail;
1889     }
1890 
1891     do {
1892         status = dlm_try_to_join_domain(dlm);
1893 
1894         /* If we're racing another node to the join, then we
1895          * need to back off temporarily and let them
1896          * complete. */
1897 #define DLM_JOIN_TIMEOUT_MSECS  90000
1898         if (status == -EAGAIN) {
1899             if (signal_pending(current)) {
1900                 status = -ERESTARTSYS;
1901                 goto bail;
1902             }
1903 
1904             if (total_backoff > DLM_JOIN_TIMEOUT_MSECS) {
1905                 status = -ERESTARTSYS;
1906                 mlog(ML_NOTICE, "Timed out joining dlm domain "
1907                      "%s after %u msecs\n", dlm->name,
1908                      total_backoff);
1909                 goto bail;
1910             }
1911 
1912             /*
1913              * <chip> After you!
1914              * <dale> No, after you!
1915              * <chip> I insist!
1916              * <dale> But you first!
1917              * ...
1918              */
1919             backoff = (unsigned int)(jiffies & 0x3);
1920             backoff *= DLM_DOMAIN_BACKOFF_MS;
1921             total_backoff += backoff;
1922             mlog(0, "backoff %d\n", backoff);
1923             msleep(backoff);
1924         }
1925     } while (status == -EAGAIN);
1926 
1927     if (status < 0) {
1928         mlog_errno(status);
1929         goto bail;
1930     }
1931 
1932     status = 0;
1933 bail:
1934     wake_up(&dlm_domain_events);
1935 
1936     if (status) {
1937         dlm_unregister_domain_handlers(dlm);
1938         dlm_complete_thread(dlm);
1939         dlm_complete_recovery_thread(dlm);
1940         dlm_destroy_dlm_worker(dlm);
1941     }
1942 
1943     return status;
1944 }
1945 
1946 static struct dlm_ctxt *dlm_alloc_ctxt(const char *domain,
1947                 u32 key)
1948 {
1949     int i;
1950     int ret;
1951     struct dlm_ctxt *dlm = NULL;
1952 
1953     dlm = kzalloc(sizeof(*dlm), GFP_KERNEL);
1954     if (!dlm) {
1955         ret = -ENOMEM;
1956         mlog_errno(ret);
1957         goto leave;
1958     }
1959 
1960     dlm->name = kstrdup(domain, GFP_KERNEL);
1961     if (dlm->name == NULL) {
1962         ret = -ENOMEM;
1963         mlog_errno(ret);
1964         goto leave;
1965     }
1966 
1967     dlm->lockres_hash = (struct hlist_head **)dlm_alloc_pagevec(DLM_HASH_PAGES);
1968     if (!dlm->lockres_hash) {
1969         ret = -ENOMEM;
1970         mlog_errno(ret);
1971         goto leave;
1972     }
1973 
1974     for (i = 0; i < DLM_HASH_BUCKETS; i++)
1975         INIT_HLIST_HEAD(dlm_lockres_hash(dlm, i));
1976 
1977     dlm->master_hash = (struct hlist_head **)
1978                 dlm_alloc_pagevec(DLM_HASH_PAGES);
1979     if (!dlm->master_hash) {
1980         ret = -ENOMEM;
1981         mlog_errno(ret);
1982         goto leave;
1983     }
1984 
1985     for (i = 0; i < DLM_HASH_BUCKETS; i++)
1986         INIT_HLIST_HEAD(dlm_master_hash(dlm, i));
1987 
1988     dlm->key = key;
1989     dlm->node_num = o2nm_this_node();
1990 
1991     dlm_create_debugfs_subroot(dlm);
1992 
1993     spin_lock_init(&dlm->spinlock);
1994     spin_lock_init(&dlm->master_lock);
1995     spin_lock_init(&dlm->ast_lock);
1996     spin_lock_init(&dlm->track_lock);
1997     INIT_LIST_HEAD(&dlm->list);
1998     INIT_LIST_HEAD(&dlm->dirty_list);
1999     INIT_LIST_HEAD(&dlm->reco.resources);
2000     INIT_LIST_HEAD(&dlm->reco.node_data);
2001     INIT_LIST_HEAD(&dlm->purge_list);
2002     INIT_LIST_HEAD(&dlm->dlm_domain_handlers);
2003     INIT_LIST_HEAD(&dlm->tracking_list);
2004     dlm->reco.state = 0;
2005 
2006     INIT_LIST_HEAD(&dlm->pending_asts);
2007     INIT_LIST_HEAD(&dlm->pending_basts);
2008 
2009     mlog(0, "dlm->recovery_map=%p, &(dlm->recovery_map[0])=%p\n",
2010           dlm->recovery_map, &(dlm->recovery_map[0]));
2011 
2012     memset(dlm->recovery_map, 0, sizeof(dlm->recovery_map));
2013     memset(dlm->live_nodes_map, 0, sizeof(dlm->live_nodes_map));
2014     memset(dlm->domain_map, 0, sizeof(dlm->domain_map));
2015 
2016     dlm->dlm_thread_task = NULL;
2017     dlm->dlm_reco_thread_task = NULL;
2018     dlm->dlm_worker = NULL;
2019     init_waitqueue_head(&dlm->dlm_thread_wq);
2020     init_waitqueue_head(&dlm->dlm_reco_thread_wq);
2021     init_waitqueue_head(&dlm->reco.event);
2022     init_waitqueue_head(&dlm->ast_wq);
2023     init_waitqueue_head(&dlm->migration_wq);
2024     INIT_LIST_HEAD(&dlm->mle_hb_events);
2025 
2026     dlm->joining_node = DLM_LOCK_RES_OWNER_UNKNOWN;
2027     init_waitqueue_head(&dlm->dlm_join_events);
2028 
2029     dlm->migrate_done = 0;
2030 
2031     dlm->reco.new_master = O2NM_INVALID_NODE_NUM;
2032     dlm->reco.dead_node = O2NM_INVALID_NODE_NUM;
2033 
2034     atomic_set(&dlm->res_tot_count, 0);
2035     atomic_set(&dlm->res_cur_count, 0);
2036     for (i = 0; i < DLM_MLE_NUM_TYPES; ++i) {
2037         atomic_set(&dlm->mle_tot_count[i], 0);
2038         atomic_set(&dlm->mle_cur_count[i], 0);
2039     }
2040 
2041     spin_lock_init(&dlm->work_lock);
2042     INIT_LIST_HEAD(&dlm->work_list);
2043     INIT_WORK(&dlm->dispatched_work, dlm_dispatch_work);
2044 
2045     kref_init(&dlm->dlm_refs);
2046     dlm->dlm_state = DLM_CTXT_NEW;
2047 
2048     INIT_LIST_HEAD(&dlm->dlm_eviction_callbacks);
2049 
2050     mlog(0, "context init: refcount %u\n",
2051           kref_read(&dlm->dlm_refs));
2052 
2053     ret = 0;
2054 leave:
2055     if (ret < 0 && dlm) {
2056         if (dlm->master_hash)
2057             dlm_free_pagevec((void **)dlm->master_hash,
2058                     DLM_HASH_PAGES);
2059 
2060         if (dlm->lockres_hash)
2061             dlm_free_pagevec((void **)dlm->lockres_hash,
2062                     DLM_HASH_PAGES);
2063 
2064         kfree(dlm->name);
2065         kfree(dlm);
2066         dlm = NULL;
2067     }
2068     return dlm;
2069 }
2070 
2071 /*
2072  * Compare a requested locking protocol version against the current one.
2073  *
2074  * If the major numbers are different, they are incompatible.
2075  * If the current minor is greater than the request, they are incompatible.
2076  * If the current minor is less than or equal to the request, they are
2077  * compatible, and the requester should run at the current minor version.
2078  */
2079 static int dlm_protocol_compare(struct dlm_protocol_version *existing,
2080                 struct dlm_protocol_version *request)
2081 {
2082     if (existing->pv_major != request->pv_major)
2083         return 1;
2084 
2085     if (existing->pv_minor > request->pv_minor)
2086         return 1;
2087 
2088     if (existing->pv_minor < request->pv_minor)
2089         request->pv_minor = existing->pv_minor;
2090 
2091     return 0;
2092 }
2093 
2094 /*
2095  * dlm_register_domain: one-time setup per "domain".
2096  *
2097  * The filesystem passes in the requested locking version via proto.
2098  * If registration was successful, proto will contain the negotiated
2099  * locking protocol.
2100  */
2101 struct dlm_ctxt * dlm_register_domain(const char *domain,
2102                    u32 key,
2103                    struct dlm_protocol_version *fs_proto)
2104 {
2105     int ret;
2106     struct dlm_ctxt *dlm = NULL;
2107     struct dlm_ctxt *new_ctxt = NULL;
2108 
2109     if (strlen(domain) >= O2NM_MAX_NAME_LEN) {
2110         ret = -ENAMETOOLONG;
2111         mlog(ML_ERROR, "domain name length too long\n");
2112         goto leave;
2113     }
2114 
2115     mlog(0, "register called for domain \"%s\"\n", domain);
2116 
2117 retry:
2118     dlm = NULL;
2119     if (signal_pending(current)) {
2120         ret = -ERESTARTSYS;
2121         mlog_errno(ret);
2122         goto leave;
2123     }
2124 
2125     spin_lock(&dlm_domain_lock);
2126 
2127     dlm = __dlm_lookup_domain(domain);
2128     if (dlm) {
2129         if (dlm->dlm_state != DLM_CTXT_JOINED) {
2130             spin_unlock(&dlm_domain_lock);
2131 
2132             mlog(0, "This ctxt is not joined yet!\n");
2133             wait_event_interruptible(dlm_domain_events,
2134                          dlm_wait_on_domain_helper(
2135                              domain));
2136             goto retry;
2137         }
2138 
2139         if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) {
2140             spin_unlock(&dlm_domain_lock);
2141             mlog(ML_ERROR,
2142                  "Requested locking protocol version is not "
2143                  "compatible with already registered domain "
2144                  "\"%s\"\n", domain);
2145             ret = -EPROTO;
2146             goto leave;
2147         }
2148 
2149         __dlm_get(dlm);
2150         dlm->num_joins++;
2151 
2152         spin_unlock(&dlm_domain_lock);
2153 
2154         ret = 0;
2155         goto leave;
2156     }
2157 
2158     /* doesn't exist */
2159     if (!new_ctxt) {
2160         spin_unlock(&dlm_domain_lock);
2161 
2162         new_ctxt = dlm_alloc_ctxt(domain, key);
2163         if (new_ctxt)
2164             goto retry;
2165 
2166         ret = -ENOMEM;
2167         mlog_errno(ret);
2168         goto leave;
2169     }
2170 
2171     /* a little variable switch-a-roo here... */
2172     dlm = new_ctxt;
2173     new_ctxt = NULL;
2174 
2175     /* add the new domain */
2176     list_add_tail(&dlm->list, &dlm_domains);
2177     spin_unlock(&dlm_domain_lock);
2178 
2179     /*
2180      * Pass the locking protocol version into the join.  If the join
2181      * succeeds, it will have the negotiated protocol set.
2182      */
2183     dlm->dlm_locking_proto = dlm_protocol;
2184     dlm->fs_locking_proto = *fs_proto;
2185 
2186     ret = dlm_join_domain(dlm);
2187     if (ret) {
2188         mlog_errno(ret);
2189         dlm_put(dlm);
2190         goto leave;
2191     }
2192 
2193     /* Tell the caller what locking protocol we negotiated */
2194     *fs_proto = dlm->fs_locking_proto;
2195 
2196     ret = 0;
2197 leave:
2198     if (new_ctxt)
2199         dlm_free_ctxt_mem(new_ctxt);
2200 
2201     if (ret < 0)
2202         dlm = ERR_PTR(ret);
2203 
2204     return dlm;
2205 }
2206 EXPORT_SYMBOL_GPL(dlm_register_domain);
2207 
2208 static LIST_HEAD(dlm_join_handlers);
2209 
2210 static void dlm_unregister_net_handlers(void)
2211 {
2212     o2net_unregister_handler_list(&dlm_join_handlers);
2213 }
2214 
2215 static int dlm_register_net_handlers(void)
2216 {
2217     int status = 0;
2218 
2219     status = o2net_register_handler(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY,
2220                     sizeof(struct dlm_query_join_request),
2221                     dlm_query_join_handler,
2222                     NULL, NULL, &dlm_join_handlers);
2223     if (status)
2224         goto bail;
2225 
2226     status = o2net_register_handler(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
2227                     sizeof(struct dlm_assert_joined),
2228                     dlm_assert_joined_handler,
2229                     NULL, NULL, &dlm_join_handlers);
2230     if (status)
2231         goto bail;
2232 
2233     status = o2net_register_handler(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
2234                     sizeof(struct dlm_cancel_join),
2235                     dlm_cancel_join_handler,
2236                     NULL, NULL, &dlm_join_handlers);
2237     if (status)
2238         goto bail;
2239 
2240     status = o2net_register_handler(DLM_QUERY_REGION, DLM_MOD_KEY,
2241                     sizeof(struct dlm_query_region),
2242                     dlm_query_region_handler,
2243                     NULL, NULL, &dlm_join_handlers);
2244 
2245     if (status)
2246         goto bail;
2247 
2248     status = o2net_register_handler(DLM_QUERY_NODEINFO, DLM_MOD_KEY,
2249                     sizeof(struct dlm_query_nodeinfo),
2250                     dlm_query_nodeinfo_handler,
2251                     NULL, NULL, &dlm_join_handlers);
2252 bail:
2253     if (status < 0)
2254         dlm_unregister_net_handlers();
2255 
2256     return status;
2257 }
2258 
2259 /* Domain eviction callback handling.
2260  *
2261  * The file system requires notification of node death *before* the
2262  * dlm completes it's recovery work, otherwise it may be able to
2263  * acquire locks on resources requiring recovery. Since the dlm can
2264  * evict a node from it's domain *before* heartbeat fires, a similar
2265  * mechanism is required. */
2266 
2267 /* Eviction is not expected to happen often, so a per-domain lock is
2268  * not necessary. Eviction callbacks are allowed to sleep for short
2269  * periods of time. */
2270 static DECLARE_RWSEM(dlm_callback_sem);
2271 
2272 void dlm_fire_domain_eviction_callbacks(struct dlm_ctxt *dlm,
2273                     int node_num)
2274 {
2275     struct dlm_eviction_cb *cb;
2276 
2277     down_read(&dlm_callback_sem);
2278     list_for_each_entry(cb, &dlm->dlm_eviction_callbacks, ec_item) {
2279         cb->ec_func(node_num, cb->ec_data);
2280     }
2281     up_read(&dlm_callback_sem);
2282 }
2283 
2284 void dlm_setup_eviction_cb(struct dlm_eviction_cb *cb,
2285                dlm_eviction_func *f,
2286                void *data)
2287 {
2288     INIT_LIST_HEAD(&cb->ec_item);
2289     cb->ec_func = f;
2290     cb->ec_data = data;
2291 }
2292 EXPORT_SYMBOL_GPL(dlm_setup_eviction_cb);
2293 
2294 void dlm_register_eviction_cb(struct dlm_ctxt *dlm,
2295                   struct dlm_eviction_cb *cb)
2296 {
2297     down_write(&dlm_callback_sem);
2298     list_add_tail(&cb->ec_item, &dlm->dlm_eviction_callbacks);
2299     up_write(&dlm_callback_sem);
2300 }
2301 EXPORT_SYMBOL_GPL(dlm_register_eviction_cb);
2302 
2303 void dlm_unregister_eviction_cb(struct dlm_eviction_cb *cb)
2304 {
2305     down_write(&dlm_callback_sem);
2306     list_del_init(&cb->ec_item);
2307     up_write(&dlm_callback_sem);
2308 }
2309 EXPORT_SYMBOL_GPL(dlm_unregister_eviction_cb);
2310 
2311 static int __init dlm_init(void)
2312 {
2313     int status;
2314 
2315     status = dlm_init_mle_cache();
2316     if (status) {
2317         mlog(ML_ERROR, "Could not create o2dlm_mle slabcache\n");
2318         goto error;
2319     }
2320 
2321     status = dlm_init_master_caches();
2322     if (status) {
2323         mlog(ML_ERROR, "Could not create o2dlm_lockres and "
2324              "o2dlm_lockname slabcaches\n");
2325         goto error;
2326     }
2327 
2328     status = dlm_init_lock_cache();
2329     if (status) {
2330         mlog(ML_ERROR, "Count not create o2dlm_lock slabcache\n");
2331         goto error;
2332     }
2333 
2334     status = dlm_register_net_handlers();
2335     if (status) {
2336         mlog(ML_ERROR, "Unable to register network handlers\n");
2337         goto error;
2338     }
2339 
2340     dlm_create_debugfs_root();
2341 
2342     return 0;
2343 error:
2344     dlm_unregister_net_handlers();
2345     dlm_destroy_lock_cache();
2346     dlm_destroy_master_caches();
2347     dlm_destroy_mle_cache();
2348     return -1;
2349 }
2350 
2351 static void __exit dlm_exit (void)
2352 {
2353     dlm_destroy_debugfs_root();
2354     dlm_unregister_net_handlers();
2355     dlm_destroy_lock_cache();
2356     dlm_destroy_master_caches();
2357     dlm_destroy_mle_cache();
2358 }
2359 
2360 MODULE_AUTHOR("Oracle");
2361 MODULE_LICENSE("GPL");
2362 MODULE_DESCRIPTION("OCFS2 Distributed Lock Management");
2363 
2364 module_init(dlm_init);
2365 module_exit(dlm_exit);
[ Source navigation ] [ Diff markup ] [ Identifier search ] [ general search ]

This page was automatically generated by the 2.1.0 LXR engine.

The LXR team
Valid CSS 2.1! Valid HTML 4.01!