Whamcloud - gitweb
LU-12616 obclass: fix MDS start/stop race
[fs/lustre-release.git] / lustre / mdc / mdc_changelog.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2017, Commissariat a l'Energie Atomique et aux Energies
24  *                     Alternatives.
25  *
26  * Copyright (c) 2017, Intel Corporation.
27  *
28  * Author: Henri Doreau <henri.doreau@cea.fr>
29  */
30
31 #define DEBUG_SUBSYSTEM S_MDC
32
33 #include <linux/init.h>
34 #include <linux/kthread.h>
35 #include <linux/poll.h>
36 #include <linux/miscdevice.h>
37
38 #include <lustre_log.h>
39 #include <uapi/linux/lustre/lustre_ioctl.h>
40
41 #include "mdc_internal.h"
42
43
44 /*
45  * -- Changelog delivery through character device --
46  */
47
48 /**
49  * Mutex to protect chlg_registered_devices below
50  */
51 static DEFINE_MUTEX(chlg_registered_dev_lock);
52
53 /**
54  * Global linked list of all registered devices (one per MDT).
55  */
56 static LIST_HEAD(chlg_registered_devices);
57
58
59 struct chlg_registered_dev {
60         /* Device name of the form "changelog-{MDTNAME}" */
61         char                    ced_name[32];
62         /* Misc device descriptor */
63         struct miscdevice       ced_misc;
64         /* OBDs referencing this device (multiple mount point) */
65         struct list_head        ced_obds;
66         /* Reference counter for proper deregistration */
67         struct kref             ced_refs;
68         /* Link within the global chlg_registered_devices */
69         struct list_head        ced_link;
70 };
71
72 struct chlg_reader_state {
73         /* Shortcut to the corresponding OBD device */
74         struct obd_device       *crs_obd;
75         /* Producer thread (if any) */
76         struct task_struct      *crs_prod_task;
77         /* An error occurred that prevents from reading further */
78         int                      crs_err;
79         /* EOF, no more records available */
80         bool                     crs_eof;
81         /* Desired start position */
82         __u64                    crs_start_offset;
83         /* Wait queue for the catalog processing thread */
84         wait_queue_head_t        crs_waitq_prod;
85         /* Wait queue for the record copy threads */
86         wait_queue_head_t        crs_waitq_cons;
87         /* Mutex protecting crs_rec_count and crs_rec_queue */
88         struct mutex             crs_lock;
89         /* Number of item in the list */
90         __u64                    crs_rec_count;
91         /* List of prefetched enqueued_record::enq_linkage_items */
92         struct list_head         crs_rec_queue;
93         unsigned int             crs_last_catidx;
94         unsigned int             crs_last_idx;
95         bool                     crs_poll;
96 };
97
98 struct chlg_rec_entry {
99         /* Link within the chlg_reader_state::crs_rec_queue list */
100         struct list_head        enq_linkage;
101         /* Data (enq_record) field length */
102         __u64                   enq_length;
103         /* Copy of a changelog record (see struct llog_changelog_rec) */
104         struct changelog_rec    enq_record[];
105 };
106
107 enum {
108         /* Number of records to prefetch locally. */
109         CDEV_CHLG_MAX_PREFETCH = 1024,
110 };
111
112 /**
113  * ChangeLog catalog processing callback invoked on each record.
114  * If the current record is eligible to userland delivery, push
115  * it into the crs_rec_queue where the consumer code will fetch it.
116  *
117  * @param[in]     env  (unused)
118  * @param[in]     llh  Client-side handle used to identify the llog
119  * @param[in]     hdr  Header of the current llog record
120  * @param[in,out] data chlg_reader_state passed from caller
121  *
122  * @return 0 or LLOG_PROC_* control code on success, negated error on failure.
123  */
124 static int chlg_read_cat_process_cb(const struct lu_env *env,
125                                     struct llog_handle *llh,
126                                     struct llog_rec_hdr *hdr, void *data)
127 {
128         struct llog_changelog_rec *rec;
129         struct chlg_reader_state *crs = data;
130         struct chlg_rec_entry *enq;
131         size_t len;
132         int rc;
133         ENTRY;
134
135         LASSERT(crs != NULL);
136         LASSERT(hdr != NULL);
137
138         rec = container_of(hdr, struct llog_changelog_rec, cr_hdr);
139
140         crs->crs_last_catidx = llh->lgh_hdr->llh_cat_idx;
141         crs->crs_last_idx = hdr->lrh_index;
142
143         if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
144                 rc = -EINVAL;
145                 CERROR("%s: not a changelog rec %x/%d in llog "DFID" rc = %d\n",
146                        crs->crs_obd->obd_name, rec->cr_hdr.lrh_type,
147                        rec->cr.cr_type,
148                        PFID(lu_object_fid(&llh->lgh_obj->do_lu)), rc);
149                 RETURN(rc);
150         }
151
152         /* Skip undesired records */
153         if (rec->cr.cr_index < crs->crs_start_offset)
154                 RETURN(0);
155
156         CDEBUG(D_HSM, "%llu %02d%-5s %llu 0x%x t="DFID" p="DFID" %.*s\n",
157                rec->cr.cr_index, rec->cr.cr_type,
158                changelog_type2str(rec->cr.cr_type), rec->cr.cr_time,
159                rec->cr.cr_flags & CLF_FLAGMASK,
160                PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid),
161                rec->cr.cr_namelen, changelog_rec_name(&rec->cr));
162
163         wait_event_interruptible(crs->crs_waitq_prod,
164                                  crs->crs_rec_count < CDEV_CHLG_MAX_PREFETCH ||
165                                  kthread_should_stop());
166
167         if (kthread_should_stop())
168                 RETURN(LLOG_PROC_BREAK);
169
170         len = changelog_rec_size(&rec->cr) + rec->cr.cr_namelen;
171         OBD_ALLOC(enq, sizeof(*enq) + len);
172         if (enq == NULL)
173                 RETURN(-ENOMEM);
174
175         INIT_LIST_HEAD(&enq->enq_linkage);
176         enq->enq_length = len;
177         memcpy(enq->enq_record, &rec->cr, len);
178
179         mutex_lock(&crs->crs_lock);
180         list_add_tail(&enq->enq_linkage, &crs->crs_rec_queue);
181         crs->crs_rec_count++;
182         mutex_unlock(&crs->crs_lock);
183
184         wake_up_all(&crs->crs_waitq_cons);
185
186         RETURN(0);
187 }
188
189 /**
190  * Remove record from the list it is attached to and free it.
191  */
192 static void enq_record_delete(struct chlg_rec_entry *rec)
193 {
194         list_del(&rec->enq_linkage);
195         OBD_FREE(rec, sizeof(*rec) + rec->enq_length);
196 }
197
198 /**
199  * Record prefetch thread entry point. Opens the changelog catalog and starts
200  * reading records.
201  *
202  * @param[in,out]  args  chlg_reader_state passed from caller.
203  * @return 0 on success, negated error code on failure.
204  */
205 static int chlg_load(void *args)
206 {
207         struct chlg_reader_state *crs = args;
208         struct obd_device *obd = crs->crs_obd;
209         struct llog_ctxt *ctx = NULL;
210         struct llog_handle *llh = NULL;
211         int rc;
212         ENTRY;
213
214         ctx = llog_get_context(obd, LLOG_CHANGELOG_REPL_CTXT);
215         if (ctx == NULL)
216                 GOTO(err_out, rc = -ENOENT);
217
218         crs->crs_last_catidx = -1;
219         crs->crs_last_idx = 0;
220
221 again:
222         rc = llog_open(NULL, ctx, &llh, NULL, CHANGELOG_CATALOG,
223                        LLOG_OPEN_EXISTS);
224         if (rc) {
225                 CERROR("%s: fail to open changelog catalog: rc = %d\n",
226                        obd->obd_name, rc);
227                 GOTO(err_out, rc);
228         }
229
230
231         rc = llog_init_handle(NULL, llh,
232                               LLOG_F_IS_CAT |
233                               LLOG_F_EXT_JOBID |
234                               LLOG_F_EXT_EXTRA_FLAGS |
235                               LLOG_F_EXT_X_UIDGID |
236                               LLOG_F_EXT_X_NID |
237                               LLOG_F_EXT_X_OMODE |
238                               LLOG_F_EXT_X_XATTR,
239                               NULL);
240         if (rc) {
241                 CERROR("%s: fail to init llog handle: rc = %d\n",
242                        obd->obd_name, rc);
243                 GOTO(err_out, rc);
244         }
245
246         rc = llog_cat_process(NULL, llh, chlg_read_cat_process_cb, crs,
247                                 crs->crs_last_catidx, crs->crs_last_idx);
248         if (rc < 0) {
249                 CERROR("%s: fail to process llog: rc = %d\n", obd->obd_name, rc);
250                 GOTO(err_out, rc);
251         }
252         if (!kthread_should_stop() && crs->crs_poll) {
253                 llog_cat_close(NULL, llh);
254                 schedule_timeout_interruptible(HZ);
255                 goto again;
256         }
257
258         crs->crs_eof = true;
259
260 err_out:
261         if (rc < 0)
262                 crs->crs_err = rc;
263
264         wake_up_all(&crs->crs_waitq_cons);
265
266         if (llh != NULL)
267                 llog_cat_close(NULL, llh);
268
269         if (ctx != NULL)
270                 llog_ctxt_put(ctx);
271
272         wait_event_interruptible(crs->crs_waitq_prod, kthread_should_stop());
273
274         RETURN(rc);
275 }
276
277 /**
278  * Read handler, dequeues records from the chlg_reader_state if any.
279  * No partial records are copied to userland so this function can return less
280  * data than required (short read).
281  *
282  * @param[in]   file   File pointer to the character device.
283  * @param[out]  buff   Userland buffer where to copy the records.
284  * @param[in]   count  Userland buffer size.
285  * @param[out]  ppos   File position, updated with the index number of the next
286  *                     record to read.
287  * @return number of copied bytes on success, negated error code on failure.
288  */
289 static ssize_t chlg_read(struct file *file, char __user *buff, size_t count,
290                          loff_t *ppos)
291 {
292         struct chlg_reader_state *crs = file->private_data;
293         struct chlg_rec_entry *rec;
294         struct chlg_rec_entry *tmp;
295         size_t written_total = 0;
296         ssize_t rc;
297         LIST_HEAD(consumed);
298         ENTRY;
299
300         if (file->f_flags & O_NONBLOCK && crs->crs_rec_count == 0) {
301                 if (crs->crs_err < 0)
302                         RETURN(crs->crs_err);
303                 else if (crs->crs_eof)
304                         RETURN(0);
305                 else
306                         RETURN(-EAGAIN);
307         }
308
309         rc = wait_event_interruptible(crs->crs_waitq_cons,
310                         crs->crs_rec_count > 0 || crs->crs_eof || crs->crs_err);
311
312         mutex_lock(&crs->crs_lock);
313         list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage) {
314                 if (written_total + rec->enq_length > count)
315                         break;
316
317                 if (copy_to_user(buff, rec->enq_record, rec->enq_length)) {
318                         rc = -EFAULT;
319                         break;
320                 }
321
322                 buff += rec->enq_length;
323                 written_total += rec->enq_length;
324
325                 crs->crs_rec_count--;
326                 list_move_tail(&rec->enq_linkage, &consumed);
327
328                 crs->crs_start_offset = rec->enq_record->cr_index + 1;
329         }
330         mutex_unlock(&crs->crs_lock);
331
332         if (written_total > 0) {
333                 rc = written_total;
334                 wake_up_all(&crs->crs_waitq_prod);
335         } else if (rc == 0) {
336                 rc = crs->crs_err;
337         }
338
339         list_for_each_entry_safe(rec, tmp, &consumed, enq_linkage)
340                 enq_record_delete(rec);
341
342         *ppos = crs->crs_start_offset;
343
344         RETURN(rc);
345 }
346
347 /**
348  * Jump to a given record index. Helper for chlg_llseek().
349  *
350  * @param[in,out]  crs     Internal reader state.
351  * @param[in]      offset  Desired offset (index record).
352  * @return 0 on success, negated error code on failure.
353  */
354 static int chlg_set_start_offset(struct chlg_reader_state *crs, __u64 offset)
355 {
356         struct chlg_rec_entry *rec;
357         struct chlg_rec_entry *tmp;
358
359         mutex_lock(&crs->crs_lock);
360         if (offset < crs->crs_start_offset) {
361                 mutex_unlock(&crs->crs_lock);
362                 return -ERANGE;
363         }
364
365         crs->crs_start_offset = offset;
366         list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage) {
367                 struct changelog_rec *cr = rec->enq_record;
368
369                 if (cr->cr_index >= crs->crs_start_offset)
370                         break;
371
372                 crs->crs_rec_count--;
373                 enq_record_delete(rec);
374         }
375
376         mutex_unlock(&crs->crs_lock);
377         wake_up_all(&crs->crs_waitq_prod);
378         return 0;
379 }
380
381 /**
382  * Move read pointer to a certain record index, encoded as an offset.
383  *
384  * @param[in,out] file   File pointer to the changelog character device
385  * @param[in]     off    Offset to skip, actually a record index, not byte count
386  * @param[in]     whence Relative/Absolute interpretation of the offset
387  * @return the resulting position on success or negated error code on failure.
388  */
389 static loff_t chlg_llseek(struct file *file, loff_t off, int whence)
390 {
391         struct chlg_reader_state *crs = file->private_data;
392         loff_t pos;
393         int rc;
394
395         switch (whence) {
396         case SEEK_SET:
397                 pos = off;
398                 break;
399         case SEEK_CUR:
400                 pos = file->f_pos + off;
401                 break;
402         case SEEK_END:
403         default:
404                 return -EINVAL;
405         }
406
407         /* We cannot go backward */
408         if (pos < file->f_pos)
409                 return -EINVAL;
410
411         rc = chlg_set_start_offset(crs, pos);
412         if (rc != 0)
413                 return rc;
414
415         file->f_pos = pos;
416         return pos;
417 }
418
419 /**
420  * Clear record range for a given changelog reader.
421  *
422  * @param[in]  crs     Current internal state.
423  * @param[in]  reader  Changelog reader ID (cl1, cl2...)
424  * @param[in]  record  Record index up which to clear
425  * @return 0 on success, negated error code on failure.
426  */
427 static int chlg_clear(struct chlg_reader_state *crs, __u32 reader, __u64 record)
428 {
429         struct obd_device *obd = crs->crs_obd;
430         struct changelog_setinfo cs  = {
431                 .cs_recno = record,
432                 .cs_id    = reader
433         };
434
435         return obd_set_info_async(NULL, obd->obd_self_export,
436                                   strlen(KEY_CHANGELOG_CLEAR),
437                                   KEY_CHANGELOG_CLEAR, sizeof(cs), &cs, NULL);
438 }
439
440 /** Maximum changelog control command size */
441 #define CHLG_CONTROL_CMD_MAX    64
442
443 /**
444  * Handle writes() into the changelog character device. Write() can be used
445  * to request special control operations.
446  *
447  * @param[in]  file  File pointer to the changelog character device
448  * @param[in]  buff  User supplied data (written data)
449  * @param[in]  count Number of written bytes
450  * @param[in]  off   (unused)
451  * @return number of written bytes on success, negated error code on failure.
452  */
453 static ssize_t chlg_write(struct file *file, const char __user *buff,
454                           size_t count, loff_t *off)
455 {
456         struct chlg_reader_state *crs = file->private_data;
457         char *kbuf;
458         __u64 record;
459         __u32 reader;
460         int rc = 0;
461         ENTRY;
462
463         if (count > CHLG_CONTROL_CMD_MAX)
464                 RETURN(-EINVAL);
465
466         OBD_ALLOC(kbuf, CHLG_CONTROL_CMD_MAX);
467         if (kbuf == NULL)
468                 RETURN(-ENOMEM);
469
470         if (copy_from_user(kbuf, buff, count))
471                 GOTO(out_kbuf, rc = -EFAULT);
472
473         kbuf[CHLG_CONTROL_CMD_MAX - 1] = '\0';
474
475         if (sscanf(kbuf, "clear:cl%u:%llu", &reader, &record) == 2)
476                 rc = chlg_clear(crs, reader, record);
477         else
478                 rc = -EINVAL;
479
480         EXIT;
481 out_kbuf:
482         OBD_FREE(kbuf, CHLG_CONTROL_CMD_MAX);
483         return rc < 0 ? rc : count;
484 }
485
486 /**
487  * Open handler, initialize internal CRS state and spawn prefetch thread if
488  * needed.
489  * @param[in]  inode  Inode struct for the open character device.
490  * @param[in]  file   Corresponding file pointer.
491  * @return 0 on success, negated error code on failure.
492  */
493 static int chlg_open(struct inode *inode, struct file *file)
494 {
495         struct chlg_reader_state *crs;
496         struct miscdevice *misc = file->private_data;
497         struct chlg_registered_dev *dev;
498         struct obd_device *obd;
499         struct task_struct *task;
500         int rc;
501         ENTRY;
502
503         dev = container_of(misc, struct chlg_registered_dev, ced_misc);
504         obd = list_first_entry(&dev->ced_obds,
505                                struct obd_device,
506                                u.cli.cl_chg_dev_linkage);
507
508         OBD_ALLOC_PTR(crs);
509         if (!crs)
510                 RETURN(-ENOMEM);
511
512         crs->crs_obd = obd;
513         crs->crs_err = false;
514         crs->crs_eof = false;
515
516         mutex_init(&crs->crs_lock);
517         INIT_LIST_HEAD(&crs->crs_rec_queue);
518         init_waitqueue_head(&crs->crs_waitq_prod);
519         init_waitqueue_head(&crs->crs_waitq_cons);
520
521         if (file->f_mode & FMODE_READ) {
522                 task = kthread_run(chlg_load, crs, "chlg_load_thread");
523                 if (IS_ERR(task)) {
524                         rc = PTR_ERR(task);
525                         CERROR("%s: cannot start changelog thread: rc = %d\n",
526                                obd->obd_name, rc);
527                         GOTO(err_crs, rc);
528                 }
529                 crs->crs_prod_task = task;
530         }
531
532         file->private_data = crs;
533         RETURN(0);
534
535 err_crs:
536         OBD_FREE_PTR(crs);
537         return rc;
538 }
539
540 /**
541  * Close handler, release resources.
542  *
543  * @param[in]  inode  Inode struct for the open character device.
544  * @param[in]  file   Corresponding file pointer.
545  * @return 0 on success, negated error code on failure.
546  */
547 static int chlg_release(struct inode *inode, struct file *file)
548 {
549         struct chlg_reader_state *crs = file->private_data;
550         struct chlg_rec_entry *rec;
551         struct chlg_rec_entry *tmp;
552         int rc = 0;
553
554         if (crs->crs_prod_task)
555                 rc = kthread_stop(crs->crs_prod_task);
556
557         list_for_each_entry_safe(rec, tmp, &crs->crs_rec_queue, enq_linkage)
558                 enq_record_delete(rec);
559
560         OBD_FREE_PTR(crs);
561
562         return rc;
563 }
564
565 /**
566  * Poll handler, indicates whether the device is readable (new records) and
567  * writable (always).
568  *
569  * @param[in]  file   Device file pointer.
570  * @param[in]  wait   (opaque)
571  * @return combination of the poll status flags.
572  */
573 static unsigned int chlg_poll(struct file *file, poll_table *wait)
574 {
575         struct chlg_reader_state *crs = file->private_data;
576         unsigned int mask = 0;
577
578         mutex_lock(&crs->crs_lock);
579         poll_wait(file, &crs->crs_waitq_cons, wait);
580         if (crs->crs_rec_count > 0)
581                 mask |= POLLIN | POLLRDNORM;
582         if (crs->crs_err)
583                 mask |= POLLERR;
584         if (crs->crs_eof)
585                 mask |= POLLHUP;
586         mutex_unlock(&crs->crs_lock);
587         return mask;
588 }
589
590 static long chlg_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
591 {
592         int rc;
593
594         struct chlg_reader_state *crs = file->private_data;
595         switch (cmd) {
596         case OBD_IOC_CHLG_POLL:
597                 crs->crs_poll = !!arg;
598                 rc = 0;
599                 break;
600         default:
601                 rc = -EINVAL;
602                 break;
603         }
604         return rc;
605 }
606
607 static const struct file_operations chlg_fops = {
608         .owner          = THIS_MODULE,
609         .llseek         = chlg_llseek,
610         .read           = chlg_read,
611         .write          = chlg_write,
612         .open           = chlg_open,
613         .release        = chlg_release,
614         .poll           = chlg_poll,
615         .unlocked_ioctl = chlg_ioctl,
616 };
617
618 /**
619  * This uses obd_name of the form: "testfs-MDT0000-mdc-ffff88006501600"
620  * and returns a name of the form: "changelog-testfs-MDT0000".
621  */
622 static void get_chlg_name(char *name, size_t name_len, struct obd_device *obd)
623 {
624         int i;
625
626         snprintf(name, name_len, "changelog-%s", obd->obd_name);
627
628         /* Find the 2nd '-' from the end and truncate on it */
629         for (i = 0; i < 2; i++) {
630                 char *p = strrchr(name, '-');
631
632                 if (p == NULL)
633                         return;
634                 *p = '\0';
635         }
636 }
637
638 /**
639  * Find a changelog character device by name.
640  * All devices registered during MDC setup are listed in a global list with
641  * their names attached.
642  */
643 static struct chlg_registered_dev *
644 chlg_registered_dev_find_by_name(const char *name)
645 {
646         struct chlg_registered_dev *dit;
647
648         LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
649         list_for_each_entry(dit, &chlg_registered_devices, ced_link)
650                 if (strcmp(name, dit->ced_name) == 0)
651                         return dit;
652         return NULL;
653 }
654
655 /**
656  * Find chlg_registered_dev structure for a given OBD device.
657  * This is bad O(n^2) but for each filesystem:
658  *   - N is # of MDTs times # of mount points
659  *   - this only runs at shutdown
660  */
661 static struct chlg_registered_dev *
662 chlg_registered_dev_find_by_obd(const struct obd_device *obd)
663 {
664         struct chlg_registered_dev *dit;
665         struct obd_device *oit;
666
667         LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
668         list_for_each_entry(dit, &chlg_registered_devices, ced_link)
669                 list_for_each_entry(oit, &dit->ced_obds,
670                                     u.cli.cl_chg_dev_linkage)
671                         if (oit == obd)
672                                 return dit;
673         return NULL;
674 }
675
676 /**
677  * Changelog character device initialization.
678  * Register a misc character device with a dynamic minor number, under a name
679  * of the form: 'changelog-fsname-MDTxxxx'. Reference this OBD device with it.
680  *
681  * @param[in] obd  This MDC obd_device.
682  * @return 0 on success, negated error code on failure.
683  */
684 int mdc_changelog_cdev_init(struct obd_device *obd)
685 {
686         struct chlg_registered_dev *exist;
687         struct chlg_registered_dev *entry;
688         int rc;
689         ENTRY;
690
691         OBD_ALLOC_PTR(entry);
692         if (entry == NULL)
693                 RETURN(-ENOMEM);
694
695         get_chlg_name(entry->ced_name, sizeof(entry->ced_name), obd);
696
697         entry->ced_misc.minor = MISC_DYNAMIC_MINOR;
698         entry->ced_misc.name  = entry->ced_name;
699         entry->ced_misc.fops  = &chlg_fops;
700
701         kref_init(&entry->ced_refs);
702         INIT_LIST_HEAD(&entry->ced_obds);
703         INIT_LIST_HEAD(&entry->ced_link);
704
705         mutex_lock(&chlg_registered_dev_lock);
706         exist = chlg_registered_dev_find_by_name(entry->ced_name);
707         if (exist != NULL) {
708                 kref_get(&exist->ced_refs);
709                 list_add_tail(&obd->u.cli.cl_chg_dev_linkage, &exist->ced_obds);
710                 GOTO(out_unlock, rc = 0);
711         }
712
713         list_add_tail(&obd->u.cli.cl_chg_dev_linkage, &entry->ced_obds);
714         list_add_tail(&entry->ced_link, &chlg_registered_devices);
715
716         /* Register new character device */
717         rc = misc_register(&entry->ced_misc);
718         if (rc != 0) {
719                 list_del_init(&obd->u.cli.cl_chg_dev_linkage);
720                 list_del(&entry->ced_link);
721                 GOTO(out_unlock, rc);
722         }
723
724         entry = NULL;   /* prevent it from being freed below */
725
726 out_unlock:
727         mutex_unlock(&chlg_registered_dev_lock);
728         if (entry)
729                 OBD_FREE_PTR(entry);
730         RETURN(rc);
731 }
732
733 /**
734  * Deregister a changelog character device whose refcount has reached zero.
735  */
736 static void chlg_dev_clear(struct kref *kref)
737 {
738         struct chlg_registered_dev *entry = container_of(kref,
739                                                       struct chlg_registered_dev,
740                                                       ced_refs);
741         ENTRY;
742
743         LASSERT(mutex_is_locked(&chlg_registered_dev_lock));
744         list_del(&entry->ced_link);
745         misc_deregister(&entry->ced_misc);
746         OBD_FREE_PTR(entry);
747         EXIT;
748 }
749
750 /**
751  * Release OBD, decrease reference count of the corresponding changelog device.
752  */
753 void mdc_changelog_cdev_finish(struct obd_device *obd)
754 {
755         struct chlg_registered_dev *dev;
756
757         ENTRY;
758         mutex_lock(&chlg_registered_dev_lock);
759         dev = chlg_registered_dev_find_by_obd(obd);
760         list_del_init(&obd->u.cli.cl_chg_dev_linkage);
761         kref_put(&dev->ced_refs, chlg_dev_clear);
762         mutex_unlock(&chlg_registered_dev_lock);
763         EXIT;
764 }