fs/ocfs2/stack_o2cb.c

0001 // SPDX-License-Identifier: GPL-2.0-only
0002 /*
0003  * stack_o2cb.c
0004  *
0005  * Code which interfaces ocfs2 with the o2cb stack.
0006  *
0007  * Copyright (C) 2007 Oracle.  All rights reserved.
0008  */
0009
0010 #include <linux/kernel.h>
0011 #include <linux/crc32.h>
0012 #include <linux/slab.h>
0013 #include <linux/module.h>
0014
0015 /* Needed for AOP_TRUNCATED_PAGE in mlog_errno() */
0016 #include <linux/fs.h>
0017
0018 #include "cluster/masklog.h"
0019 #include "cluster/nodemanager.h"
0020 #include "cluster/heartbeat.h"
0021 #include "cluster/tcp.h"
0022
0023 #include "stackglue.h"
0024
0025 struct o2dlm_private {
0026     struct dlm_eviction_cb op_eviction_cb;
0027 };
0028
0029 static struct ocfs2_stack_plugin o2cb_stack;
0030
0031 /* These should be identical */
0032 #if (DLM_LOCK_IV != LKM_IVMODE)
0033 # error Lock modes do not match
0034 #endif
0035 #if (DLM_LOCK_NL != LKM_NLMODE)
0036 # error Lock modes do not match
0037 #endif
0038 #if (DLM_LOCK_CR != LKM_CRMODE)
0039 # error Lock modes do not match
0040 #endif
0041 #if (DLM_LOCK_CW != LKM_CWMODE)
0042 # error Lock modes do not match
0043 #endif
0044 #if (DLM_LOCK_PR != LKM_PRMODE)
0045 # error Lock modes do not match
0046 #endif
0047 #if (DLM_LOCK_PW != LKM_PWMODE)
0048 # error Lock modes do not match
0049 #endif
0050 #if (DLM_LOCK_EX != LKM_EXMODE)
0051 # error Lock modes do not match
0052 #endif
0053 static inline int mode_to_o2dlm(int mode)
0054 {
0055     BUG_ON(mode > LKM_MAXMODE);
0056
0057     return mode;
0058 }
0059
0060 static int flags_to_o2dlm(u32 flags)
0061 {
0062     int o2dlm_flags = 0;
0063
0064     if (flags & DLM_LKF_NOQUEUE)
0065         o2dlm_flags |= LKM_NOQUEUE;
0066     if (flags & DLM_LKF_CANCEL)
0067         o2dlm_flags |= LKM_CANCEL;
0068     if (flags & DLM_LKF_CONVERT)
0069         o2dlm_flags |= LKM_CONVERT;
0070     if (flags & DLM_LKF_VALBLK)
0071         o2dlm_flags |= LKM_VALBLK;
0072     if (flags & DLM_LKF_IVVALBLK)
0073         o2dlm_flags |= LKM_INVVALBLK;
0074     if (flags & DLM_LKF_ORPHAN)
0075         o2dlm_flags |= LKM_ORPHAN;
0076     if (flags & DLM_LKF_FORCEUNLOCK)
0077         o2dlm_flags |= LKM_FORCE;
0078     if (flags & DLM_LKF_TIMEOUT)
0079         o2dlm_flags |= LKM_TIMEOUT;
0080     if (flags & DLM_LKF_LOCAL)
0081         o2dlm_flags |= LKM_LOCAL;
0082
0083     return o2dlm_flags;
0084 }
0085
0086 /*
0087  * Map an o2dlm status to standard errno values.
0088  *
0089  * o2dlm only uses a handful of these, and returns even fewer to the
0090  * caller. Still, we try to assign sane values to each error.
0091  *
0092  * The following value pairs have special meanings to dlmglue, thus
0093  * the right hand side needs to stay unique - never duplicate the
0094  * mapping elsewhere in the table!
0095  *
0096  * DLM_NORMAL:      0
0097  * DLM_NOTQUEUED:   -EAGAIN
0098  * DLM_CANCELGRANT: -EBUSY
0099  * DLM_CANCEL:      -DLM_ECANCEL
0100  */
0101 /* Keep in sync with dlmapi.h */
0102 static int status_map[] = {
0103     [DLM_NORMAL]            = 0,        /* Success */
0104     [DLM_GRANTED]           = -EINVAL,
0105     [DLM_DENIED]            = -EACCES,
0106     [DLM_DENIED_NOLOCKS]        = -EACCES,
0107     [DLM_WORKING]           = -EACCES,
0108     [DLM_BLOCKED]           = -EINVAL,
0109     [DLM_BLOCKED_ORPHAN]        = -EINVAL,
0110     [DLM_DENIED_GRACE_PERIOD]   = -EACCES,
0111     [DLM_SYSERR]            = -ENOMEM,  /* It is what it is */
0112     [DLM_NOSUPPORT]         = -EPROTO,
0113     [DLM_CANCELGRANT]       = -EBUSY,   /* Cancel after grant */
0114     [DLM_IVLOCKID]          = -EINVAL,
0115     [DLM_SYNC]          = -EINVAL,
0116     [DLM_BADTYPE]           = -EINVAL,
0117     [DLM_BADRESOURCE]       = -EINVAL,
0118     [DLM_MAXHANDLES]        = -ENOMEM,
0119     [DLM_NOCLINFO]          = -EINVAL,
0120     [DLM_NOLOCKMGR]         = -EINVAL,
0121     [DLM_NOPURGED]          = -EINVAL,
0122     [DLM_BADARGS]           = -EINVAL,
0123     [DLM_VOID]          = -EINVAL,
0124     [DLM_NOTQUEUED]         = -EAGAIN,  /* Trylock failed */
0125     [DLM_IVBUFLEN]          = -EINVAL,
0126     [DLM_CVTUNGRANT]        = -EPERM,
0127     [DLM_BADPARAM]          = -EINVAL,
0128     [DLM_VALNOTVALID]       = -EINVAL,
0129     [DLM_REJECTED]          = -EPERM,
0130     [DLM_ABORT]         = -EINVAL,
0131     [DLM_CANCEL]            = -DLM_ECANCEL, /* Successful cancel */
0132     [DLM_IVRESHANDLE]       = -EINVAL,
0133     [DLM_DEADLOCK]          = -EDEADLK,
0134     [DLM_DENIED_NOASTS]     = -EINVAL,
0135     [DLM_FORWARD]           = -EINVAL,
0136     [DLM_TIMEOUT]           = -ETIMEDOUT,
0137     [DLM_IVGROUPID]         = -EINVAL,
0138     [DLM_VERS_CONFLICT]     = -EOPNOTSUPP,
0139     [DLM_BAD_DEVICE_PATH]       = -ENOENT,
0140     [DLM_NO_DEVICE_PERMISSION]  = -EPERM,
0141     [DLM_NO_CONTROL_DEVICE]     = -ENOENT,
0142     [DLM_RECOVERING]        = -ENOTCONN,
0143     [DLM_MIGRATING]         = -ERESTART,
0144     [DLM_MAXSTATS]          = -EINVAL,
0145 };
0146
0147 static int dlm_status_to_errno(enum dlm_status status)
0148 {
0149     BUG_ON(status < 0 || status >= ARRAY_SIZE(status_map));
0150
0151     return status_map[status];
0152 }
0153
0154 static void o2dlm_lock_ast_wrapper(void *astarg)
0155 {
0156     struct ocfs2_dlm_lksb *lksb = astarg;
0157
0158     lksb->lksb_conn->cc_proto->lp_lock_ast(lksb);
0159 }
0160
0161 static void o2dlm_blocking_ast_wrapper(void *astarg, int level)
0162 {
0163     struct ocfs2_dlm_lksb *lksb = astarg;
0164
0165     lksb->lksb_conn->cc_proto->lp_blocking_ast(lksb, level);
0166 }
0167
0168 static void o2dlm_unlock_ast_wrapper(void *astarg, enum dlm_status status)
0169 {
0170     struct ocfs2_dlm_lksb *lksb = astarg;
0171     int error = dlm_status_to_errno(status);
0172
0173     /*
0174      * In o2dlm, you can get both the lock_ast() for the lock being
0175      * granted and the unlock_ast() for the CANCEL failing.  A
0176      * successful cancel sends DLM_NORMAL here.  If the
0177      * lock grant happened before the cancel arrived, you get
0178      * DLM_CANCELGRANT.
0179      *
0180      * There's no need for the double-ast.  If we see DLM_CANCELGRANT,
0181      * we just ignore it.  We expect the lock_ast() to handle the
0182      * granted lock.
0183      */
0184     if (status == DLM_CANCELGRANT)
0185         return;
0186
0187     lksb->lksb_conn->cc_proto->lp_unlock_ast(lksb, error);
0188 }
0189
0190 static int o2cb_dlm_lock(struct ocfs2_cluster_connection *conn,
0191              int mode,
0192              struct ocfs2_dlm_lksb *lksb,
0193              u32 flags,
0194              void *name,
0195              unsigned int namelen)
0196 {
0197     enum dlm_status status;
0198     int o2dlm_mode = mode_to_o2dlm(mode);
0199     int o2dlm_flags = flags_to_o2dlm(flags);
0200     int ret;
0201
0202     status = dlmlock(conn->cc_lockspace, o2dlm_mode, &lksb->lksb_o2dlm,
0203              o2dlm_flags, name, namelen,
0204              o2dlm_lock_ast_wrapper, lksb,
0205              o2dlm_blocking_ast_wrapper);
0206     ret = dlm_status_to_errno(status);
0207     return ret;
0208 }
0209
0210 static int o2cb_dlm_unlock(struct ocfs2_cluster_connection *conn,
0211                struct ocfs2_dlm_lksb *lksb,
0212                u32 flags)
0213 {
0214     enum dlm_status status;
0215     int o2dlm_flags = flags_to_o2dlm(flags);
0216     int ret;
0217
0218     status = dlmunlock(conn->cc_lockspace, &lksb->lksb_o2dlm,
0219                o2dlm_flags, o2dlm_unlock_ast_wrapper, lksb);
0220     ret = dlm_status_to_errno(status);
0221     return ret;
0222 }
0223
0224 static int o2cb_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
0225 {
0226     return dlm_status_to_errno(lksb->lksb_o2dlm.status);
0227 }
0228
0229 /*
0230  * o2dlm aways has a "valid" LVB. If the dlm loses track of the LVB
0231  * contents, it will zero out the LVB.  Thus the caller can always trust
0232  * the contents.
0233  */
0234 static int o2cb_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
0235 {
0236     return 1;
0237 }
0238
0239 static void *o2cb_dlm_lvb(struct ocfs2_dlm_lksb *lksb)
0240 {
0241     return (void *)(lksb->lksb_o2dlm.lvb);
0242 }
0243
0244 static void o2cb_dump_lksb(struct ocfs2_dlm_lksb *lksb)
0245 {
0246     dlm_print_one_lock(lksb->lksb_o2dlm.lockid);
0247 }
0248
0249 /*
0250  * Check if this node is heartbeating and is connected to all other
0251  * heartbeating nodes.
0252  */
0253 static int o2cb_cluster_check(void)
0254 {
0255     u8 node_num;
0256     int i;
0257     unsigned long hbmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
0258     unsigned long netmap[BITS_TO_LONGS(O2NM_MAX_NODES)];
0259
0260     node_num = o2nm_this_node();
0261     if (node_num == O2NM_MAX_NODES) {
0262         printk(KERN_ERR "o2cb: This node has not been configured.\n");
0263         return -EINVAL;
0264     }
0265
0266     /*
0267      * o2dlm expects o2net sockets to be created. If not, then
0268      * dlm_join_domain() fails with a stack of errors which are both cryptic
0269      * and incomplete. The idea here is to detect upfront whether we have
0270      * managed to connect to all nodes or not. If not, then list the nodes
0271      * to allow the user to check the configuration (incorrect IP, firewall,
0272      * etc.) Yes, this is racy. But its not the end of the world.
0273      */
0274 #define O2CB_MAP_STABILIZE_COUNT    60
0275     for (i = 0; i < O2CB_MAP_STABILIZE_COUNT; ++i) {
0276         o2hb_fill_node_map(hbmap, sizeof(hbmap));
0277         if (!test_bit(node_num, hbmap)) {
0278             printk(KERN_ERR "o2cb: %s heartbeat has not been "
0279                    "started.\n", (o2hb_global_heartbeat_active() ?
0280                           "Global" : "Local"));
0281             return -EINVAL;
0282         }
0283         o2net_fill_node_map(netmap, sizeof(netmap));
0284         /* Force set the current node to allow easy compare */
0285         set_bit(node_num, netmap);
0286         if (!memcmp(hbmap, netmap, sizeof(hbmap)))
0287             return 0;
0288         if (i < O2CB_MAP_STABILIZE_COUNT - 1)
0289             msleep(1000);
0290     }
0291
0292     printk(KERN_ERR "o2cb: This node could not connect to nodes:");
0293     i = -1;
0294     while ((i = find_next_bit(hbmap, O2NM_MAX_NODES,
0295                   i + 1)) < O2NM_MAX_NODES) {
0296         if (!test_bit(i, netmap))
0297             printk(" %u", i);
0298     }
0299     printk(".\n");
0300
0301     return -ENOTCONN;
0302 }
0303
0304 /*
0305  * Called from the dlm when it's about to evict a node. This is how the
0306  * classic stack signals node death.
0307  */
0308 static void o2dlm_eviction_cb(int node_num, void *data)
0309 {
0310     struct ocfs2_cluster_connection *conn = data;
0311
0312     printk(KERN_NOTICE "o2cb: o2dlm has evicted node %d from domain %.*s\n",
0313            node_num, conn->cc_namelen, conn->cc_name);
0314
0315     conn->cc_recovery_handler(node_num, conn->cc_recovery_data);
0316 }
0317
0318 static int o2cb_cluster_connect(struct ocfs2_cluster_connection *conn)
0319 {
0320     int rc = 0;
0321     u32 dlm_key;
0322     struct dlm_ctxt *dlm;
0323     struct o2dlm_private *priv;
0324     struct dlm_protocol_version fs_version;
0325
0326     BUG_ON(conn == NULL);
0327     BUG_ON(conn->cc_proto == NULL);
0328
0329     /* Ensure cluster stack is up and all nodes are connected */
0330     rc = o2cb_cluster_check();
0331     if (rc) {
0332         printk(KERN_ERR "o2cb: Cluster check failed. Fix errors "
0333                "before retrying.\n");
0334         goto out;
0335     }
0336
0337     priv = kzalloc(sizeof(struct o2dlm_private), GFP_KERNEL);
0338     if (!priv) {
0339         rc = -ENOMEM;
0340         goto out_free;
0341     }
0342
0343     /* This just fills the structure in.  It is safe to pass conn. */
0344     dlm_setup_eviction_cb(&priv->op_eviction_cb, o2dlm_eviction_cb,
0345                   conn);
0346
0347     conn->cc_private = priv;
0348
0349     /* used by the dlm code to make message headers unique, each
0350      * node in this domain must agree on this. */
0351     dlm_key = crc32_le(0, conn->cc_name, conn->cc_namelen);
0352     fs_version.pv_major = conn->cc_version.pv_major;
0353     fs_version.pv_minor = conn->cc_version.pv_minor;
0354
0355     dlm = dlm_register_domain(conn->cc_name, dlm_key, &fs_version);
0356     if (IS_ERR(dlm)) {
0357         rc = PTR_ERR(dlm);
0358         mlog_errno(rc);
0359         goto out_free;
0360     }
0361
0362     conn->cc_version.pv_major = fs_version.pv_major;
0363     conn->cc_version.pv_minor = fs_version.pv_minor;
0364     conn->cc_lockspace = dlm;
0365
0366     dlm_register_eviction_cb(dlm, &priv->op_eviction_cb);
0367
0368 out_free:
0369     if (rc)
0370         kfree(conn->cc_private);
0371
0372 out:
0373     return rc;
0374 }
0375
0376 static int o2cb_cluster_disconnect(struct ocfs2_cluster_connection *conn)
0377 {
0378     struct dlm_ctxt *dlm = conn->cc_lockspace;
0379     struct o2dlm_private *priv = conn->cc_private;
0380
0381     dlm_unregister_eviction_cb(&priv->op_eviction_cb);
0382     conn->cc_private = NULL;
0383     kfree(priv);
0384
0385     dlm_unregister_domain(dlm);
0386     conn->cc_lockspace = NULL;
0387
0388     return 0;
0389 }
0390
0391 static int o2cb_cluster_this_node(struct ocfs2_cluster_connection *conn,
0392                   unsigned int *node)
0393 {
0394     int node_num;
0395
0396     node_num = o2nm_this_node();
0397     if (node_num == O2NM_INVALID_NODE_NUM)
0398         return -ENOENT;
0399
0400     if (node_num >= O2NM_MAX_NODES)
0401         return -EOVERFLOW;
0402
0403     *node = node_num;
0404     return 0;
0405 }
0406
0407 static struct ocfs2_stack_operations o2cb_stack_ops = {
0408     .connect    = o2cb_cluster_connect,
0409     .disconnect = o2cb_cluster_disconnect,
0410     .this_node  = o2cb_cluster_this_node,
0411     .dlm_lock   = o2cb_dlm_lock,
0412     .dlm_unlock = o2cb_dlm_unlock,
0413     .lock_status    = o2cb_dlm_lock_status,
0414     .lvb_valid  = o2cb_dlm_lvb_valid,
0415     .lock_lvb   = o2cb_dlm_lvb,
0416     .dump_lksb  = o2cb_dump_lksb,
0417 };
0418
0419 static struct ocfs2_stack_plugin o2cb_stack = {
0420     .sp_name    = "o2cb",
0421     .sp_ops     = &o2cb_stack_ops,
0422     .sp_owner   = THIS_MODULE,
0423 };
0424
0425 static int __init o2cb_stack_init(void)
0426 {
0427     return ocfs2_stack_glue_register(&o2cb_stack);
0428 }
0429
0430 static void __exit o2cb_stack_exit(void)
0431 {
0432     ocfs2_stack_glue_unregister(&o2cb_stack);
0433 }
0434
0435 MODULE_AUTHOR("Oracle");
0436 MODULE_DESCRIPTION("ocfs2 driver for the classic o2cb stack");
0437 MODULE_LICENSE("GPL");
0438 module_init(o2cb_stack_init);
0439 module_exit(o2cb_stack_exit);