llog_lvfs.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 927 行 · 第 1/3 页

        if (left != 0 && left != reclen &&
            left < (reclen + LLOG_MIN_REC_SIZE)) {
                loghandle->lgh_last_idx++;
                rc = llog_lvfs_pad(obd, file, left, loghandle->lgh_last_idx);
                if (rc)
                        RETURN(rc);
                /* if it's the last idx in log file, then return -ENOSPC */
                if (loghandle->lgh_last_idx == LLOG_BITMAP_SIZE(llh) - 1)
                        RETURN(-ENOSPC);
        }

        loghandle->lgh_last_idx++;
        index = loghandle->lgh_last_idx;
        LASSERT(index < LLOG_BITMAP_SIZE(llh));
        rec->lrh_index = index;
        if (buf == NULL) {
                lrt = (struct llog_rec_tail *)
                        ((char *)rec + rec->lrh_len - sizeof(*lrt));
                lrt->lrt_len = rec->lrh_len;
                lrt->lrt_index = rec->lrh_index;
        }
        if (ext2_set_bit(index, llh->llh_bitmap)) {
                CERROR("argh, index %u already set in log bitmap?\n", index);
                LBUG(); /* should never happen */
        }
        llh->llh_count++;
        llh->llh_tail.lrt_index = index;

        rc = llog_lvfs_write_blob(obd, file, &llh->llh_hdr, NULL, 0);
        if (rc)
                RETURN(rc);

        rc = llog_lvfs_write_blob(obd, file, rec, buf, file->f_pos);
        if (rc)
                RETURN(rc);

        CDEBUG(D_RPCTRACE, "added record "LPX64": idx: %u, %u bytes\n",
               loghandle->lgh_id.lgl_oid, index, rec->lrh_len);
        if (rc == 0 && reccookie) {
                reccookie->lgc_lgl = loghandle->lgh_id;
                reccookie->lgc_index = index;
                if ((rec->lrh_type == MDS_UNLINK_REC) || 
                                (rec->lrh_type == MDS_SETATTR_REC))
                        reccookie->lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
                else if (rec->lrh_type == OST_SZ_REC)
                        reccookie->lgc_subsys = LLOG_SIZE_ORIG_CTXT;
                else if (rec->lrh_type == OST_RAID1_REC)
                        reccookie->lgc_subsys = LLOG_RD1_ORIG_CTXT;
                else
                        reccookie->lgc_subsys = -1;
                rc = 1;
        }
        if (rc == 0 && rec->lrh_type == LLOG_GEN_REC)
                rc = 1;

        RETURN(rc);
}

/* We can skip reading at least as many log blocks as the number of
* minimum sized log records we are skipping.  If it turns out
* that we are not far enough along the log (because the
* actual records are larger than minimum size) we just skip
* some more records. */

static void llog_skip_over(__u64 *off, int curr, int goal)
{
        if (goal <= curr)
                return;
        *off = (*off + (goal-curr-1) * LLOG_MIN_REC_SIZE) &
                ~(LLOG_CHUNK_SIZE - 1);
}


/* sets:
 *  - cur_offset to the furthest point read in the log file
 *  - cur_idx to the log index preceeding cur_offset
 * returns -EIO/-EINVAL on error
 */
static int llog_lvfs_next_block(struct llog_handle *loghandle, int *cur_idx,
                                int next_idx, __u64 *cur_offset, void *buf,
                                int len)
{
        int rc;
        ENTRY;

        if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
                RETURN(-EINVAL);

        CDEBUG(D_OTHER, "looking for log index %u (cur idx %u off "LPU64")\n",
               next_idx, *cur_idx, *cur_offset);

        while (*cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
                struct llog_rec_hdr *rec;
                struct llog_rec_tail *tail;
                loff_t ppos;

                llog_skip_over(cur_offset, *cur_idx, next_idx);

                ppos = *cur_offset;
                rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
                                        loghandle->lgh_file, buf, len,
                                        &ppos);
                if (rc) {
                        CERROR("Cant read llog block at log id "LPU64
                               "/%u offset "LPU64"\n",
                               loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen,
                               *cur_offset);
                        RETURN(rc);
                }

                /* put number of bytes read into rc to make code simpler */
                rc = ppos - *cur_offset;
                *cur_offset = ppos;
                
                if (rc < len) {
                        /* signal the end of the valid buffer to llog_process */
                        memset(buf + rc, 0, len - rc);
                }

                if (rc == 0) /* end of file, nothing to do */
                        RETURN(0);

                if (rc < sizeof(*tail)) {
                        CERROR("Invalid llog block at log id "LPU64"/%u offset "
                               LPU64"\n", loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen, *cur_offset);
                        RETURN(-EINVAL);
                }

                rec = buf;
                tail = (struct llog_rec_tail *)((char *)buf + rc -
                                                sizeof(struct llog_rec_tail));

                if (LLOG_REC_HDR_NEEDS_SWABBING(rec)) {
                        lustre_swab_llog_rec(rec, tail);
                }

                *cur_idx = tail->lrt_index;

                /* this shouldn't happen */
                if (tail->lrt_index == 0) {
                        CERROR("Invalid llog tail at log id "LPU64"/%u offset "
                               LPU64"\n", loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen, *cur_offset);
                        RETURN(-EINVAL);
                }
                if (tail->lrt_index < next_idx)
                        continue;

                /* sanity check that the start of the new buffer is no farther
                 * than the record that we wanted.  This shouldn't happen. */
                if (rec->lrh_index > next_idx) {
                        CERROR("missed desired record? %u > %u\n",
                               rec->lrh_index, next_idx);
                        RETURN(-ENOENT);
                }
                RETURN(0);
        }
        RETURN(-EIO);
}

