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 }