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