~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/fs/ocfs2/stack_user.c

Version: ~ [ linux-5.12-rc7 ] ~ [ linux-5.11.13 ] ~ [ linux-5.10.29 ] ~ [ linux-5.9.16 ] ~ [ linux-5.8.18 ] ~ [ linux-5.7.19 ] ~ [ linux-5.6.19 ] ~ [ linux-5.5.19 ] ~ [ linux-5.4.111 ] ~ [ linux-5.3.18 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.186 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.230 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.266 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.266 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.18.140 ] ~ [ linux-3.16.85 ] ~ [ linux-3.14.79 ] ~ [ linux-3.12.74 ] ~ [ linux-3.10.108 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

  1 // SPDX-License-Identifier: GPL-2.0-only
  2 /* -*- mode: c; c-basic-offset: 8; -*-
  3  * vim: noexpandtab sw=8 ts=8 sts=0:
  4  *
  5  * stack_user.c
  6  *
  7  * Code which interfaces ocfs2 with fs/dlm and a userspace stack.
  8  *
  9  * Copyright (C) 2007 Oracle.  All rights reserved.
 10  */
 11 
 12 #include <linux/module.h>
 13 #include <linux/fs.h>
 14 #include <linux/miscdevice.h>
 15 #include <linux/mutex.h>
 16 #include <linux/slab.h>
 17 #include <linux/reboot.h>
 18 #include <linux/sched.h>
 19 #include <linux/uaccess.h>
 20 
 21 #include "stackglue.h"
 22 
 23 #include <linux/dlm_plock.h>
 24 
 25 /*
 26  * The control protocol starts with a handshake.  Until the handshake
 27  * is complete, the control device will fail all write(2)s.
 28  *
 29  * The handshake is simple.  First, the client reads until EOF.  Each line
 30  * of output is a supported protocol tag.  All protocol tags are a single
 31  * character followed by a two hex digit version number.  Currently the
 32  * only things supported is T01, for "Text-base version 0x01".  Next, the
 33  * client writes the version they would like to use, including the newline.
 34  * Thus, the protocol tag is 'T01\n'.  If the version tag written is
 35  * unknown, -EINVAL is returned.  Once the negotiation is complete, the
 36  * client can start sending messages.
 37  *
 38  * The T01 protocol has three messages.  First is the "SETN" message.
 39  * It has the following syntax:
 40  *
 41  *  SETN<space><8-char-hex-nodenum><newline>
 42  *
 43  * This is 14 characters.
 44  *
 45  * The "SETN" message must be the first message following the protocol.
 46  * It tells ocfs2_control the local node number.
 47  *
 48  * Next comes the "SETV" message.  It has the following syntax:
 49  *
 50  *  SETV<space><2-char-hex-major><space><2-char-hex-minor><newline>
 51  *
 52  * This is 11 characters.
 53  *
 54  * The "SETV" message sets the filesystem locking protocol version as
 55  * negotiated by the client.  The client negotiates based on the maximum
 56  * version advertised in /sys/fs/ocfs2/max_locking_protocol.  The major
 57  * number from the "SETV" message must match
 58  * ocfs2_user_plugin.sp_max_proto.pv_major, and the minor number
 59  * must be less than or equal to ...sp_max_version.pv_minor.
 60  *
 61  * Once this information has been set, mounts will be allowed.  From this
 62  * point on, the "DOWN" message can be sent for node down notification.
 63  * It has the following syntax:
 64  *
 65  *  DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline>
 66  *
 67  * eg:
 68  *
 69  *  DOWN 632A924FDD844190BDA93C0DF6B94899 00000001\n
 70  *
 71  * This is 47 characters.
 72  */
 73 
 74 /*
 75  * Whether or not the client has done the handshake.
 76  * For now, we have just one protocol version.
 77  */
 78 #define OCFS2_CONTROL_PROTO                     "T01\n"
 79 #define OCFS2_CONTROL_PROTO_LEN                 4
 80 
 81 /* Handshake states */
 82 #define OCFS2_CONTROL_HANDSHAKE_INVALID         (0)
 83 #define OCFS2_CONTROL_HANDSHAKE_READ            (1)
 84 #define OCFS2_CONTROL_HANDSHAKE_PROTOCOL        (2)
 85 #define OCFS2_CONTROL_HANDSHAKE_VALID           (3)
 86 
 87 /* Messages */
 88 #define OCFS2_CONTROL_MESSAGE_OP_LEN            4
 89 #define OCFS2_CONTROL_MESSAGE_SETNODE_OP        "SETN"
 90 #define OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN 14
 91 #define OCFS2_CONTROL_MESSAGE_SETVERSION_OP     "SETV"
 92 #define OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN      11
 93 #define OCFS2_CONTROL_MESSAGE_DOWN_OP           "DOWN"
 94 #define OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN    47
 95 #define OCFS2_TEXT_UUID_LEN                     32
 96 #define OCFS2_CONTROL_MESSAGE_VERNUM_LEN        2
 97 #define OCFS2_CONTROL_MESSAGE_NODENUM_LEN       8
 98 #define VERSION_LOCK                            "version_lock"
 99 
100 enum ocfs2_connection_type {
101         WITH_CONTROLD,
102         NO_CONTROLD
103 };
104 
105 /*
106  * ocfs2_live_connection is refcounted because the filesystem and
107  * miscdevice sides can detach in different order.  Let's just be safe.
108  */
109 struct ocfs2_live_connection {
110         struct list_head                oc_list;
111         struct ocfs2_cluster_connection *oc_conn;
112         enum ocfs2_connection_type      oc_type;
113         atomic_t                        oc_this_node;
114         int                             oc_our_slot;
115         struct dlm_lksb                 oc_version_lksb;
116         char                            oc_lvb[DLM_LVB_LEN];
117         struct completion               oc_sync_wait;
118         wait_queue_head_t               oc_wait;
119 };
120 
121 struct ocfs2_control_private {
122         struct list_head op_list;
123         int op_state;
124         int op_this_node;
125         struct ocfs2_protocol_version op_proto;
126 };
127 
128 /* SETN<space><8-char-hex-nodenum><newline> */
129 struct ocfs2_control_message_setn {
130         char    tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
131         char    space;
132         char    nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
133         char    newline;
134 };
135 
136 /* SETV<space><2-char-hex-major><space><2-char-hex-minor><newline> */
137 struct ocfs2_control_message_setv {
138         char    tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
139         char    space1;
140         char    major[OCFS2_CONTROL_MESSAGE_VERNUM_LEN];
141         char    space2;
142         char    minor[OCFS2_CONTROL_MESSAGE_VERNUM_LEN];
143         char    newline;
144 };
145 
146 /* DOWN<space><32-char-cap-hex-uuid><space><8-char-hex-nodenum><newline> */
147 struct ocfs2_control_message_down {
148         char    tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
149         char    space1;
150         char    uuid[OCFS2_TEXT_UUID_LEN];
151         char    space2;
152         char    nodestr[OCFS2_CONTROL_MESSAGE_NODENUM_LEN];
153         char    newline;
154 };
155 
156 union ocfs2_control_message {
157         char                                    tag[OCFS2_CONTROL_MESSAGE_OP_LEN];
158         struct ocfs2_control_message_setn       u_setn;
159         struct ocfs2_control_message_setv       u_setv;
160         struct ocfs2_control_message_down       u_down;
161 };
162 
163 static struct ocfs2_stack_plugin ocfs2_user_plugin;
164 
165 static atomic_t ocfs2_control_opened;
166 static int ocfs2_control_this_node = -1;
167 static struct ocfs2_protocol_version running_proto;
168 
169 static LIST_HEAD(ocfs2_live_connection_list);
170 static LIST_HEAD(ocfs2_control_private_list);
171 static DEFINE_MUTEX(ocfs2_control_lock);
172 
173 static inline void ocfs2_control_set_handshake_state(struct file *file,
174                                                      int state)
175 {
176         struct ocfs2_control_private *p = file->private_data;
177         p->op_state = state;
178 }
179 
180 static inline int ocfs2_control_get_handshake_state(struct file *file)
181 {
182         struct ocfs2_control_private *p = file->private_data;
183         return p->op_state;
184 }
185 
186 static struct ocfs2_live_connection *ocfs2_connection_find(const char *name)
187 {
188         size_t len = strlen(name);
189         struct ocfs2_live_connection *c;
190 
191         BUG_ON(!mutex_is_locked(&ocfs2_control_lock));
192 
193         list_for_each_entry(c, &ocfs2_live_connection_list, oc_list) {
194                 if ((c->oc_conn->cc_namelen == len) &&
195                     !strncmp(c->oc_conn->cc_name, name, len))
196                         return c;
197         }
198 
199         return NULL;
200 }
201 
202 /*
203  * ocfs2_live_connection structures are created underneath the ocfs2
204  * mount path.  Since the VFS prevents multiple calls to
205  * fill_super(), we can't get dupes here.
206  */
207 static int ocfs2_live_connection_attach(struct ocfs2_cluster_connection *conn,
208                                      struct ocfs2_live_connection *c)
209 {
210         int rc = 0;
211 
212         mutex_lock(&ocfs2_control_lock);
213         c->oc_conn = conn;
214 
215         if ((c->oc_type == NO_CONTROLD) || atomic_read(&ocfs2_control_opened))
216                 list_add(&c->oc_list, &ocfs2_live_connection_list);
217         else {
218                 printk(KERN_ERR
219                        "ocfs2: Userspace control daemon is not present\n");
220                 rc = -ESRCH;
221         }
222 
223         mutex_unlock(&ocfs2_control_lock);
224         return rc;
225 }
226 
227 /*
228  * This function disconnects the cluster connection from ocfs2_control.
229  * Afterwards, userspace can't affect the cluster connection.
230  */
231 static void ocfs2_live_connection_drop(struct ocfs2_live_connection *c)
232 {
233         mutex_lock(&ocfs2_control_lock);
234         list_del_init(&c->oc_list);
235         c->oc_conn = NULL;
236         mutex_unlock(&ocfs2_control_lock);
237 
238         kfree(c);
239 }
240 
241 static int ocfs2_control_cfu(void *target, size_t target_len,
242                              const char __user *buf, size_t count)
243 {
244         /* The T01 expects write(2) calls to have exactly one command */
245         if ((count != target_len) ||
246             (count > sizeof(union ocfs2_control_message)))
247                 return -EINVAL;
248 
249         if (copy_from_user(target, buf, target_len))
250                 return -EFAULT;
251 
252         return 0;
253 }
254 
255 static ssize_t ocfs2_control_validate_protocol(struct file *file,
256                                                const char __user *buf,
257                                                size_t count)
258 {
259         ssize_t ret;
260         char kbuf[OCFS2_CONTROL_PROTO_LEN];
261 
262         ret = ocfs2_control_cfu(kbuf, OCFS2_CONTROL_PROTO_LEN,
263                                 buf, count);
264         if (ret)
265                 return ret;
266 
267         if (strncmp(kbuf, OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN))
268                 return -EINVAL;
269 
270         ocfs2_control_set_handshake_state(file,
271                                           OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
272 
273         return count;
274 }
275 
276 static void ocfs2_control_send_down(const char *uuid,
277                                     int nodenum)
278 {
279         struct ocfs2_live_connection *c;
280 
281         mutex_lock(&ocfs2_control_lock);
282 
283         c = ocfs2_connection_find(uuid);
284         if (c) {
285                 BUG_ON(c->oc_conn == NULL);
286                 c->oc_conn->cc_recovery_handler(nodenum,
287                                                 c->oc_conn->cc_recovery_data);
288         }
289 
290         mutex_unlock(&ocfs2_control_lock);
291 }
292 
293 /*
294  * Called whenever configuration elements are sent to /dev/ocfs2_control.
295  * If all configuration elements are present, try to set the global
296  * values.  If there is a problem, return an error.  Skip any missing
297  * elements, and only bump ocfs2_control_opened when we have all elements
298  * and are successful.
299  */
300 static int ocfs2_control_install_private(struct file *file)
301 {
302         int rc = 0;
303         int set_p = 1;
304         struct ocfs2_control_private *p = file->private_data;
305 
306         BUG_ON(p->op_state != OCFS2_CONTROL_HANDSHAKE_PROTOCOL);
307 
308         mutex_lock(&ocfs2_control_lock);
309 
310         if (p->op_this_node < 0) {
311                 set_p = 0;
312         } else if ((ocfs2_control_this_node >= 0) &&
313                    (ocfs2_control_this_node != p->op_this_node)) {
314                 rc = -EINVAL;
315                 goto out_unlock;
316         }
317 
318         if (!p->op_proto.pv_major) {
319                 set_p = 0;
320         } else if (!list_empty(&ocfs2_live_connection_list) &&
321                    ((running_proto.pv_major != p->op_proto.pv_major) ||
322                     (running_proto.pv_minor != p->op_proto.pv_minor))) {
323                 rc = -EINVAL;
324                 goto out_unlock;
325         }
326 
327         if (set_p) {
328                 ocfs2_control_this_node = p->op_this_node;
329                 running_proto.pv_major = p->op_proto.pv_major;
330                 running_proto.pv_minor = p->op_proto.pv_minor;
331         }
332 
333 out_unlock:
334         mutex_unlock(&ocfs2_control_lock);
335 
336         if (!rc && set_p) {
337                 /* We set the global values successfully */
338                 atomic_inc(&ocfs2_control_opened);
339                 ocfs2_control_set_handshake_state(file,
340                                         OCFS2_CONTROL_HANDSHAKE_VALID);
341         }
342 
343         return rc;
344 }
345 
346 static int ocfs2_control_get_this_node(void)
347 {
348         int rc;
349 
350         mutex_lock(&ocfs2_control_lock);
351         if (ocfs2_control_this_node < 0)
352                 rc = -EINVAL;
353         else
354                 rc = ocfs2_control_this_node;
355         mutex_unlock(&ocfs2_control_lock);
356 
357         return rc;
358 }
359 
360 static int ocfs2_control_do_setnode_msg(struct file *file,
361                                         struct ocfs2_control_message_setn *msg)
362 {
363         long nodenum;
364         char *ptr = NULL;
365         struct ocfs2_control_private *p = file->private_data;
366 
367         if (ocfs2_control_get_handshake_state(file) !=
368             OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
369                 return -EINVAL;
370 
371         if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
372                     OCFS2_CONTROL_MESSAGE_OP_LEN))
373                 return -EINVAL;
374 
375         if ((msg->space != ' ') || (msg->newline != '\n'))
376                 return -EINVAL;
377         msg->space = msg->newline = '\0';
378 
379         nodenum = simple_strtol(msg->nodestr, &ptr, 16);
380         if (!ptr || *ptr)
381                 return -EINVAL;
382 
383         if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
384             (nodenum > INT_MAX) || (nodenum < 0))
385                 return -ERANGE;
386         p->op_this_node = nodenum;
387 
388         return ocfs2_control_install_private(file);
389 }
390 
391 static int ocfs2_control_do_setversion_msg(struct file *file,
392                                            struct ocfs2_control_message_setv *msg)
393 {
394         long major, minor;
395         char *ptr = NULL;
396         struct ocfs2_control_private *p = file->private_data;
397         struct ocfs2_protocol_version *max =
398                 &ocfs2_user_plugin.sp_max_proto;
399 
400         if (ocfs2_control_get_handshake_state(file) !=
401             OCFS2_CONTROL_HANDSHAKE_PROTOCOL)
402                 return -EINVAL;
403 
404         if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP,
405                     OCFS2_CONTROL_MESSAGE_OP_LEN))
406                 return -EINVAL;
407 
408         if ((msg->space1 != ' ') || (msg->space2 != ' ') ||
409             (msg->newline != '\n'))
410                 return -EINVAL;
411         msg->space1 = msg->space2 = msg->newline = '\0';
412 
413         major = simple_strtol(msg->major, &ptr, 16);
414         if (!ptr || *ptr)
415                 return -EINVAL;
416         minor = simple_strtol(msg->minor, &ptr, 16);
417         if (!ptr || *ptr)
418                 return -EINVAL;
419 
420         /*
421          * The major must be between 1 and 255, inclusive.  The minor
422          * must be between 0 and 255, inclusive.  The version passed in
423          * must be within the maximum version supported by the filesystem.
424          */
425         if ((major == LONG_MIN) || (major == LONG_MAX) ||
426             (major > (u8)-1) || (major < 1))
427                 return -ERANGE;
428         if ((minor == LONG_MIN) || (minor == LONG_MAX) ||
429             (minor > (u8)-1) || (minor < 0))
430                 return -ERANGE;
431         if ((major != max->pv_major) ||
432             (minor > max->pv_minor))
433                 return -EINVAL;
434 
435         p->op_proto.pv_major = major;
436         p->op_proto.pv_minor = minor;
437 
438         return ocfs2_control_install_private(file);
439 }
440 
441 static int ocfs2_control_do_down_msg(struct file *file,
442                                      struct ocfs2_control_message_down *msg)
443 {
444         long nodenum;
445         char *p = NULL;
446 
447         if (ocfs2_control_get_handshake_state(file) !=
448             OCFS2_CONTROL_HANDSHAKE_VALID)
449                 return -EINVAL;
450 
451         if (strncmp(msg->tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
452                     OCFS2_CONTROL_MESSAGE_OP_LEN))
453                 return -EINVAL;
454 
455         if ((msg->space1 != ' ') || (msg->space2 != ' ') ||
456             (msg->newline != '\n'))
457                 return -EINVAL;
458         msg->space1 = msg->space2 = msg->newline = '\0';
459 
460         nodenum = simple_strtol(msg->nodestr, &p, 16);
461         if (!p || *p)
462                 return -EINVAL;
463 
464         if ((nodenum == LONG_MIN) || (nodenum == LONG_MAX) ||
465             (nodenum > INT_MAX) || (nodenum < 0))
466                 return -ERANGE;
467 
468         ocfs2_control_send_down(msg->uuid, nodenum);
469 
470         return 0;
471 }
472 
473 static ssize_t ocfs2_control_message(struct file *file,
474                                      const char __user *buf,
475                                      size_t count)
476 {
477         ssize_t ret;
478         union ocfs2_control_message msg;
479 
480         /* Try to catch padding issues */
481         WARN_ON(offsetof(struct ocfs2_control_message_down, uuid) !=
482                 (sizeof(msg.u_down.tag) + sizeof(msg.u_down.space1)));
483 
484         memset(&msg, 0, sizeof(union ocfs2_control_message));
485         ret = ocfs2_control_cfu(&msg, count, buf, count);
486         if (ret)
487                 goto out;
488 
489         if ((count == OCFS2_CONTROL_MESSAGE_SETNODE_TOTAL_LEN) &&
490             !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETNODE_OP,
491                      OCFS2_CONTROL_MESSAGE_OP_LEN))
492                 ret = ocfs2_control_do_setnode_msg(file, &msg.u_setn);
493         else if ((count == OCFS2_CONTROL_MESSAGE_SETVERSION_TOTAL_LEN) &&
494                  !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_SETVERSION_OP,
495                           OCFS2_CONTROL_MESSAGE_OP_LEN))
496                 ret = ocfs2_control_do_setversion_msg(file, &msg.u_setv);
497         else if ((count == OCFS2_CONTROL_MESSAGE_DOWN_TOTAL_LEN) &&
498                  !strncmp(msg.tag, OCFS2_CONTROL_MESSAGE_DOWN_OP,
499                           OCFS2_CONTROL_MESSAGE_OP_LEN))
500                 ret = ocfs2_control_do_down_msg(file, &msg.u_down);
501         else
502                 ret = -EINVAL;
503 
504 out:
505         return ret ? ret : count;
506 }
507 
508 static ssize_t ocfs2_control_write(struct file *file,
509                                    const char __user *buf,
510                                    size_t count,
511                                    loff_t *ppos)
512 {
513         ssize_t ret;
514 
515         switch (ocfs2_control_get_handshake_state(file)) {
516                 case OCFS2_CONTROL_HANDSHAKE_INVALID:
517                         ret = -EINVAL;
518                         break;
519 
520                 case OCFS2_CONTROL_HANDSHAKE_READ:
521                         ret = ocfs2_control_validate_protocol(file, buf,
522                                                               count);
523                         break;
524 
525                 case OCFS2_CONTROL_HANDSHAKE_PROTOCOL:
526                 case OCFS2_CONTROL_HANDSHAKE_VALID:
527                         ret = ocfs2_control_message(file, buf, count);
528                         break;
529 
530                 default:
531                         BUG();
532                         ret = -EIO;
533                         break;
534         }
535 
536         return ret;
537 }
538 
539 /*
540  * This is a naive version.  If we ever have a new protocol, we'll expand
541  * it.  Probably using seq_file.
542  */
543 static ssize_t ocfs2_control_read(struct file *file,
544                                   char __user *buf,
545                                   size_t count,
546                                   loff_t *ppos)
547 {
548         ssize_t ret;
549 
550         ret = simple_read_from_buffer(buf, count, ppos,
551                         OCFS2_CONTROL_PROTO, OCFS2_CONTROL_PROTO_LEN);
552 
553         /* Have we read the whole protocol list? */
554         if (ret > 0 && *ppos >= OCFS2_CONTROL_PROTO_LEN)
555                 ocfs2_control_set_handshake_state(file,
556                                                   OCFS2_CONTROL_HANDSHAKE_READ);
557 
558         return ret;
559 }
560 
561 static int ocfs2_control_release(struct inode *inode, struct file *file)
562 {
563         struct ocfs2_control_private *p = file->private_data;
564 
565         mutex_lock(&ocfs2_control_lock);
566 
567         if (ocfs2_control_get_handshake_state(file) !=
568             OCFS2_CONTROL_HANDSHAKE_VALID)
569                 goto out;
570 
571         if (atomic_dec_and_test(&ocfs2_control_opened)) {
572                 if (!list_empty(&ocfs2_live_connection_list)) {
573                         /* XXX: Do bad things! */
574                         printk(KERN_ERR
575                                "ocfs2: Unexpected release of ocfs2_control!\n"
576                                "       Loss of cluster connection requires "
577                                "an emergency restart!\n");
578                         emergency_restart();
579                 }
580                 /*
581                  * Last valid close clears the node number and resets
582                  * the locking protocol version
583                  */
584                 ocfs2_control_this_node = -1;
585                 running_proto.pv_major = 0;
586                 running_proto.pv_minor = 0;
587         }
588 
589 out:
590         list_del_init(&p->op_list);
591         file->private_data = NULL;
592 
593         mutex_unlock(&ocfs2_control_lock);
594 
595         kfree(p);
596 
597         return 0;
598 }
599 
600 static int ocfs2_control_open(struct inode *inode, struct file *file)
601 {
602         struct ocfs2_control_private *p;
603 
604         p = kzalloc(sizeof(struct ocfs2_control_private), GFP_KERNEL);
605         if (!p)
606                 return -ENOMEM;
607         p->op_this_node = -1;
608 
609         mutex_lock(&ocfs2_control_lock);
610         file->private_data = p;
611         list_add(&p->op_list, &ocfs2_control_private_list);
612         mutex_unlock(&ocfs2_control_lock);
613 
614         return 0;
615 }
616 
617 static const struct file_operations ocfs2_control_fops = {
618         .open    = ocfs2_control_open,
619         .release = ocfs2_control_release,
620         .read    = ocfs2_control_read,
621         .write   = ocfs2_control_write,
622         .owner   = THIS_MODULE,
623         .llseek  = default_llseek,
624 };
625 
626 static struct miscdevice ocfs2_control_device = {
627         .minor          = MISC_DYNAMIC_MINOR,
628         .name           = "ocfs2_control",
629         .fops           = &ocfs2_control_fops,
630 };
631 
632 static int ocfs2_control_init(void)
633 {
634         int rc;
635 
636         atomic_set(&ocfs2_control_opened, 0);
637 
638         rc = misc_register(&ocfs2_control_device);
639         if (rc)
640                 printk(KERN_ERR
641                        "ocfs2: Unable to register ocfs2_control device "
642                        "(errno %d)\n",
643                        -rc);
644 
645         return rc;
646 }
647 
648 static void ocfs2_control_exit(void)
649 {
650         misc_deregister(&ocfs2_control_device);
651 }
652 
653 static void fsdlm_lock_ast_wrapper(void *astarg)
654 {
655         struct ocfs2_dlm_lksb *lksb = astarg;
656         int status = lksb->lksb_fsdlm.sb_status;
657 
658         /*
659          * For now we're punting on the issue of other non-standard errors
660          * where we can't tell if the unlock_ast or lock_ast should be called.
661          * The main "other error" that's possible is EINVAL which means the
662          * function was called with invalid args, which shouldn't be possible
663          * since the caller here is under our control.  Other non-standard
664          * errors probably fall into the same category, or otherwise are fatal
665          * which means we can't carry on anyway.
666          */
667 
668         if (status == -DLM_EUNLOCK || status == -DLM_ECANCEL)
669                 lksb->lksb_conn->cc_proto->lp_unlock_ast(lksb, 0);
670         else
671                 lksb->lksb_conn->cc_proto->lp_lock_ast(lksb);
672 }
673 
674 static void fsdlm_blocking_ast_wrapper(void *astarg, int level)
675 {
676         struct ocfs2_dlm_lksb *lksb = astarg;
677 
678         lksb->lksb_conn->cc_proto->lp_blocking_ast(lksb, level);
679 }
680 
681 static int user_dlm_lock(struct ocfs2_cluster_connection *conn,
682                          int mode,
683                          struct ocfs2_dlm_lksb *lksb,
684                          u32 flags,
685                          void *name,
686                          unsigned int namelen)
687 {
688         int ret;
689 
690         if (!lksb->lksb_fsdlm.sb_lvbptr)
691                 lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
692                                              sizeof(struct dlm_lksb);
693 
694         ret = dlm_lock(conn->cc_lockspace, mode, &lksb->lksb_fsdlm,
695                        flags|DLM_LKF_NODLCKWT, name, namelen, 0,
696                        fsdlm_lock_ast_wrapper, lksb,
697                        fsdlm_blocking_ast_wrapper);
698         return ret;
699 }
700 
701 static int user_dlm_unlock(struct ocfs2_cluster_connection *conn,
702                            struct ocfs2_dlm_lksb *lksb,
703                            u32 flags)
704 {
705         int ret;
706 
707         ret = dlm_unlock(conn->cc_lockspace, lksb->lksb_fsdlm.sb_lkid,
708                          flags, &lksb->lksb_fsdlm, lksb);
709         return ret;
710 }
711 
712 static int user_dlm_lock_status(struct ocfs2_dlm_lksb *lksb)
713 {
714         return lksb->lksb_fsdlm.sb_status;
715 }
716 
717 static int user_dlm_lvb_valid(struct ocfs2_dlm_lksb *lksb)
718 {
719         int invalid = lksb->lksb_fsdlm.sb_flags & DLM_SBF_VALNOTVALID;
720 
721         return !invalid;
722 }
723 
724 static void *user_dlm_lvb(struct ocfs2_dlm_lksb *lksb)
725 {
726         if (!lksb->lksb_fsdlm.sb_lvbptr)
727                 lksb->lksb_fsdlm.sb_lvbptr = (char *)lksb +
728                                              sizeof(struct dlm_lksb);
729         return (void *)(lksb->lksb_fsdlm.sb_lvbptr);
730 }
731 
732 static void user_dlm_dump_lksb(struct ocfs2_dlm_lksb *lksb)
733 {
734 }
735 
736 static int user_plock(struct ocfs2_cluster_connection *conn,
737                       u64 ino,
738                       struct file *file,
739                       int cmd,
740                       struct file_lock *fl)
741 {
742         /*
743          * This more or less just demuxes the plock request into any
744          * one of three dlm calls.
745          *
746          * Internally, fs/dlm will pass these to a misc device, which
747          * a userspace daemon will read and write to.
748          *
749          * For now, cancel requests (which happen internally only),
750          * are turned into unlocks. Most of this function taken from
751          * gfs2_lock.
752          */
753 
754         if (cmd == F_CANCELLK) {
755                 cmd = F_SETLK;
756                 fl->fl_type = F_UNLCK;
757         }
758 
759         if (IS_GETLK(cmd))
760                 return dlm_posix_get(conn->cc_lockspace, ino, file, fl);
761         else if (fl->fl_type == F_UNLCK)
762                 return dlm_posix_unlock(conn->cc_lockspace, ino, file, fl);
763         else
764                 return dlm_posix_lock(conn->cc_lockspace, ino, file, cmd, fl);
765 }
766 
767 /*
768  * Compare a requested locking protocol version against the current one.
769  *
770  * If the major numbers are different, they are incompatible.
771  * If the current minor is greater than the request, they are incompatible.
772  * If the current minor is less than or equal to the request, they are
773  * compatible, and the requester should run at the current minor version.
774  */
775 static int fs_protocol_compare(struct ocfs2_protocol_version *existing,
776                                struct ocfs2_protocol_version *request)
777 {
778         if (existing->pv_major != request->pv_major)
779                 return 1;
780 
781         if (existing->pv_minor > request->pv_minor)
782                 return 1;
783 
784         if (existing->pv_minor < request->pv_minor)
785                 request->pv_minor = existing->pv_minor;
786 
787         return 0;
788 }
789 
790 static void lvb_to_version(char *lvb, struct ocfs2_protocol_version *ver)
791 {
792         struct ocfs2_protocol_version *pv =
793                 (struct ocfs2_protocol_version *)lvb;
794         /*
795          * ocfs2_protocol_version has two u8 variables, so we don't
796          * need any endian conversion.
797          */
798         ver->pv_major = pv->pv_major;
799         ver->pv_minor = pv->pv_minor;
800 }
801 
802 static void version_to_lvb(struct ocfs2_protocol_version *ver, char *lvb)
803 {
804         struct ocfs2_protocol_version *pv =
805                 (struct ocfs2_protocol_version *)lvb;
806         /*
807          * ocfs2_protocol_version has two u8 variables, so we don't
808          * need any endian conversion.
809          */
810         pv->pv_major = ver->pv_major;
811         pv->pv_minor = ver->pv_minor;
812 }
813 
814 static void sync_wait_cb(void *arg)
815 {
816         struct ocfs2_cluster_connection *conn = arg;
817         struct ocfs2_live_connection *lc = conn->cc_private;
818         complete(&lc->oc_sync_wait);
819 }
820 
821 static int sync_unlock(struct ocfs2_cluster_connection *conn,
822                 struct dlm_lksb *lksb, char *name)
823 {
824         int error;
825         struct ocfs2_live_connection *lc = conn->cc_private;
826 
827         error = dlm_unlock(conn->cc_lockspace, lksb->sb_lkid, 0, lksb, conn);
828         if (error) {
829                 printk(KERN_ERR "%s lkid %x error %d\n",
830                                 name, lksb->sb_lkid, error);
831                 return error;
832         }
833 
834         wait_for_completion(&lc->oc_sync_wait);
835 
836         if (lksb->sb_status != -DLM_EUNLOCK) {
837                 printk(KERN_ERR "%s lkid %x status %d\n",
838                                 name, lksb->sb_lkid, lksb->sb_status);
839                 return -1;
840         }
841         return 0;
842 }
843 
844 static int sync_lock(struct ocfs2_cluster_connection *conn,
845                 int mode, uint32_t flags,
846                 struct dlm_lksb *lksb, char *name)
847 {
848         int error, status;
849         struct ocfs2_live_connection *lc = conn->cc_private;
850 
851         error = dlm_lock(conn->cc_lockspace, mode, lksb, flags,
852                         name, strlen(name),
853                         0, sync_wait_cb, conn, NULL);
854         if (error) {
855                 printk(KERN_ERR "%s lkid %x flags %x mode %d error %d\n",
856                                 name, lksb->sb_lkid, flags, mode, error);
857                 return error;
858         }
859 
860         wait_for_completion(&lc->oc_sync_wait);
861 
862         status = lksb->sb_status;
863 
864         if (status && status != -EAGAIN) {
865                 printk(KERN_ERR "%s lkid %x flags %x mode %d status %d\n",
866                                 name, lksb->sb_lkid, flags, mode, status);
867         }
868 
869         return status;
870 }
871 
872 
873 static int version_lock(struct ocfs2_cluster_connection *conn, int mode,
874                 int flags)
875 {
876         struct ocfs2_live_connection *lc = conn->cc_private;
877         return sync_lock(conn, mode, flags,
878                         &lc->oc_version_lksb, VERSION_LOCK);
879 }
880 
881 static int version_unlock(struct ocfs2_cluster_connection *conn)
882 {
883         struct ocfs2_live_connection *lc = conn->cc_private;
884         return sync_unlock(conn, &lc->oc_version_lksb, VERSION_LOCK);
885 }
886 
887 /* get_protocol_version()
888  *
889  * To exchange ocfs2 versioning, we use the LVB of the version dlm lock.
890  * The algorithm is:
891  * 1. Attempt to take the lock in EX mode (non-blocking).
892  * 2. If successful (which means it is the first mount), write the
893  *    version number and downconvert to PR lock.
894  * 3. If unsuccessful (returns -EAGAIN), read the version from the LVB after
895  *    taking the PR lock.
896  */
897 
898 static int get_protocol_version(struct ocfs2_cluster_connection *conn)
899 {
900         int ret;
901         struct ocfs2_live_connection *lc = conn->cc_private;
902         struct ocfs2_protocol_version pv;
903 
904         running_proto.pv_major =
905                 ocfs2_user_plugin.sp_max_proto.pv_major;
906         running_proto.pv_minor =
907                 ocfs2_user_plugin.sp_max_proto.pv_minor;
908 
909         lc->oc_version_lksb.sb_lvbptr = lc->oc_lvb;
910         ret = version_lock(conn, DLM_LOCK_EX,
911                         DLM_LKF_VALBLK|DLM_LKF_NOQUEUE);
912         if (!ret) {
913                 conn->cc_version.pv_major = running_proto.pv_major;
914                 conn->cc_version.pv_minor = running_proto.pv_minor;
915                 version_to_lvb(&running_proto, lc->oc_lvb);
916                 version_lock(conn, DLM_LOCK_PR, DLM_LKF_CONVERT|DLM_LKF_VALBLK);
917         } else if (ret == -EAGAIN) {
918                 ret = version_lock(conn, DLM_LOCK_PR, DLM_LKF_VALBLK);
919                 if (ret)
920                         goto out;
921                 lvb_to_version(lc->oc_lvb, &pv);
922 
923                 if ((pv.pv_major != running_proto.pv_major) ||
924                                 (pv.pv_minor > running_proto.pv_minor)) {
925                         ret = -EINVAL;
926                         goto out;
927                 }
928 
929                 conn->cc_version.pv_major = pv.pv_major;
930                 conn->cc_version.pv_minor = pv.pv_minor;
931         }
932 out:
933         return ret;
934 }
935 
936 static void user_recover_prep(void *arg)
937 {
938 }
939 
940 static void user_recover_slot(void *arg, struct dlm_slot *slot)
941 {
942         struct ocfs2_cluster_connection *conn = arg;
943         printk(KERN_INFO "ocfs2: Node %d/%d down. Initiating recovery.\n",
944                         slot->nodeid, slot->slot);
945         conn->cc_recovery_handler(slot->nodeid, conn->cc_recovery_data);
946 
947 }
948 
949 static void user_recover_done(void *arg, struct dlm_slot *slots,
950                 int num_slots, int our_slot,
951                 uint32_t generation)
952 {
953         struct ocfs2_cluster_connection *conn = arg;
954         struct ocfs2_live_connection *lc = conn->cc_private;
955         int i;
956 
957         for (i = 0; i < num_slots; i++)
958                 if (slots[i].slot == our_slot) {
959                         atomic_set(&lc->oc_this_node, slots[i].nodeid);
960                         break;
961                 }
962 
963         lc->oc_our_slot = our_slot;
964         wake_up(&lc->oc_wait);
965 }
966 
967 static const struct dlm_lockspace_ops ocfs2_ls_ops = {
968         .recover_prep = user_recover_prep,
969         .recover_slot = user_recover_slot,
970         .recover_done = user_recover_done,
971 };
972 
973 static int user_cluster_disconnect(struct ocfs2_cluster_connection *conn)
974 {
975         version_unlock(conn);
976         dlm_release_lockspace(conn->cc_lockspace, 2);
977         conn->cc_lockspace = NULL;
978         ocfs2_live_connection_drop(conn->cc_private);
979         conn->cc_private = NULL;
980         return 0;
981 }
982 
983 static int user_cluster_connect(struct ocfs2_cluster_connection *conn)
984 {
985         dlm_lockspace_t *fsdlm;
986         struct ocfs2_live_connection *lc;
987         int rc, ops_rv;
988 
989         BUG_ON(conn == NULL);
990 
991         lc = kzalloc(sizeof(struct ocfs2_live_connection), GFP_KERNEL);
992         if (!lc)
993                 return -ENOMEM;
994 
995         init_waitqueue_head(&lc->oc_wait);
996         init_completion(&lc->oc_sync_wait);
997         atomic_set(&lc->oc_this_node, 0);
998         conn->cc_private = lc;
999         lc->oc_type = NO_CONTROLD;
1000 
1001         rc = dlm_new_lockspace(conn->cc_name, conn->cc_cluster_name,
1002                                DLM_LSFL_FS | DLM_LSFL_NEWEXCL, DLM_LVB_LEN,
1003                                &ocfs2_ls_ops, conn, &ops_rv, &fsdlm);
1004         if (rc) {
1005                 if (rc == -EEXIST || rc == -EPROTO)
1006                         printk(KERN_ERR "ocfs2: Unable to create the "
1007                                 "lockspace %s (%d), because a ocfs2-tools "
1008                                 "program is running on this file system "
1009                                 "with the same name lockspace\n",
1010                                 conn->cc_name, rc);
1011                 goto out;
1012         }
1013 
1014         if (ops_rv == -EOPNOTSUPP) {
1015                 lc->oc_type = WITH_CONTROLD;
1016                 printk(KERN_NOTICE "ocfs2: You seem to be using an older "
1017                                 "version of dlm_controld and/or ocfs2-tools."
1018                                 " Please consider upgrading.\n");
1019         } else if (ops_rv) {
1020                 rc = ops_rv;
1021                 goto out;
1022         }
1023         conn->cc_lockspace = fsdlm;
1024 
1025         rc = ocfs2_live_connection_attach(conn, lc);
1026         if (rc)
1027                 goto out;
1028 
1029         if (lc->oc_type == NO_CONTROLD) {
1030                 rc = get_protocol_version(conn);
1031                 if (rc) {
1032                         printk(KERN_ERR "ocfs2: Could not determine"
1033                                         " locking version\n");
1034                         user_cluster_disconnect(conn);
1035                         goto out;
1036                 }
1037                 wait_event(lc->oc_wait, (atomic_read(&lc->oc_this_node) > 0));
1038         }
1039 
1040         /*
1041          * running_proto must have been set before we allowed any mounts
1042          * to proceed.
1043          */
1044         if (fs_protocol_compare(&running_proto, &conn->cc_version)) {
1045                 printk(KERN_ERR
1046                        "Unable to mount with fs locking protocol version "
1047                        "%u.%u because negotiated protocol is %u.%u\n",
1048                        conn->cc_version.pv_major, conn->cc_version.pv_minor,
1049                        running_proto.pv_major, running_proto.pv_minor);
1050                 rc = -EPROTO;
1051                 ocfs2_live_connection_drop(lc);
1052                 lc = NULL;
1053         }
1054 
1055 out:
1056         if (rc)
1057                 kfree(lc);
1058         return rc;
1059 }
1060 
1061 
1062 static int user_cluster_this_node(struct ocfs2_cluster_connection *conn,
1063                                   unsigned int *this_node)
1064 {
1065         int rc;
1066         struct ocfs2_live_connection *lc = conn->cc_private;
1067 
1068         if (lc->oc_type == WITH_CONTROLD)
1069                 rc = ocfs2_control_get_this_node();
1070         else if (lc->oc_type == NO_CONTROLD)
1071                 rc = atomic_read(&lc->oc_this_node);
1072         else
1073                 rc = -EINVAL;
1074 
1075         if (rc < 0)
1076                 return rc;
1077 
1078         *this_node = rc;
1079         return 0;
1080 }
1081 
1082 static struct ocfs2_stack_operations ocfs2_user_plugin_ops = {
1083         .connect        = user_cluster_connect,
1084         .disconnect     = user_cluster_disconnect,
1085         .this_node      = user_cluster_this_node,
1086         .dlm_lock       = user_dlm_lock,
1087         .dlm_unlock     = user_dlm_unlock,
1088         .lock_status    = user_dlm_lock_status,
1089         .lvb_valid      = user_dlm_lvb_valid,
1090         .lock_lvb       = user_dlm_lvb,
1091         .plock          = user_plock,
1092         .dump_lksb      = user_dlm_dump_lksb,
1093 };
1094 
1095 static struct ocfs2_stack_plugin ocfs2_user_plugin = {
1096         .sp_name        = "user",
1097         .sp_ops         = &ocfs2_user_plugin_ops,
1098         .sp_owner       = THIS_MODULE,
1099 };
1100 
1101 
1102 static int __init ocfs2_user_plugin_init(void)
1103 {
1104         int rc;
1105 
1106         rc = ocfs2_control_init();
1107         if (!rc) {
1108                 rc = ocfs2_stack_glue_register(&ocfs2_user_plugin);
1109                 if (rc)
1110                         ocfs2_control_exit();
1111         }
1112 
1113         return rc;
1114 }
1115 
1116 static void __exit ocfs2_user_plugin_exit(void)
1117 {
1118         ocfs2_stack_glue_unregister(&ocfs2_user_plugin);
1119         ocfs2_control_exit();
1120 }
1121 
1122 MODULE_AUTHOR("Oracle");
1123 MODULE_DESCRIPTION("ocfs2 driver for userspace cluster stacks");
1124 MODULE_LICENSE("GPL");
1125 module_init(ocfs2_user_plugin_init);
1126 module_exit(ocfs2_user_plugin_exit);
1127 

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp