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