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