Something went wrong. Try again.
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103// SPDX-License-Identifier: GPL-2.0-only/* * stack_user.c * * Code which interfaces ocfs2 with fs/dlm and a userspace stack. * * Copyright (C) 2007 Oracle. All rights reserved. */
#include <linux/module.h>#include <linux/fs.h>#include <linux/filelock.h>#include <linux/miscdevice.h>#include <linux/mutex.h>#include <linux/slab.h>#include <linux/reboot.h>#include <linux/sched.h>#include <linux/uaccess.h>
#include "stackglue.h"
#include <linux/dlm_plock.h>
/* * The control protocol starts with a handshake. Until the handshake * is complete, the control device will fail all write(2)s. * * The handshake is simple. First, the client reads until EOF. Each line * of output is a supported protocol tag. All protocol tags are a single * character followed by a two hex digit version number. Currently the * only things supported is T01, for "Text-base version 0x01". Next, the * client writes the version they would like to use, including the newline. * Thus, the protocol tag is 'T01\n'. If the version tag written is * unknown, -EINVAL is returned. Once the negotiation is complete, the * client can start sending messages. * * The T01 protocol has three messages. First is the "SETN" message. * It has the following syntax: * * SETN<space><8-char-hex-nodenum><newline> * * This is 14 characters. * * The "SETN" message must be the first message following the protocol. * It tells ocfs2_control the local node number. * * Next comes the "SETV" message. It has the following syntax: * * SETV<space><2-char-hex-major><space><2-char-hex-minor><newline> * * This is 11 characters. * * The "SETV" message sets the filesystem locking protocol version as * negotiated by the client. The client negotiates based on the maximum * version advertised in /sys/fs/ocfs2/max_locking_protocol. The major * number from the "SETV" message must match * ocfs2_user_plugin.sp_max_proto.pv_major, and the minor number * must be less than or equal to ...sp_max_version.pv_minor. * * Once this information has been set, mounts will be allowed. From this * point on, the "DOWN" message can be sent for node down notification. * It has the following syntax: * * DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> * * eg: * * DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n * * This is 47 characters. */
/* * Whether or not the client has done the handshake. * For now, we have just one protocol version. */#define OCFS2_CONTROL_PROTO "T01\n"#define OCFS2_CONTROL_PROTO_LEN 4
/* Handshake states */#define OCFS2_CONTROL_HANDSHAKE_INVALID (0)#define OCFS2_CONTROL_HANDSHAKE_READ (1)#define OCFS2_CONTROL_HANDSHAKE_PROTOCOL (2)#define OCFS2_CONTROL_HANDSHAKE_VALID (3)
/* Messages */#define OCFS2_CONTROL_MESSAGE_OP_LEN 4#define OCFS2_CONTROL_MESSAGE_SETNODE_OP "SETN"#define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN 14#define OCFS2_CONTROL_MESSAGE_SETVERSION_OP "SETV"#define OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN 11#define OCFS2_CONTROL_MESSAGE_DOWN_OP "DOWN"#define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN 47#define OCFS2_TEXT_UUID_LEN 32#define OCFS2_CONTROL_MESSAGE_VERNUM_LEN 2#define OCFS2_CONTROL_MESSAGE_NODENUM_LEN 8#define VERSION_LOCK "version_lock"
enum ocfs2_connection_type { WITH_CONTROLD, NO_CONTROLD};
/* * ocfs2_live_connection is refcounted because the filesystem and * miscdevice sides can detach in different order. Let's just be safe. */struct ocfs2_live_connection { struct list_head oc_list; struct ocfs2_cluster_connection *oc_conn; enum ocfs2_connection_type oc_type; atomic_t oc_this_node; int oc_our_slot; struct dlm_lksb oc_version_lksb; char oc_lvb[DLM_LVB_LEN]; struct completion oc_sync_wait; wait_queue_head_t oc_wait;};
struct ocfs2_control_private { struct list_head op_list; int op_state; int op_this_node; struct ocfs2_protocol_version op_proto;};
/* SETN<space><8-char-hex-nodenum><newline> */struct ocfs2_control_message_setn { char tag[OCFS2_CONTROL_MESSAGE_OP_LEN]; char space; char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN]; char newline;};
/* SETV<space><2-char-hex-major><space><2-char-hex-minor><newline> */struct ocfs2_control_message_setv { char tag[OCFS2_CONTROL_MESSAGE_OP_LEN]; char space1; char major[OCFS2_CONTROL_MESSAGE_VERNUM_LEN]; char space2; char minor[OCFS2_CONTROL_MESSAGE_VERNUM_LEN]; char newline;};
/* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */struct ocfs2_control_message_down { char tag[OCFS2_CONTROL_MESSAGE_OP_LEN]; char space1; char uuid[OCFS2_TEXT_UUID_LEN]; char space2; char nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN]; char newline;};
union ocfs2_control_message { char tag[OCFS2_CONTROL_MESSAGE_OP_LEN]; struct ocfs2_control_message_setn u_setn; struct ocfs2_control_message_setv u_setv; struct ocfs2_control_message_down u_down;};
static struct ocfs2_stack_plugin ocfs2_user_plugin;
static atomic_t ocfs2_control_opened;static int ocfs2_control_this_node = -1;static struct ocfs2_protocol_version running_proto;
static LIST_HEAD(ocfs2_live_connection_list);static LIST_HEAD(ocfs2_control_private_list);static DEFINE_MUTEX(ocfs2_control_lock);
static inline void ocfs2_control_set_handshake_state(struct file *file, int state){ struct ocfs2_control_private *p = file->private_data; p->op_state = state;}
static inline int ocfs2_control_get_handshake_state(struct file *file){ struct ocfs2_control_private *p = file->private_data; return p->op_state;}
static struct ocfs2_live_connection *ocfs2_connection_find(const char *name){ size_t len = strlen(name); struct ocfs2_live_connection *c;
BUG_ON(!mutex_is_locked(&ocfs2_control_lock));
list_for_each_entry(c, &ocfs2_live_connection_list, oc_list) { if ((c->oc_conn->cc_namelen == len) && !strncmp(c->oc_conn->cc_name, name, len)) return c; }
return NULL;}
/* * ocfs2_live_connection structures are created underneath the ocfs2 * mount path. Since the VFS prevents multiple calls to * fill_super(), we can't get dupes here. */static int ocfs2_live_connection_attach(struct ocfs2_cluster_connection *conn, struct ocfs2_live_connection *c){ int rc = 0;
mutex_lock(&ocfs2_control_lock); c->oc_conn = conn;
if ((c->oc_type == NO_CONTROLD) || atomic_read(&ocfs2_control_opened)) list_add(&c->oc_list, &ocfs2_live_connection_list); else { printk(KERN_ERR "ocfs2: Userspace control daemon is not present\n"); rc = -ESRCH; }
mutex_unlock(&ocfs2_control_lock); return rc;}
/* * This function disconnects the cluster connection from ocfs2_control. * Afterwards, userspace can't affect the cluster connection. */static void ocfs2_live_connection_drop(struct ocfs2_live_connection *c){ mutex_lock(&ocfs2_control_lock); list_del_init(&c->oc_list); c->oc_conn = NULL; mutex_unlock(&ocfs2_control_lock);
kfree(c);}
static int ocfs2_control_cfu(void *target, size_t target_len, const char __user *buf, size_t count){ /* The T01 expects write(2) calls to have exactly one command */ if ((count != target_len) || (count > sizeof(union ocfs2_control_message))) return -EINVAL;
if (copy_from_user(target, buf, target_len)) return -EFAULT;
return 0;}
static ssize_t ocfs2_control_validate_protocol(struct file *file, const char __user *buf, size_t count){ ssize_t ret; char kbuf[OCFS2_CONTROL_PROTO_LEN];
ret = ocfs2_control_cfu(kbuf, OCFS2_CONTROL_PROTO_LEN, buf, count); if (ret) return ret;
if (strncmp(kbuf, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN)) return -EINVAL;
ocfs2_control_set_handshake_state(file, OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
return count;}
static void ocfs2_control_send_down(const char *uuid, int nodenum){ struct ocfs2_live_connection *c;
mutex_lock(&ocfs2_control_lock);
c = ocfs2_connection_find(uuid); if (c) { BUG_ON(c->oc_conn == NULL); c->oc_conn->cc_recovery_handler(nodenum, c->oc_conn->cc_recovery_data); }
mutex_unlock(&ocfs2_control_lock);}
/* * Called whenever configuration elements are sent to /dev/ocfs2_control. * If all configuration elements are present, try to set the global * values. If there is a problem, return an error. Skip any missing * elements, and only bump ocfs2_control_opened when we have all elements * and are successful. */static int ocfs2_control_install_private(struct file *file){ int rc = 0; int set_p = 1; struct ocfs2_control_private *p = file->private_data;
BUG_ON(p->op_state != OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
mutex_lock(&ocfs2_control_lock);
if (p->op_this_node < 0) { set_p = 0; } else if ((ocfs2_control_this_node >= 0) && (ocfs2_control_this_node != p->op_this_node)) { rc = -EINVAL; goto out_unlock; }
if (!p->op_proto.pv_major) { set_p = 0; } else if (!list_empty(&ocfs2_live_connection_list) && ((running_proto.pv_major != p->op_proto.pv_major) || (running_proto.pv_minor != p->op_proto.pv_minor))) { rc = -EINVAL; goto out_unlock; }
if (set_p) { ocfs2_control_this_node = p->op_this_node; running_proto.pv_major = p->op_proto.pv_major; running_proto.pv_minor = p->op_proto.pv_minor; atomic_inc(&ocfs2_control_opened); ocfs2_control_set_handshake_state(file, OCFS2_CONTROL_HANDSHAKE_VALID); }
out_unlock: mutex_unlock(&ocfs2_control_lock);
return rc;}
static int ocfs2_control_get_this_node(void){ int rc;
mutex_lock(&ocfs2_control_lock); if (ocfs2_control_this_node < 0) rc = -EINVAL; else rc = ocfs2_control_this_node; mutex_unlock(&ocfs2_control_lock);
return rc;}
static int ocfs2_control_do_setnode_msg(struct file *file, struct ocfs2_control_message_setn *msg){ long nodenum; struct ocfs2_control_private *p = file->private_data;
if (ocfs2_control_get_handshake_state(file) != OCFS2_CONTROL_HANDSHAKE_PROTOCOL) return -EINVAL;
if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) return -EINVAL;
if ((msg->space != ' ') || (msg->newline != '\n')) return -EINVAL; msg->space = msg->newline = '\0';
if (kstrtol(msg->nodestr, 16, &nodenum)) return -EINVAL;
if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) || (nodenum > INT_MAX) || (nodenum < 0)) return -ERANGE; p->op_this_node = nodenum;
return ocfs2_control_install_private(file);}
static int ocfs2_control_do_setversion_msg(struct file *file, struct ocfs2_control_message_setv *msg){ long major, minor; struct ocfs2_control_private *p = file->private_data; struct ocfs2_protocol_version *max = &ocfs2_user_plugin.sp_max_proto;
if (ocfs2_control_get_handshake_state(file) != OCFS2_CONTROL_HANDSHAKE_PROTOCOL) return -EINVAL;
if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) return -EINVAL;
if ((msg->space1 != ' ') || (msg->space2 != ' ') || (msg->newline != '\n')) return -EINVAL; msg->space1 = msg->space2 = msg->newline = '\0';
if (kstrtol(msg->major, 16, &major)) return -EINVAL; if (kstrtol(msg->minor, 16, &minor)) return -EINVAL;
/* * The major must be between 1 and 255, inclusive. The minor * must be between 0 and 255, inclusive. The version passed in * must be within the maximum version supported by the filesystem. */ if ((major == LONG_MIN) || (major == LONG_MAX) || (major > (u8)-1) || (major < 1)) return -ERANGE; if ((minor == LONG_MIN) || (minor == LONG_MAX) || (minor > (u8)-1) || (minor < 0)) return -ERANGE; if ((major != max->pv_major) || (minor > max->pv_minor)) return -EINVAL;
p->op_proto.pv_major = major; p->op_proto.pv_minor = minor;
return ocfs2_control_install_private(file);}
static int ocfs2_control_do_down_msg(struct file *file, struct ocfs2_control_message_down *msg){ long nodenum;
if (ocfs2_control_get_handshake_state(file) != OCFS2_CONTROL_HANDSHAKE_VALID) return -EINVAL;
if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_DOWN_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) return -EINVAL;
if ((msg->space1 != ' ') || (msg->space2 != ' ') || (msg->newline != '\n')) return -EINVAL; msg->space1 = msg->space2 = msg->newline = '\0';
if (kstrtol(msg->nodestr, 16, &nodenum)) return -EINVAL;
if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) || (nodenum > INT_MAX) || (nodenum < 0)) return -ERANGE;
ocfs2_control_send_down(msg->uuid, nodenum);
return 0;}
static ssize_t ocfs2_control_message(struct file *file, const char __user *buf, size_t count){ ssize_t ret; union ocfs2_control_message msg;
/* Try to catch padding issues */ WARN_ON(offsetof(struct ocfs2_control_message_down, uuid) != (sizeof(msg.u_down.tag) + sizeof(msg.u_down.space1)));
memset(&msg, 0, sizeof(union ocfs2_control_message)); ret = ocfs2_control_cfu(&msg, count, buf, count); if (ret) goto out;
if ((count == OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN) && !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) ret = ocfs2_control_do_setnode_msg(file, &msg.u_setn); else if ((count == OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN) && !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) ret = ocfs2_control_do_setversion_msg(file, &msg.u_setv); else if ((count == OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN) && !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_DOWN_OP, OCFS2_CONTROL_MESSAGE_OP_LEN)) ret = ocfs2_control_do_down_msg(file, &msg.u_down); else ret = -EINVAL;
out: return ret ? ret : count;}
static ssize_t ocfs2_control_write(struct file *file, const char __user *buf, size_t count, loff_t *ppos){ ssize_t ret;
switch (ocfs2_control_get_handshake_state(file)) { case OCFS2_CONTROL_HANDSHAKE_INVALID: ret = -EINVAL; break;
case OCFS2_CONTROL_HANDSHAKE_READ: ret = ocfs2_control_validate_protocol(file, buf, count); break;
case OCFS2_CONTROL_HANDSHAKE_PROTOCOL: case OCFS2_CONTROL_HANDSHAKE_VALID: ret = ocfs2_control_message(file, buf, count); break;
default: BUG(); ret = -EIO; break; }
return ret;}
/* * This is a naive version. If we ever have a new protocol, we'll expand * it. Probably using seq_file. */static ssize_t ocfs2_control_read(struct file *file, char __user *buf, size_t count, loff_t *ppos){ ssize_t ret;
ret = simple_read_from_buffer(buf, count, ppos, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN);
/* Have we read the whole protocol list? */ if (ret > 0 && *ppos >= OCFS2_CONTROL_PROTO_LEN) ocfs2_control_set_handshake_state(file, OCFS2_CONTROL_HANDSHAKE_READ);
return ret;}
static int ocfs2_control_release(struct inode *inode, struct file *file){ struct ocfs2_control_private *p = file->private_data;
mutex_lock(&ocfs2_control_lock);
if (ocfs2_control_get_handshake_state(file) != OCFS2_CONTROL_HANDSHAKE_VALID) goto out;
if (atomic_dec_and_test(&ocfs2_control_opened)) { if (!list_empty(&ocfs2_live_connection_list)) { /* XXX: Do bad things! */ printk(KERN_ERR "ocfs2: Unexpected release of ocfs2_control!\n" " Loss of cluster connection requires " "an emergency restart!\n"); emergency_restart(); } /* * Last valid close clears the node number and resets * the locking protocol version */ ocfs2_control_this_node = -1; running_proto.pv_major = 0; running_proto.pv_minor = 0; }
out: list_del_init(&p->op_list); file->private_data = NULL;
mutex_unlock(&ocfs2_control_lock);
kfree(p);
return 0;}
static int ocfs2_control_open(struct inode *inode, struct file *file){ struct ocfs2_control_private *p;
p = kzalloc_obj(struct ocfs2_control_private); if (!p) return -ENOMEM; p->op_this_node = -1;
mutex_lock(&ocfs2_control_lock); file->private_data = p; list_add(&p->op_list, &ocfs2_control_private_list); mutex_unlock(&ocfs2_control_lock);
return 0;}
static const struct file_operations ocfs2_control_fops = { .open = ocfs2_control_open, .release = ocfs2_control_release, .read = ocfs2_control_read, .write = ocfs2_control_write, .owner = THIS_MODULE, .llseek = default_llseek,};
static struct miscdevice ocfs2_control_device = { .minor = MISC_DYNAMIC_MINOR, .name = "ocfs2_control", .fops = &ocfs2_control_fops,};
static int ocfs2_control_init(void){ int rc;
atomic_set(&ocfs2_control_opened, 0);
rc = misc_register(&ocfs2_control_device); if (rc) printk(KERN_ERR "ocfs2: Unable to register ocfs2_control device " "(errno %d)\n", -rc);
return rc;}
static void ocfs2_control_exit(void){ misc_deregister(&ocfs2_control_device);}
static void fsdlm_lock_ast_wrapper(void *astarg){ struct ocfs2_dlm_lksb *lksb = astarg; int status = lksb->lksb_fsdlm.sb_status;
/* * For now we're punting on the issue of other non-standard errors * where we can't tell if the unlock_ast or lock_ast should be called. * The main "other error" that's possible is EINVAL which means the * function was called with invalid args, which shouldn't be possible * since the caller here is under our control. Other non-standard * errors probably fall into the same category, or otherwise are fatal * which means we can't carry on anyway. */
if (status == -DLM_EUNLOCK || status == -DLM_ECANCEL) lksb->lksb_conn->cc_proto->lp_unlock_ast(lksb, 0); else lksb->lksb_conn->cc_proto->lp_lock_ast(lksb);}
static void fsdlm_blocking_ast_wrapper(void *astarg, int level){ struct ocfs2_dlm_lksb *lksb = astarg;
lksb->lksb_conn->cc_proto->lp_blocking_ast(lksb, level);}
static int user_dlm_lock(struct ocfs2_cluster_connection *conn, int mode, struct ocfs2_dlm_lksb *lksb, u32 flags, void *name, unsigned int namelen){ if (!lksb->lksb_fsdlm.sb_lvbptr) lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb + sizeof(struct dlm_lksb);
return dlm_lock(conn->cc_lockspace, mode, &lksb->lksb_fsdlm, flags|DLM_LKF_NODLCKWT, name, namelen, 0, fsdlm_lock_ast_wrapper, lksb, fsdlm_blocking_ast_wrapper);}
static int user_dlm_unlock(struct ocfs2_cluster_connection *conn, struct ocfs2_dlm_lksb *lksb, u32 flags){ return dlm_unlock(conn->cc_lockspace, lksb->lksb_fsdlm.sb_lkid, flags, &lksb->lksb_fsdlm, lksb);}
static int user_dlm_lock_status(struct ocfs2_dlm_lksb *lksb){ return lksb->lksb_fsdlm.sb_status;}
static int user_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb){ int invalid = lksb->lksb_fsdlm.sb_flags & DLM_SBF_VALNOTVALID;
return !invalid;}
static void *user_dlm_lvb(struct ocfs2_dlm_lksb *lksb){ if (!lksb->lksb_fsdlm.sb_lvbptr) lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb + sizeof(struct dlm_lksb); return (void *)(lksb->lksb_fsdlm.sb_lvbptr);}
static void user_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb){}
static int user_plock(struct ocfs2_cluster_connection *conn, u64 ino, struct file *file, int cmd, struct file_lock *fl){ /* * This more or less just demuxes the plock request into any * one of three dlm calls. * * Internally, fs/dlm will pass these to a misc device, which * a userspace daemon will read and write to. */
if (cmd == F_CANCELLK) return dlm_posix_cancel(conn->cc_lockspace, ino, file, fl); else if (IS_GETLK(cmd)) return dlm_posix_get(conn->cc_lockspace, ino, file, fl); else if (lock_is_unlock(fl)) return dlm_posix_unlock(conn->cc_lockspace, ino, file, fl); else return dlm_posix_lock(conn->cc_lockspace, ino, file, cmd, fl);}
/* * Compare a requested locking protocol version against the current one. * * If the major numbers are different, they are incompatible. * If the current minor is greater than the request, they are incompatible. * If the current minor is less than or equal to the request, they are * compatible, and the requester should run at the current minor version. */static int fs_protocol_compare(struct ocfs2_protocol_version *existing, struct ocfs2_protocol_version *request){ if (existing->pv_major != request->pv_major) return 1;
if (existing->pv_minor > request->pv_minor) return 1;
if (existing->pv_minor < request->pv_minor) request->pv_minor = existing->pv_minor;
return 0;}
static void lvb_to_version(char *lvb, struct ocfs2_protocol_version *ver){ struct ocfs2_protocol_version *pv = (struct ocfs2_protocol_version *)lvb; /* * ocfs2_protocol_version has two u8 variables, so we don't * need any endian conversion. */ ver->pv_major = pv->pv_major; ver->pv_minor = pv->pv_minor;}
static void version_to_lvb(struct ocfs2_protocol_version *ver, char *lvb){ struct ocfs2_protocol_version *pv = (struct ocfs2_protocol_version *)lvb; /* * ocfs2_protocol_version has two u8 variables, so we don't * need any endian conversion. */ pv->pv_major = ver->pv_major; pv->pv_minor = ver->pv_minor;}
static void sync_wait_cb(void *arg){ struct ocfs2_cluster_connection *conn = arg; struct ocfs2_live_connection *lc = conn->cc_private; complete(&lc->oc_sync_wait);}
static int sync_unlock(struct ocfs2_cluster_connection *conn, struct dlm_lksb *lksb, char *name){ int error; struct ocfs2_live_connection *lc = conn->cc_private;
error = dlm_unlock(conn->cc_lockspace, lksb->sb_lkid, 0, lksb, conn); if (error) { printk(KERN_ERR "%s lkid %x error %d\n", name, lksb->sb_lkid, error); return error; }
wait_for_completion(&lc->oc_sync_wait);
if (lksb->sb_status != -DLM_EUNLOCK) { printk(KERN_ERR "%s lkid %x status %d\n", name, lksb->sb_lkid, lksb->sb_status); return -1; } return 0;}
static int sync_lock(struct ocfs2_cluster_connection *conn, int mode, uint32_t flags, struct dlm_lksb *lksb, char *name){ int error, status; struct ocfs2_live_connection *lc = conn->cc_private;
error = dlm_lock(conn->cc_lockspace, mode, lksb, flags, name, strlen(name), 0, sync_wait_cb, conn, NULL); if (error) { printk(KERN_ERR "%s lkid %x flags %x mode %d error %d\n", name, lksb->sb_lkid, flags, mode, error); return error; }
wait_for_completion(&lc->oc_sync_wait);
status = lksb->sb_status;
if (status && status != -EAGAIN) { printk(KERN_ERR "%s lkid %x flags %x mode %d status %d\n", name, lksb->sb_lkid, flags, mode, status); }
return status;}
static int version_lock(struct ocfs2_cluster_connection *conn, int mode, int flags){ struct ocfs2_live_connection *lc = conn->cc_private; return sync_lock(conn, mode, flags, &lc->oc_version_lksb, VERSION_LOCK);}
static int version_unlock(struct ocfs2_cluster_connection *conn){ struct ocfs2_live_connection *lc = conn->cc_private; return sync_unlock(conn, &lc->oc_version_lksb, VERSION_LOCK);}
/* get_protocol_version() * * To exchange ocfs2 versioning, we use the LVB of the version dlm lock. * The algorithm is: * 1. Attempt to take the lock in EX mode (non-blocking). * 2. If successful (which means it is the first mount), write the * version number and downconvert to PR lock. * 3. If unsuccessful (returns -EAGAIN), read the version from the LVB after * taking the PR lock. */
static int get_protocol_version(struct ocfs2_cluster_connection *conn){ int ret; struct ocfs2_live_connection *lc = conn->cc_private; struct ocfs2_protocol_version pv;
running_proto.pv_major = ocfs2_user_plugin.sp_max_proto.pv_major; running_proto.pv_minor = ocfs2_user_plugin.sp_max_proto.pv_minor;
lc->oc_version_lksb.sb_lvbptr = lc->oc_lvb; ret = version_lock(conn, DLM_LOCK_EX, DLM_LKF_VALBLK|DLM_LKF_NOQUEUE); if (!ret) { conn->cc_version.pv_major = running_proto.pv_major; conn->cc_version.pv_minor = running_proto.pv_minor; version_to_lvb(&running_proto, lc->oc_lvb); version_lock(conn, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_VALBLK); } else if (ret == -EAGAIN) { ret = version_lock(conn, DLM_LOCK_PR, DLM_LKF_VALBLK); if (ret) goto out; lvb_to_version(lc->oc_lvb, &pv);
if ((pv.pv_major != running_proto.pv_major) || (pv.pv_minor > running_proto.pv_minor)) { ret = -EINVAL; goto out; }
conn->cc_version.pv_major = pv.pv_major; conn->cc_version.pv_minor = pv.pv_minor; }out: return ret;}
static void user_recover_prep(void *arg){}
static void user_recover_slot(void *arg, struct dlm_slot *slot){ struct ocfs2_cluster_connection *conn = arg; printk(KERN_INFO "ocfs2: Node %d/%d down. Initiating recovery.\n", slot->nodeid, slot->slot); conn->cc_recovery_handler(slot->nodeid, conn->cc_recovery_data);
}
static void user_recover_done(void *arg, struct dlm_slot *slots, int num_slots, int our_slot, uint32_t generation){ struct ocfs2_cluster_connection *conn = arg; struct ocfs2_live_connection *lc = conn->cc_private; int i;
for (i = 0; i < num_slots; i++) if (slots[i].slot == our_slot) { atomic_set(&lc->oc_this_node, slots[i].nodeid); break; }
lc->oc_our_slot = our_slot; wake_up(&lc->oc_wait);}
static const struct dlm_lockspace_ops ocfs2_ls_ops = { .recover_prep = user_recover_prep, .recover_slot = user_recover_slot, .recover_done = user_recover_done,};
static int user_cluster_disconnect(struct ocfs2_cluster_connection *conn){ version_unlock(conn); dlm_release_lockspace(conn->cc_lockspace, DLM_RELEASE_NORMAL); conn->cc_lockspace = NULL; ocfs2_live_connection_drop(conn->cc_private); conn->cc_private = NULL; return 0;}
static int user_cluster_connect(struct ocfs2_cluster_connection *conn){ dlm_lockspace_t *fsdlm; struct ocfs2_live_connection *lc; int rc, ops_rv;
BUG_ON(conn == NULL);
lc = kzalloc_obj(struct ocfs2_live_connection); if (!lc) return -ENOMEM;
init_waitqueue_head(&lc->oc_wait); init_completion(&lc->oc_sync_wait); atomic_set(&lc->oc_this_node, 0); conn->cc_private = lc; lc->oc_type = NO_CONTROLD;
rc = dlm_new_lockspace(conn->cc_name, conn->cc_cluster_name, DLM_LSFL_NEWEXCL, DLM_LVB_LEN, &ocfs2_ls_ops, conn, &ops_rv, &fsdlm); if (rc) { if (rc == -EEXIST || rc == -EPROTO) printk(KERN_ERR "ocfs2: Unable to create the " "lockspace %s (%d), because a ocfs2-tools " "program is running on this file system " "with the same name lockspace\n", conn->cc_name, rc); goto out; }
if (ops_rv == -EOPNOTSUPP) { lc->oc_type = WITH_CONTROLD; printk(KERN_NOTICE "ocfs2: You seem to be using an older " "version of dlm_controld and/or ocfs2-tools." " Please consider upgrading.\n"); } else if (ops_rv) { rc = ops_rv; goto out; } conn->cc_lockspace = fsdlm;
rc = ocfs2_live_connection_attach(conn, lc); if (rc) goto out;
if (lc->oc_type == NO_CONTROLD) { rc = get_protocol_version(conn); if (rc) { printk(KERN_ERR "ocfs2: Could not determine" " locking version\n"); user_cluster_disconnect(conn); lc = NULL; goto out; } wait_event(lc->oc_wait, (atomic_read(&lc->oc_this_node) > 0)); }
/* * running_proto must have been set before we allowed any mounts * to proceed. */ if (fs_protocol_compare(&running_proto, &conn->cc_version)) { printk(KERN_ERR "Unable to mount with fs locking protocol version " "%u.%u because negotiated protocol is %u.%u\n", conn->cc_version.pv_major, conn->cc_version.pv_minor, running_proto.pv_major, running_proto.pv_minor); rc = -EPROTO; ocfs2_live_connection_drop(lc); lc = NULL; }
out: if (rc) kfree(lc); return rc;}
static int user_cluster_this_node(struct ocfs2_cluster_connection *conn, unsigned int *this_node){ int rc; struct ocfs2_live_connection *lc = conn->cc_private;
if (lc->oc_type == WITH_CONTROLD) rc = ocfs2_control_get_this_node(); else if (lc->oc_type == NO_CONTROLD) rc = atomic_read(&lc->oc_this_node); else rc = -EINVAL;
if (rc < 0) return rc;
*this_node = rc; return 0;}
static const struct ocfs2_stack_operations ocfs2_user_plugin_ops = { .connect = user_cluster_connect, .disconnect = user_cluster_disconnect, .this_node = user_cluster_this_node, .dlm_lock = user_dlm_lock, .dlm_unlock = user_dlm_unlock, .lock_status = user_dlm_lock_status, .lvb_valid = user_dlm_lvb_valid, .lock_lvb = user_dlm_lvb, .plock = user_plock, .dump_lksb = user_dlm_dump_lksb,};
static struct ocfs2_stack_plugin ocfs2_user_plugin = { .sp_name = "user", .sp_ops = &ocfs2_user_plugin_ops, .sp_owner = THIS_MODULE,};
static int __init ocfs2_user_plugin_init(void){ int rc;
rc = ocfs2_control_init(); if (!rc) { rc = ocfs2_stack_glue_register(&ocfs2_user_plugin); if (rc) ocfs2_control_exit(); }
return rc;}
static void __exit ocfs2_user_plugin_exit(void){ ocfs2_stack_glue_unregister(&ocfs2_user_plugin); ocfs2_control_exit();}
MODULE_AUTHOR("Oracle");MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");MODULE_LICENSE("GPL");module_init(ocfs2_user_plugin_init);module_exit(ocfs2_user_plugin_exit);