static int llog_lvfs_prev_block(struct llog_handle *loghandle,
                                int prev_idx, void *buf, int len)
{
        __u64 cur_offset;
        int rc;
        ENTRY;

        if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
                RETURN(-EINVAL);

        CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);

        cur_offset = LLOG_CHUNK_SIZE;
        llog_skip_over(&cur_offset, 0, prev_idx);

        while (cur_offset < i_size_read(loghandle->lgh_file->f_dentry->d_inode)) {
                struct llog_rec_hdr *rec;
                struct llog_rec_tail *tail;
                loff_t ppos;

                ppos = cur_offset;

                rc = fsfilt_read_record(loghandle->lgh_ctxt->loc_exp->exp_obd,
                                        loghandle->lgh_file, buf, len,
                                        &ppos);
                if (rc) {
                        CERROR("Cant read llog block at log id "LPU64
                               "/%u offset "LPU64"\n",
                               loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen,
                               cur_offset);
                        RETURN(rc);
                }

                /* put number of bytes read into rc to make code simpler */
                rc = ppos - cur_offset;
                cur_offset = ppos;

                if (rc == 0) /* end of file, nothing to do */
                        RETURN(0);

                if (rc < sizeof(*tail)) {
                        CERROR("Invalid llog block at log id "LPU64"/%u offset "
                               LPU64"\n", loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen, cur_offset);
                        RETURN(-EINVAL);
                }

                tail = buf + rc - sizeof(struct llog_rec_tail);

                /* this shouldn't happen */
                if (tail->lrt_index == 0) {
                        CERROR("Invalid llog tail at log id "LPU64"/%u offset "
                               LPU64"\n", loghandle->lgh_id.lgl_oid,
                               loghandle->lgh_id.lgl_ogen, cur_offset);
                        RETURN(-EINVAL);
                }
                if (le32_to_cpu(tail->lrt_index) < prev_idx)
                        continue;

                /* sanity check that the start of the new buffer is no farther
                 * than the record that we wanted.  This shouldn't happen. */
                rec = buf;
                if (le32_to_cpu(rec->lrh_index) > prev_idx) {
                        CERROR("missed desired record? %u > %u\n",
                               le32_to_cpu(rec->lrh_index), prev_idx);
                        RETURN(-ENOENT);
                }
                RETURN(0);
        }
        RETURN(-EIO);
}

static struct file *llog_filp_open(char *dir, char *name, int flags, int mode)
{
        char *logname;
        struct file *filp;
        int len;

        OBD_ALLOC(logname, PATH_MAX);
        if (logname == NULL)
                return ERR_PTR(-ENOMEM);

        len = snprintf(logname, PATH_MAX, "%s/%s", dir, name);
        if (len >= PATH_MAX - 1) {
                filp = ERR_PTR(-ENAMETOOLONG);
        } else {
                filp = l_filp_open(logname, flags, mode);
                if (IS_ERR(filp))
                        CERROR("logfile creation %s: %ld\n", logname,
                               PTR_ERR(filp));
        }
        OBD_FREE(logname, PATH_MAX);
        return filp;
}

/* This is a callback from the llog_* functions.
 * Assumes caller has already pushed us into the kernel context. */
static int llog_lvfs_create(struct llog_ctxt *ctxt, struct llog_handle **res,
                            struct llog_logid *logid, char *name)
{
        struct llog_handle *handle;
        struct obd_device *obd;
        struct l_dentry *dchild = NULL;
        struct obdo *oa = NULL;
        int rc = 0, cleanup_phase = 1;
        int open_flags = O_RDWR | O_CREAT | O_LARGEFILE;
        ENTRY;

        handle = llog_alloc_handle();
        if (handle == NULL)
                RETURN(-ENOMEM);
        *res = handle;

        LASSERT(ctxt);
        LASSERT(ctxt->loc_exp);
        obd = ctxt->loc_exp->exp_obd;

        if (logid != NULL) {
                dchild = obd_lvfs_fid2dentry(ctxt->loc_exp, logid->lgl_oid,
                                             logid->lgl_ogen, logid->lgl_ogr);

                if (IS_ERR(dchild)) {
                        rc = PTR_ERR(dchild);
                        CERROR("error looking up logfile "LPX64":0x%x: rc %d\n",
                               logid->lgl_oid, logid->lgl_ogen, rc);
                        GOTO(cleanup, rc);
                }

                cleanup_phase = 2;
                if (dchild->d_inode == NULL) {
                        rc = -ENOENT;
                        CERROR("nonexistent log file "LPX64":"LPX64": rc %d\n",
                               logid->lgl_oid, logid->lgl_ogr, rc);
                        GOTO(cleanup, rc);
                }

                handle->lgh_file = l_dentry_open(&obd->obd_lvfs_ctxt, dchild,
                                                    O_RDWR | O_LARGEFILE);
                if (IS_ERR(handle->lgh_file)) {
                        rc = PTR_ERR(handle->lgh_file);
                        CERROR("error opening logfile "LPX64"0x%x: rc %d\n",
                               logid->lgl_oid, logid->lgl_ogen, rc);
                        GOTO(cleanup, rc);
                }

                /* assign the value of lgh_id for handle directly */