>Reviewed-by: Yan, Zheng <zheng.z....@intel.com>
Thanks very much!
Jianpeng Ma
>
>On 09/12/2013 01:54 PM, majianpeng wrote:
>> For writev/pwritev sync-operatoin, ceph only do the first iov.
>> It don't think other iovs.Now implement this.
>> I divided the write-sync-operation into two functions.One for
>> direct-write,other for none-direct-sync-write.This is because for
>> none-direct-sync-write we can merge iovs to one.But for direct-write,
>> we can't merge iovs.
>>
>> V4:
>> reconstruct the code by Yan, Zheng
>> V2:
>> -using struct iov_iter replace clone iovs in ceph_sync_write.
>>
>> Signed-off-by: Jianpeng Ma <majianp...@gmail.com>
>> Reviewed-by: Yan, Zheng <zheng.z....@intel.com>
>> ---
>> fs/ceph/file.c | 273
>> ++++++++++++++++++++++++++++++++++++++++-----------------
>> 1 file changed, 193 insertions(+), 80 deletions(-)
>>
>> diff --git a/fs/ceph/file.c b/fs/ceph/file.c
>> index 3de8982..5cf034e 100644
>> --- a/fs/ceph/file.c
>> +++ b/fs/ceph/file.c
>> @@ -489,83 +489,79 @@ static void ceph_sync_write_unsafe(struct
>> ceph_osd_request *req, bool unsafe)
>> }
>> }
>>
>> +
>> /*
>> - * Synchronous write, straight from __user pointer or user pages (if
>> - * O_DIRECT).
>> + * Synchronous write, straight from __user pointer or user pages.
>> *
>> * If write spans object boundary, just do multiple writes. (For a
>> * correct atomic write, we should e.g. take write locks on all
>> * objects, rollback on failure, etc.)
>> */
>> -static ssize_t ceph_sync_write(struct file *file, const char __user *data,
>> - size_t left, loff_t pos, loff_t *ppos)
>> +static ssize_t
>> +ceph_sync_direct_write(struct kiocb *iocb, const struct iovec *iov,
>> + unsigned long nr_segs, size_t count)
>> {
>> + struct file *file = iocb->ki_filp;
>> struct inode *inode = file_inode(file);
>> struct ceph_inode_info *ci = ceph_inode(inode);
>> struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
>> struct ceph_snap_context *snapc;
>> struct ceph_vino vino;
>> struct ceph_osd_request *req;
>> - int num_ops = 1;
>> struct page **pages;
>> int num_pages;
>> - u64 len;
>> int written = 0;
>> int flags;
>> int check_caps = 0;
>> - int page_align, io_align;
>> - unsigned long buf_align;
>> + int page_align;
>> int ret;
>> struct timespec mtime = CURRENT_TIME;
>> - bool own_pages = false;
>> + loff_t pos = iocb->ki_pos;
>> + struct iov_iter i;
>>
>> if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
>> return -EROFS;
>>
>> - dout("sync_write on file %p %lld~%u %s\n", file, pos,
>> - (unsigned)left, (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
>> + dout("sync_direct_write on file %p %lld~%u\n", file, pos,
>> + (unsigned)count);
>>
>> - ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + left);
>> + ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
>> if (ret < 0)
>> return ret;
>>
>> ret = invalidate_inode_pages2_range(inode->i_mapping,
>> pos >> PAGE_CACHE_SHIFT,
>> - (pos + left) >> PAGE_CACHE_SHIFT);
>> + (pos + count) >> PAGE_CACHE_SHIFT);
>> if (ret < 0)
>> dout("invalidate_inode_pages2_range returned %d\n", ret);
>>
>> flags = CEPH_OSD_FLAG_ORDERSNAP |
>> CEPH_OSD_FLAG_ONDISK |
>> CEPH_OSD_FLAG_WRITE;
>> - if ((file->f_flags & (O_SYNC|O_DIRECT)) == 0)
>> - flags |= CEPH_OSD_FLAG_ACK;
>> - else
>> - num_ops++; /* Also include a 'startsync' command. */
>>
>> - /*
>> - * we may need to do multiple writes here if we span an object
>> - * boundary. this isn't atomic, unfortunately. :(
>> - */
>> -more:
>> - io_align = pos & ~PAGE_MASK;
>> - buf_align = (unsigned long)data & ~PAGE_MASK;
>> - len = left;
>> -
>> - snapc = ci->i_snap_realm->cached_context;
>> - vino = ceph_vino(inode);
>> - req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
>> - vino, pos, &len, num_ops,
>> - CEPH_OSD_OP_WRITE, flags, snapc,
>> - ci->i_truncate_seq, ci->i_truncate_size,
>> - false);
>> - if (IS_ERR(req))
>> - return PTR_ERR(req);
>> + iov_iter_init(&i, iov, nr_segs, count, 0);
>> +
>> + while (iov_iter_count(&i) > 0) {
>> + void __user *data = i.iov->iov_base + i.iov_offset;
>> + u64 len = i.iov->iov_len - i.iov_offset;
>> +
>> + page_align = (unsigned long)data & ~PAGE_MASK;
>> +
>> + snapc = ci->i_snap_realm->cached_context;
>> + vino = ceph_vino(inode);
>> + req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
>> + vino, pos, &len,
>> + 2,/*include a 'startsync' command*/
>> + CEPH_OSD_OP_WRITE, flags, snapc,
>> + ci->i_truncate_seq,
>> + ci->i_truncate_size,
>> + false);
>> + if (IS_ERR(req)) {
>> + ret = PTR_ERR(req);
>> + goto out;
>> + }
>>
>> - /* write from beginning of first page, regardless of io alignment */
>> - page_align = file->f_flags & O_DIRECT ? buf_align : io_align;
>> - num_pages = calc_pages_for(page_align, len);
>> - if (file->f_flags & O_DIRECT) {
>> + num_pages = calc_pages_for(page_align, len);
>> pages = ceph_get_direct_page_vector(data, num_pages, false);
>> if (IS_ERR(pages)) {
>> ret = PTR_ERR(pages);
>> @@ -577,60 +573,175 @@ more:
>> * may block.
>> */
>> truncate_inode_pages_range(inode->i_mapping, pos,
>> - (pos+len) | (PAGE_CACHE_SIZE-1));
>> - } else {
>> + (pos+len) | (PAGE_CACHE_SIZE-1));
>> + osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
>> + false, false);
>> +
>> + /* BUG_ON(vino.snap != CEPH_NOSNAP); */
>> + ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
>> +
>> + ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
>> + if (!ret)
>> + ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
>> +
>> + ceph_put_page_vector(pages, num_pages, false);
>> +
>> +out:
>> + ceph_osdc_put_request(req);
>> + if (ret == 0) {
>> + pos += len;
>> + written += len;
>> + iov_iter_advance(&i, (size_t)len);
>> +
>> + if (pos > i_size_read(inode)) {
>> + check_caps = ceph_inode_set_size(inode, pos);
>> + if (check_caps)
>> + ceph_check_caps(ceph_inode(inode),
>> + CHECK_CAPS_AUTHONLY,
>> + NULL);
>> + }
>> + } else
>> + break;
>> + }
>> +
>> + if (ret != -EOLDSNAPC && written > 0) {
>> + iocb->ki_pos = pos;
>> + ret = written;
>> + }
>> + return ret;
>> +}
>> +
>> +
>> +/*
>> + * Synchronous write, straight from __user pointer or user pages.
>> + *
>> + * If write spans object boundary, just do multiple writes. (For a
>> + * correct atomic write, we should e.g. take write locks on all
>> + * objects, rollback on failure, etc.)
>> + */
>> +static ssize_t ceph_sync_write(struct kiocb *iocb, const struct iovec *iov,
>> + unsigned long nr_segs, size_t count)
>> +{
>> + struct file *file = iocb->ki_filp;
>> + struct inode *inode = file_inode(file);
>> + struct ceph_inode_info *ci = ceph_inode(inode);
>> + struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
>> + struct ceph_snap_context *snapc;
>> + struct ceph_vino vino;
>> + struct ceph_osd_request *req;
>> + struct page **pages;
>> + u64 len;
>> + int num_pages;
>> + int written = 0;
>> + int flags;
>> + int check_caps = 0;
>> + int ret;
>> + struct timespec mtime = CURRENT_TIME;
>> + loff_t pos = iocb->ki_pos;
>> + struct iov_iter i;
>> +
>> + if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
>> + return -EROFS;
>> +
>> + dout("sync_write on file %p %lld~%u\n", file, pos, (unsigned)count);
>> +
>> + ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
>> + if (ret < 0)
>> + return ret;
>> +
>> + ret = invalidate_inode_pages2_range(inode->i_mapping,
>> + pos >> PAGE_CACHE_SHIFT,
>> + (pos + count) >> PAGE_CACHE_SHIFT);
>> + if (ret < 0)
>> + dout("invalidate_inode_pages2_range returned %d\n", ret);
>> +
>> + flags = CEPH_OSD_FLAG_ORDERSNAP |
>> + CEPH_OSD_FLAG_ONDISK |
>> + CEPH_OSD_FLAG_WRITE |
>> + CEPH_OSD_FLAG_ACK;
>> +
>> + iov_iter_init(&i, iov, nr_segs, count, 0);
>> +
>> + while ((len = iov_iter_count(&i)) > 0) {
>> + size_t left;
>> + int n;
>> +
>> + snapc = ci->i_snap_realm->cached_context;
>> + vino = ceph_vino(inode);
>> + req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
>> + vino, pos, &len, 1,
>> + CEPH_OSD_OP_WRITE, flags, snapc,
>> + ci->i_truncate_seq,
>> + ci->i_truncate_size,
>> + false);
>> + if (IS_ERR(req)) {
>> + ret = PTR_ERR(req);
>> + goto out;
>> + }
>> +
>> + /*
>> + * write from beginning of first page,
>> + * regardless of io alignment
>> + */
>> + num_pages = (len + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
>> +
>> pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
>> if (IS_ERR(pages)) {
>> ret = PTR_ERR(pages);
>> goto out;
>> }
>> - ret = ceph_copy_user_to_page_vector(pages, data, pos, len);
>> +
>> + left = len;
>> + for (n = 0; n < num_pages; n++) {
>> + size_t plen = min(left, PAGE_SIZE);
>> + ret = iov_iter_copy_from_user(pages[n], &i, 0, plen);
>> + if (ret != plen) {
>> + ret = -EFAULT;
>> + break;
>> + }
>> + left -= ret;
>> + iov_iter_advance(&i, ret);
>> + }
>> +
>> if (ret < 0) {
>> ceph_release_page_vector(pages, num_pages);
>> goto out;
>> }
>>
>> - if ((file->f_flags & O_SYNC) == 0) {
>> - /* get a second commit callback */
>> - req->r_unsafe_callback = ceph_sync_write_unsafe;
>> - req->r_inode = inode;
>> - own_pages = true;
>> - }
>> - }
>> - osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
>> - false, own_pages);
>> + /* get a second commit callback */
>> + req->r_unsafe_callback = ceph_sync_write_unsafe;
>> + req->r_inode = inode;
>>
>> - /* BUG_ON(vino.snap != CEPH_NOSNAP); */
>> - ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
>> + osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
>> + false, true);
>>
>> - ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
>> - if (!ret)
>> - ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
>> + /* BUG_ON(vino.snap != CEPH_NOSNAP); */
>> + ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
>>
>> - if (file->f_flags & O_DIRECT)
>> - ceph_put_page_vector(pages, num_pages, false);
>> - else if (file->f_flags & O_SYNC)
>> - ceph_release_page_vector(pages, num_pages);
>> + ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
>> + if (!ret)
>> + ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
>>
>> out:
>> - ceph_osdc_put_request(req);
>> - if (ret == 0) {
>> - pos += len;
>> - written += len;
>> - left -= len;
>> - data += len;
>> - if (left)
>> - goto more;
>> + ceph_osdc_put_request(req);
>> + if (ret == 0) {
>> + pos += len;
>> + written += len;
>> +
>> + if (pos > i_size_read(inode)) {
>> + check_caps = ceph_inode_set_size(inode, pos);
>> + if (check_caps)
>> + ceph_check_caps(ceph_inode(inode),
>> + CHECK_CAPS_AUTHONLY,
>> + NULL);
>> + }
>> + } else
>> + break;
>> + }
>>
>> + if (ret != -EOLDSNAPC && written > 0) {
>> ret = written;
>> - *ppos = pos;
>> - if (pos > i_size_read(inode))
>> - check_caps = ceph_inode_set_size(inode, pos);
>> - if (check_caps)
>> - ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY,
>> - NULL);
>> - } else if (ret != -EOLDSNAPC && written > 0) {
>> - ret = written;
>> + iocb->ki_pos = pos;
>> }
>> return ret;
>> }
>> @@ -772,11 +883,13 @@ retry_snap:
>> inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
>>
>> if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
>> - (iocb->ki_filp->f_flags & O_DIRECT) ||
>> - (fi->flags & CEPH_F_SYNC)) {
>> + (file->f_flags & O_DIRECT) || (fi->flags & CEPH_F_SYNC)) {
>> mutex_unlock(&inode->i_mutex);
>> - written = ceph_sync_write(file, iov->iov_base, count,
>> - pos, &iocb->ki_pos);
>> + if (file->f_flags & O_DIRECT)
>> + written = ceph_sync_direct_write(iocb, iov,
>> + nr_segs, count);
>> + else
>> + written = ceph_sync_write(iocb, iov, nr_segs, count);
>> if (written == -EOLDSNAPC) {
>> dout("aio_write %p %llx.%llx %llu~%u"
>> "got EOLDSNAPC, retrying\n",
>>
>
