file.c 45 KB
Newer Older
1
#include <linux/ceph/ceph_debug.h>
Sage Weil's avatar
Sage Weil committed
2

3
#include <linux/module.h>
Sage Weil's avatar
Sage Weil committed
4
#include <linux/sched.h>
5
#include <linux/slab.h>
Sage Weil's avatar
Sage Weil committed
6
#include <linux/file.h>
Sage Weil's avatar
Sage Weil committed
7
#include <linux/mount.h>
Sage Weil's avatar
Sage Weil committed
8 9
#include <linux/namei.h>
#include <linux/writeback.h>
Li Wang's avatar
Li Wang committed
10
#include <linux/falloc.h>
Sage Weil's avatar
Sage Weil committed
11 12 13

#include "super.h"
#include "mds_client.h"
14
#include "cache.h"
Sage Weil's avatar
Sage Weil committed
15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

/*
 * Ceph file operations
 *
 * Implement basic open/close functionality, and implement
 * read/write.
 *
 * We implement three modes of file I/O:
 *  - buffered uses the generic_file_aio_{read,write} helpers
 *
 *  - synchronous is used when there is multi-client read/write
 *    sharing, avoids the page cache, and synchronously waits for an
 *    ack from the OSD.
 *
 *  - direct io takes the variant of the sync path that references
 *    user pages directly.
 *
 * fsync() flushes and waits on dirty pages, but just queues metadata
 * for writeback: since the MDS can recover size and mtime there is no
 * need to wait for MDS acknowledgement.
 */

37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104
/*
 * Calculate the length sum of direct io vectors that can
 * be combined into one page vector.
 */
static size_t dio_get_pagev_size(const struct iov_iter *it)
{
    const struct iovec *iov = it->iov;
    const struct iovec *iovend = iov + it->nr_segs;
    size_t size;

    size = iov->iov_len - it->iov_offset;
    /*
     * An iov can be page vectored when both the current tail
     * and the next base are page aligned.
     */
    while (PAGE_ALIGNED((iov->iov_base + iov->iov_len)) &&
           (++iov < iovend && PAGE_ALIGNED((iov->iov_base)))) {
        size += iov->iov_len;
    }
    dout("dio_get_pagevlen len = %zu\n", size);
    return size;
}

/*
 * Allocate a page vector based on (@it, @nbytes).
 * The return value is the tuple describing a page vector,
 * that is (@pages, @page_align, @num_pages).
 */
static struct page **
dio_get_pages_alloc(const struct iov_iter *it, size_t nbytes,
		    size_t *page_align, int *num_pages)
{
	struct iov_iter tmp_it = *it;
	size_t align;
	struct page **pages;
	int ret = 0, idx, npages;

	align = (unsigned long)(it->iov->iov_base + it->iov_offset) &
		(PAGE_SIZE - 1);
	npages = calc_pages_for(align, nbytes);
	pages = kmalloc(sizeof(*pages) * npages, GFP_KERNEL);
	if (!pages) {
		pages = vmalloc(sizeof(*pages) * npages);
		if (!pages)
			return ERR_PTR(-ENOMEM);
	}

	for (idx = 0; idx < npages; ) {
		size_t start;
		ret = iov_iter_get_pages(&tmp_it, pages + idx, nbytes,
					 npages - idx, &start);
		if (ret < 0)
			goto fail;

		iov_iter_advance(&tmp_it, ret);
		nbytes -= ret;
		idx += (ret + start + PAGE_SIZE - 1) / PAGE_SIZE;
	}

	BUG_ON(nbytes != 0);
	*num_pages = npages;
	*page_align = align;
	dout("dio_get_pages_alloc: got %d pages align %zu\n", npages, align);
	return pages;
fail:
	ceph_put_page_vector(pages, idx, false);
	return ERR_PTR(ret);
}
Sage Weil's avatar
Sage Weil committed
105 106 107 108 109 110 111 112

/*
 * Prepare an open request.  Preallocate ceph_cap to avoid an
 * inopportune ENOMEM later.
 */
static struct ceph_mds_request *
prepare_open_request(struct super_block *sb, int flags, int create_mode)
{
113 114
	struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
	struct ceph_mds_client *mdsc = fsc->mdsc;
Sage Weil's avatar
Sage Weil committed
115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142
	struct ceph_mds_request *req;
	int want_auth = USE_ANY_MDS;
	int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN;

	if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
		want_auth = USE_AUTH_MDS;

	req = ceph_mdsc_create_request(mdsc, op, want_auth);
	if (IS_ERR(req))
		goto out;
	req->r_fmode = ceph_flags_to_mode(flags);
	req->r_args.open.flags = cpu_to_le32(flags);
	req->r_args.open.mode = cpu_to_le32(create_mode);
out:
	return req;
}

/*
 * initialize private struct file data.
 * if we fail, clean up by dropping fmode reference on the ceph_inode
 */
static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
{
	struct ceph_file_info *cf;
	int ret = 0;

	switch (inode->i_mode & S_IFMT) {
	case S_IFREG:
143 144
		ceph_fscache_register_inode_cookie(inode);
		ceph_fscache_file_set_cookie(inode, file);
Sage Weil's avatar
Sage Weil committed
145 146 147
	case S_IFDIR:
		dout("init_file %p %p 0%o (regular)\n", inode, file,
		     inode->i_mode);
Geliang Tang's avatar
Geliang Tang committed
148
		cf = kmem_cache_zalloc(ceph_file_cachep, GFP_KERNEL);
Sage Weil's avatar
Sage Weil committed
149 150 151 152 153 154
		if (cf == NULL) {
			ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
			return -ENOMEM;
		}
		cf->fmode = fmode;
		cf->next_offset = 2;
Yan, Zheng's avatar
Yan, Zheng committed
155
		cf->readdir_cache_idx = -1;
Sage Weil's avatar
Sage Weil committed
156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181
		file->private_data = cf;
		BUG_ON(inode->i_fop->release != ceph_release);
		break;

	case S_IFLNK:
		dout("init_file %p %p 0%o (symlink)\n", inode, file,
		     inode->i_mode);
		ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
		break;

	default:
		dout("init_file %p %p 0%o (special)\n", inode, file,
		     inode->i_mode);
		/*
		 * we need to drop the open ref now, since we don't
		 * have .release set to ceph_release.
		 */
		ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
		BUG_ON(inode->i_fop->release == ceph_release);

		/* call the proper open fop */
		ret = inode->i_fop->open(inode, file);
	}
	return ret;
}

182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
/*
 * try renew caps after session gets killed.
 */
int ceph_renew_caps(struct inode *inode)
{
	struct ceph_mds_client *mdsc = ceph_sb_to_client(inode->i_sb)->mdsc;
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_mds_request *req;
	int err, flags, wanted;

	spin_lock(&ci->i_ceph_lock);
	wanted = __ceph_caps_file_wanted(ci);
	if (__ceph_is_any_real_caps(ci) &&
	    (!(wanted & CEPH_CAP_ANY_WR) == 0 || ci->i_auth_cap)) {
		int issued = __ceph_caps_issued(ci, NULL);
		spin_unlock(&ci->i_ceph_lock);
		dout("renew caps %p want %s issued %s updating mds_wanted\n",
		     inode, ceph_cap_string(wanted), ceph_cap_string(issued));
		ceph_check_caps(ci, 0, NULL);
		return 0;
	}
	spin_unlock(&ci->i_ceph_lock);

	flags = 0;
	if ((wanted & CEPH_CAP_FILE_RD) && (wanted & CEPH_CAP_FILE_WR))
		flags = O_RDWR;
	else if (wanted & CEPH_CAP_FILE_RD)
		flags = O_RDONLY;
	else if (wanted & CEPH_CAP_FILE_WR)
		flags = O_WRONLY;
#ifdef O_LAZY
	if (wanted & CEPH_CAP_FILE_LAZYIO)
		flags |= O_LAZY;
#endif

	req = prepare_open_request(inode->i_sb, flags, 0);
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out;
	}

	req->r_inode = inode;
	ihold(inode);
	req->r_num_caps = 1;
	req->r_fmode = -1;

	err = ceph_mdsc_do_request(mdsc, NULL, req);
	ceph_mdsc_put_request(req);
out:
	dout("renew caps %p open result=%d\n", inode, err);
	return err < 0 ? err : 0;
}

Sage Weil's avatar
Sage Weil committed
235 236 237 238 239 240 241 242 243
/*
 * If we already have the requisite capabilities, we can satisfy
 * the open request locally (no need to request new caps from the
 * MDS).  We do, however, need to inform the MDS (asynchronously)
 * if our wanted caps set expands.
 */
int ceph_open(struct inode *inode, struct file *file)
{
	struct ceph_inode_info *ci = ceph_inode(inode);
244 245
	struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
	struct ceph_mds_client *mdsc = fsc->mdsc;
Sage Weil's avatar
Sage Weil committed
246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271
	struct ceph_mds_request *req;
	struct ceph_file_info *cf = file->private_data;
	int err;
	int flags, fmode, wanted;

	if (cf) {
		dout("open file %p is already opened\n", file);
		return 0;
	}

	/* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
	flags = file->f_flags & ~(O_CREAT|O_EXCL);
	if (S_ISDIR(inode->i_mode))
		flags = O_DIRECTORY;  /* mds likes to know */

	dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode,
	     ceph_vinop(inode), file, flags, file->f_flags);
	fmode = ceph_flags_to_mode(flags);
	wanted = ceph_caps_for_mode(fmode);

	/* snapped files are read-only */
	if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
		return -EROFS;

	/* trivially open snapdir */
	if (ceph_snap(inode) == CEPH_SNAPDIR) {
272
		spin_lock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
273
		__ceph_get_fmode(ci, fmode);
274
		spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
275 276 277 278
		return ceph_init_file(inode, file, fmode);
	}

	/*
279 280
	 * No need to block if we have caps on the auth MDS (for
	 * write) or any MDS (for read).  Update wanted set
Sage Weil's avatar
Sage Weil committed
281 282
	 * asynchronously.
	 */
283
	spin_lock(&ci->i_ceph_lock);
284 285
	if (__ceph_is_any_real_caps(ci) &&
	    (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
286
		int mds_wanted = __ceph_caps_mds_wanted(ci, true);
Sage Weil's avatar
Sage Weil committed
287 288 289 290 291 292
		int issued = __ceph_caps_issued(ci, NULL);

		dout("open %p fmode %d want %s issued %s using existing\n",
		     inode, fmode, ceph_cap_string(wanted),
		     ceph_cap_string(issued));
		__ceph_get_fmode(ci, fmode);
293
		spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
294 295 296 297 298 299 300 301 302 303 304

		/* adjust wanted? */
		if ((issued & wanted) != wanted &&
		    (mds_wanted & wanted) != wanted &&
		    ceph_snap(inode) != CEPH_SNAPDIR)
			ceph_check_caps(ci, 0, NULL);

		return ceph_init_file(inode, file, fmode);
	} else if (ceph_snap(inode) != CEPH_NOSNAP &&
		   (ci->i_snap_caps & wanted) == wanted) {
		__ceph_get_fmode(ci, fmode);
305
		spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
306 307
		return ceph_init_file(inode, file, fmode);
	}
308

309
	spin_unlock(&ci->i_ceph_lock);
Sage Weil's avatar
Sage Weil committed
310 311 312 313 314 315 316

	dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
	req = prepare_open_request(inode->i_sb, flags, 0);
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out;
	}
317 318
	req->r_inode = inode;
	ihold(inode);
319

Sage Weil's avatar
Sage Weil committed
320
	req->r_num_caps = 1;
321
	err = ceph_mdsc_do_request(mdsc, NULL, req);
Sage Weil's avatar
Sage Weil committed
322 323 324 325 326 327 328 329 330 331
	if (!err)
		err = ceph_init_file(inode, file, req->r_fmode);
	ceph_mdsc_put_request(req);
	dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
out:
	return err;
}


/*
Sage Weil's avatar
Sage Weil committed
332 333
 * Do a lookup + open with a single request.  If we get a non-existent
 * file or symlink, return 1 so the VFS can retry.
Sage Weil's avatar
Sage Weil committed
334
 */
Sage Weil's avatar
Sage Weil committed
335
int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
Al Viro's avatar
Al Viro committed
336
		     struct file *file, unsigned flags, umode_t mode,
Al Viro's avatar
Al Viro committed
337
		     int *opened)
Sage Weil's avatar
Sage Weil committed
338
{
339 340
	struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb);
	struct ceph_mds_client *mdsc = fsc->mdsc;
Sage Weil's avatar
Sage Weil committed
341
	struct ceph_mds_request *req;
Sage Weil's avatar
Sage Weil committed
342
	struct dentry *dn;
343
	struct ceph_acls_info acls = {};
344
       int mask;
Sage Weil's avatar
Sage Weil committed
345 346
	int err;

Al Viro's avatar
Al Viro committed
347 348
	dout("atomic_open %p dentry %p '%pd' %s flags %d mode 0%o\n",
	     dir, dentry, dentry,
Sage Weil's avatar
Sage Weil committed
349 350 351 352 353
	     d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);

	if (dentry->d_name.len > NAME_MAX)
		return -ENAMETOOLONG;

354 355 356 357 358 359
	if (flags & O_CREAT) {
		err = ceph_pre_init_acls(dir, &mode, &acls);
		if (err < 0)
			return err;
	}

Sage Weil's avatar
Sage Weil committed
360 361
	/* do the open */
	req = prepare_open_request(dir->i_sb, flags, mode);
362 363 364 365
	if (IS_ERR(req)) {
		err = PTR_ERR(req);
		goto out_acl;
	}
Sage Weil's avatar
Sage Weil committed
366 367 368 369 370
	req->r_dentry = dget(dentry);
	req->r_num_caps = 2;
	if (flags & O_CREAT) {
		req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
		req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
371 372 373 374
		if (acls.pagelist) {
			req->r_pagelist = acls.pagelist;
			acls.pagelist = NULL;
		}
Sage Weil's avatar
Sage Weil committed
375
	}
376 377 378 379 380 381

       mask = CEPH_STAT_CAP_INODE | CEPH_CAP_AUTH_SHARED;
       if (ceph_security_xattr_wanted(dir))
               mask |= CEPH_CAP_XATTR_SHARED;
       req->r_args.open.mask = cpu_to_le32(mask);

382 383
	req->r_parent = dir;
	set_bit(CEPH_MDS_R_PARENT_LOCKED, &req->r_req_flags);
384 385 386
	err = ceph_mdsc_do_request(mdsc,
				   (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
				   req);
Yan, Zheng's avatar
Yan, Zheng committed
387
	err = ceph_handle_snapdir(req, dentry, err);
388
	if (err)
389
		goto out_req;
390

391
	if ((flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
Sage Weil's avatar
Sage Weil committed
392
		err = ceph_handle_notrace_create(dir, dentry);
393

394
	if (d_in_lookup(dentry)) {
Sage Weil's avatar
Sage Weil committed
395 396 397 398 399 400 401 402
		dn = ceph_finish_lookup(req, dentry, err);
		if (IS_ERR(dn))
			err = PTR_ERR(dn);
	} else {
		/* we were given a hashed negative dentry */
		dn = NULL;
	}
	if (err)
403
		goto out_req;
404
	if (dn || d_really_is_negative(dentry) || d_is_symlink(dentry)) {
Sage Weil's avatar
Sage Weil committed
405 406 407 408 409
		/* make vfs retry on splice, ENOENT, or symlink */
		dout("atomic_open finish_no_open on dn %p\n", dn);
		err = finish_no_open(file, dn);
	} else {
		dout("atomic_open finish_open on dn %p\n", dn);
410
		if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
411
			ceph_init_inode_acls(d_inode(dentry), &acls);
412 413
			*opened |= FILE_CREATED;
		}
Sage Weil's avatar
Sage Weil committed
414 415
		err = finish_open(file, dentry, ceph_open, opened);
	}
416
out_req:
417 418
	if (!req->r_err && req->r_target_inode)
		ceph_put_fmode(ceph_inode(req->r_target_inode), req->r_fmode);
Sage Weil's avatar
Sage Weil committed
419
	ceph_mdsc_put_request(req);
420 421
out_acl:
	ceph_release_acls_info(&acls);
Sage Weil's avatar
Sage Weil committed
422
	dout("atomic_open result=%d\n", err);
Al Viro's avatar
Al Viro committed
423
	return err;
Sage Weil's avatar
Sage Weil committed
424 425 426 427 428 429 430 431 432 433 434 435 436 437
}

int ceph_release(struct inode *inode, struct file *file)
{
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_file_info *cf = file->private_data;

	dout("release inode %p file %p\n", inode, file);
	ceph_put_fmode(ci, cf->fmode);
	if (cf->last_readdir)
		ceph_mdsc_put_request(cf->last_readdir);
	kfree(cf->last_name);
	kfree(cf->dir_info);
	kmem_cache_free(ceph_file_cachep, cf);
438 439

	/* wake up anyone waiting for caps on this inode */
440
	wake_up_all(&ci->i_cap_wq);
Sage Weil's avatar
Sage Weil committed
441 442 443
	return 0;
}

Yan, Zheng's avatar
Yan, Zheng committed
444
enum {
Yan, Zheng's avatar
Yan, Zheng committed
445 446 447
	HAVE_RETRIED = 1,
	CHECK_EOF =    2,
	READ_INLINE =  3,
Yan, Zheng's avatar
Yan, Zheng committed
448 449
};

Sage Weil's avatar
Sage Weil committed
450 451 452 453 454 455 456 457
/*
 * Read a range of bytes striped over one or more objects.  Iterate over
 * objects we stripe over.  (That's not atomic, but good enough for now.)
 *
 * If we get a short result from the OSD, check against i_size; we need to
 * only return a short read to the caller if we hit EOF.
 */
static int striped_read(struct inode *inode,
458
			u64 pos, u64 len,
459
			struct page **pages, int num_pages,
460
			int page_align, int *checkeof)
Sage Weil's avatar
Sage Weil committed
461
{
462
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
Sage Weil's avatar
Sage Weil committed
463
	struct ceph_inode_info *ci = ceph_inode(inode);
464
	u64 this_len;
465
	loff_t i_size;
466 467
	int page_idx;
	int ret, read = 0;
Sage Weil's avatar
Sage Weil committed
468 469 470 471 472 473
	bool hit_stripe, was_short;

	/*
	 * we may need to do multiple reads.  not atomic, unfortunately.
	 */
more:
474 475
	this_len = len;
	page_idx = (page_align + read) >> PAGE_SHIFT;
476
	ret = ceph_osdc_readpages(&fsc->client->osdc, ceph_vino(inode),
Sage Weil's avatar
Sage Weil committed
477
				  &ci->i_layout, pos, &this_len,
478 479 480
				  ci->i_truncate_seq, ci->i_truncate_size,
				  pages + page_idx, num_pages - page_idx,
				  ((page_align + read) & ~PAGE_MASK));
Sage Weil's avatar
Sage Weil committed
481 482
	if (ret == -ENOENT)
		ret = 0;
483
	hit_stripe = this_len < len;
484
	was_short = ret >= 0 && ret < this_len;
485
	dout("striped_read %llu~%llu (read %u) got %d%s%s\n", pos, len, read,
Sage Weil's avatar
Sage Weil committed
486 487
	     ret, hit_stripe ? " HITSTRIPE" : "", was_short ? " SHORT" : "");

488
	i_size = i_size_read(inode);
489
	if (ret >= 0) {
490 491
		if (was_short && (pos + ret < i_size)) {
			int zlen = min(this_len - ret, i_size - pos - ret);
492
			int zoff = page_align + read + ret;
493
			dout(" zero gap %llu to %llu\n",
494
			     pos + ret, pos + ret + zlen);
495 496
			ceph_zero_page_vector_range(zoff, zlen, pages);
			ret += zlen;
Sage Weil's avatar
Sage Weil committed
497
		}
498

499
		read += ret;
Sage Weil's avatar
Sage Weil committed
500
		pos += ret;
501
		len -= ret;
Sage Weil's avatar
Sage Weil committed
502

503
		/* hit stripe and need continue*/
504
		if (len && hit_stripe && pos < i_size)
Sage Weil's avatar
Sage Weil committed
505 506 507
			goto more;
	}

508
	if (read > 0) {
509
		ret = read;
510
		/* did we bounce off eof? */
511
		if (pos + len > i_size)
Yan, Zheng's avatar
Yan, Zheng committed
512
			*checkeof = CHECK_EOF;
Sage Weil's avatar
Sage Weil committed
513 514 515 516 517 518 519 520 521 522 523 524
	}

	dout("striped_read returns %d\n", ret);
	return ret;
}

/*
 * Completely synchronous read and write methods.  Direct from __user
 * buffer to osd, or directly to user pages (if O_DIRECT).
 *
 * If the read spans object boundary, just do multiple reads.
 */
525 526
static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
			      int *checkeof)
Sage Weil's avatar
Sage Weil committed
527
{
528
	struct file *file = iocb->ki_filp;
Al Viro's avatar
Al Viro committed
529
	struct inode *inode = file_inode(file);
Sage Weil's avatar
Sage Weil committed
530
	struct page **pages;
531
	u64 off = iocb->ki_pos;
532 533 534
	int num_pages;
	ssize_t ret;
	size_t len = iov_iter_count(to);
Sage Weil's avatar
Sage Weil committed
535

536 537
	dout("sync_read on file %p %llu~%u %s\n", file, off,
	     (unsigned)len,
Sage Weil's avatar
Sage Weil committed
538
	     (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
539 540 541

	if (!len)
		return 0;
542 543 544 545 546 547
	/*
	 * flush any page cache pages in this range.  this
	 * will make concurrent normal and sync io slow,
	 * but it will at least behave sensibly when they are
	 * in sequence.
	 */
548 549
	ret = filemap_write_and_wait_range(inode->i_mapping, off,
						off + len);
550
	if (ret < 0)
551
		return ret;
552

553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592
	if (unlikely(to->type & ITER_PIPE)) {
		size_t page_off;
		ret = iov_iter_get_pages_alloc(to, &pages, len,
					       &page_off);
		if (ret <= 0)
			return -ENOMEM;
		num_pages = DIV_ROUND_UP(ret + page_off, PAGE_SIZE);

		ret = striped_read(inode, off, ret, pages, num_pages,
				   page_off, checkeof);
		if (ret > 0) {
			iov_iter_advance(to, ret);
			off += ret;
		} else {
			iov_iter_advance(to, 0);
		}
		ceph_put_page_vector(pages, num_pages, false);
	} else {
		num_pages = calc_pages_for(off, len);
		pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
		if (IS_ERR(pages))
			return PTR_ERR(pages);

		ret = striped_read(inode, off, len, pages, num_pages,
				   (off & ~PAGE_MASK), checkeof);
		if (ret > 0) {
			int l, k = 0;
			size_t left = ret;

			while (left) {
				size_t page_off = off & ~PAGE_MASK;
				size_t copy = min_t(size_t, left,
						    PAGE_SIZE - page_off);
				l = copy_page_to_iter(pages[k++], page_off,
						      copy, to);
				off += l;
				left -= l;
				if (l < copy)
					break;
			}
593
		}
594
		ceph_release_page_vector(pages, num_pages);
595
	}
Sage Weil's avatar
Sage Weil committed
596

597 598 599 600
	if (off > iocb->ki_pos) {
		ret = off - iocb->ki_pos;
		iocb->ki_pos = off;
	}
Sage Weil's avatar
Sage Weil committed
601

602
	dout("sync_read result %zd\n", ret);
Sage Weil's avatar
Sage Weil committed
603 604 605
	return ret;
}

Yan, Zheng's avatar
Yan, Zheng committed
606 607 608 609 610 611 612 613
struct ceph_aio_request {
	struct kiocb *iocb;
	size_t total_len;
	int write;
	int error;
	struct list_head osd_reqs;
	unsigned num_reqs;
	atomic_t pending_reqs;
614
	struct timespec mtime;
Yan, Zheng's avatar
Yan, Zheng committed
615 616 617
	struct ceph_cap_flush *prealloc_cf;
};

618 619 620 621 622 623 624
struct ceph_aio_work {
	struct work_struct work;
	struct ceph_osd_request *req;
};

static void ceph_aio_retry_work(struct work_struct *work);

Yan, Zheng's avatar
Yan, Zheng committed
625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667
static void ceph_aio_complete(struct inode *inode,
			      struct ceph_aio_request *aio_req)
{
	struct ceph_inode_info *ci = ceph_inode(inode);
	int ret;

	if (!atomic_dec_and_test(&aio_req->pending_reqs))
		return;

	ret = aio_req->error;
	if (!ret)
		ret = aio_req->total_len;

	dout("ceph_aio_complete %p rc %d\n", inode, ret);

	if (ret >= 0 && aio_req->write) {
		int dirty;

		loff_t endoff = aio_req->iocb->ki_pos + aio_req->total_len;
		if (endoff > i_size_read(inode)) {
			if (ceph_inode_set_size(inode, endoff))
				ceph_check_caps(ci, CHECK_CAPS_AUTHONLY, NULL);
		}

		spin_lock(&ci->i_ceph_lock);
		ci->i_inline_version = CEPH_INLINE_NONE;
		dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR,
					       &aio_req->prealloc_cf);
		spin_unlock(&ci->i_ceph_lock);
		if (dirty)
			__mark_inode_dirty(inode, dirty);

	}

	ceph_put_cap_refs(ci, (aio_req->write ? CEPH_CAP_FILE_WR :
						CEPH_CAP_FILE_RD));

	aio_req->iocb->ki_complete(aio_req->iocb, ret, 0);

	ceph_free_cap_flush(aio_req->prealloc_cf);
	kfree(aio_req);
}

668
static void ceph_aio_complete_req(struct ceph_osd_request *req)
Yan, Zheng's avatar
Yan, Zheng committed
669 670 671 672 673 674 675 676 677 678 679 680
{
	int rc = req->r_result;
	struct inode *inode = req->r_inode;
	struct ceph_aio_request *aio_req = req->r_priv;
	struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
	int num_pages = calc_pages_for((u64)osd_data->alignment,
				       osd_data->length);

	dout("ceph_aio_complete_req %p rc %d bytes %llu\n",
	     inode, rc, osd_data->length);

	if (rc == -EOLDSNAPC) {
681 682 683 684 685 686 687 688 689 690 691 692 693
		struct ceph_aio_work *aio_work;
		BUG_ON(!aio_req->write);

		aio_work = kmalloc(sizeof(*aio_work), GFP_NOFS);
		if (aio_work) {
			INIT_WORK(&aio_work->work, ceph_aio_retry_work);
			aio_work->req = req;
			queue_work(ceph_inode_to_client(inode)->wb_wq,
				   &aio_work->work);
			return;
		}
		rc = -ENOMEM;
	} else if (!aio_req->write) {
Yan, Zheng's avatar
Yan, Zheng committed
694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718
		if (rc == -ENOENT)
			rc = 0;
		if (rc >= 0 && osd_data->length > rc) {
			int zoff = osd_data->alignment + rc;
			int zlen = osd_data->length - rc;
			/*
			 * If read is satisfied by single OSD request,
			 * it can pass EOF. Otherwise read is within
			 * i_size.
			 */
			if (aio_req->num_reqs == 1) {
				loff_t i_size = i_size_read(inode);
				loff_t endoff = aio_req->iocb->ki_pos + rc;
				if (endoff < i_size)
					zlen = min_t(size_t, zlen,
						     i_size - endoff);
				aio_req->total_len = rc + zlen;
			}

			if (zlen > 0)
				ceph_zero_page_vector_range(zoff, zlen,
							    osd_data->pages);
		}
	}

719
	ceph_put_page_vector(osd_data->pages, num_pages, !aio_req->write);
Yan, Zheng's avatar
Yan, Zheng committed
720 721 722 723 724 725 726 727 728
	ceph_osdc_put_request(req);

	if (rc < 0)
		cmpxchg(&aio_req->error, 0, rc);

	ceph_aio_complete(inode, aio_req);
	return;
}

729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755
static void ceph_aio_retry_work(struct work_struct *work)
{
	struct ceph_aio_work *aio_work =
		container_of(work, struct ceph_aio_work, work);
	struct ceph_osd_request *orig_req = aio_work->req;
	struct ceph_aio_request *aio_req = orig_req->r_priv;
	struct inode *inode = orig_req->r_inode;
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct ceph_snap_context *snapc;
	struct ceph_osd_request *req;
	int ret;

	spin_lock(&ci->i_ceph_lock);
	if (__ceph_have_pending_cap_snap(ci)) {
		struct ceph_cap_snap *capsnap =
			list_last_entry(&ci->i_cap_snaps,
					struct ceph_cap_snap,
					ci_item);
		snapc = ceph_get_snap_context(capsnap->context);
	} else {
		BUG_ON(!ci->i_head_snapc);
		snapc = ceph_get_snap_context(ci->i_head_snapc);
	}
	spin_unlock(&ci->i_ceph_lock);

	req = ceph_osdc_alloc_request(orig_req->r_osdc, snapc, 2,
			false, GFP_NOFS);
756 757
	if (!req) {
		ret = -ENOMEM;
758 759 760 761 762 763 764
		req = orig_req;
		goto out;
	}

	req->r_flags =	CEPH_OSD_FLAG_ORDERSNAP |
			CEPH_OSD_FLAG_ONDISK |
			CEPH_OSD_FLAG_WRITE;
765
	ceph_oloc_copy(&req->r_base_oloc, &orig_req->r_base_oloc);
766
	ceph_oid_copy(&req->r_base_oid, &orig_req->r_base_oid);
767

768 769 770 771 772 773
	ret = ceph_osdc_alloc_messages(req, GFP_NOFS);
	if (ret) {
		ceph_osdc_put_request(req);
		req = orig_req;
		goto out;
	}
774 775 776 777

	req->r_ops[0] = orig_req->r_ops[0];
	osd_req_op_init(req, 1, CEPH_OSD_OP_STARTSYNC, 0);

778 779
	req->r_mtime = aio_req->mtime;
	req->r_data_offset = req->r_ops[0].extent.offset;
780 781 782 783 784 785 786 787 788 789 790

	ceph_osdc_put_request(orig_req);

	req->r_callback = ceph_aio_complete_req;
	req->r_inode = inode;
	req->r_priv = aio_req;

	ret = ceph_osdc_start_request(req->r_osdc, req, false);
out:
	if (ret < 0) {
		req->r_result = ret;
791
		ceph_aio_complete_req(req);
792 793
	}

794
	ceph_put_snap_context(snapc);
795 796 797
	kfree(aio_work);
}

Sage Weil's avatar
Sage Weil committed
798
/*
799 800 801 802 803 804 805 806 807 808
 * Write commit request unsafe callback, called to tell us when a
 * request is unsafe (that is, in flight--has been handed to the
 * messenger to send to its target osd).  It is called again when
 * we've received a response message indicating the request is
 * "safe" (its CEPH_OSD_FLAG_ONDISK flag is set), or when a request
 * is completed early (and unsuccessfully) due to a timeout or
 * interrupt.
 *
 * This is used if we requested both an ACK and ONDISK commit reply
 * from the OSD.
Sage Weil's avatar
Sage Weil committed
809
 */
810
static void ceph_sync_write_unsafe(struct ceph_osd_request *req, bool unsafe)
Sage Weil's avatar
Sage Weil committed
811 812 813
{
	struct ceph_inode_info *ci = ceph_inode(req->r_inode);

814 815 816 817 818 819 820 821
	dout("%s %p tid %llu %ssafe\n", __func__, req, req->r_tid,
		unsafe ? "un" : "");
	if (unsafe) {
		ceph_get_cap_refs(ci, CEPH_CAP_FILE_WR);
		spin_lock(&ci->i_unsafe_lock);
		list_add_tail(&req->r_unsafe_item,
			      &ci->i_unsafe_writes);
		spin_unlock(&ci->i_unsafe_lock);
822 823

		complete_all(&req->r_completion);
824 825 826 827 828 829
	} else {
		spin_lock(&ci->i_unsafe_lock);
		list_del_init(&req->r_unsafe_item);
		spin_unlock(&ci->i_unsafe_lock);
		ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR);
	}
Sage Weil's avatar
Sage Weil committed
830 831
}

832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863
/*
 * Wait on any unsafe replies for the given inode.  First wait on the
 * newest request, and make that the upper bound.  Then, if there are
 * more requests, keep waiting on the oldest as long as it is still older
 * than the original request.
 */
void ceph_sync_write_wait(struct inode *inode)
{
	struct ceph_inode_info *ci = ceph_inode(inode);
	struct list_head *head = &ci->i_unsafe_writes;
	struct ceph_osd_request *req;
	u64 last_tid;

	if (!S_ISREG(inode->i_mode))
		return;

	spin_lock(&ci->i_unsafe_lock);
	if (list_empty(head))
		goto out;

	/* set upper bound as _last_ entry in chain */

	req = list_last_entry(head, struct ceph_osd_request,
			      r_unsafe_item);
	last_tid = req->r_tid;

	do {
		ceph_osdc_get_request(req);
		spin_unlock(&ci->i_unsafe_lock);

		dout("sync_write_wait on tid %llu (until %llu)\n",
		     req->r_tid, last_tid);
864
		wait_for_completion(&req->r_done_completion);
865 866 867 868 869 870 871 872 873 874 875 876 877 878 879
		ceph_osdc_put_request(req);

		spin_lock(&ci->i_unsafe_lock);
		/*
		 * from here on look at first entry in chain, since we
		 * only want to wait for anything older than last_tid
		 */
		if (list_empty(head))
			break;
		req = list_first_entry(head, struct ceph_osd_request,
				       r_unsafe_item);
	} while (req->r_tid < last_tid);
out:
	spin_unlock(&ci->i_unsafe_lock);
}
880 881

static ssize_t
Yan, Zheng's avatar
Yan, Zheng committed
882 883 884
ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
		       struct ceph_snap_context *snapc,
		       struct ceph_cap_flush **pcf)
Sage Weil's avatar
Sage Weil committed
885
{
886
	struct file *file = iocb->ki_filp;
Al Viro's avatar
Al Viro committed
887
	struct inode *inode = file_inode(file);
Sage Weil's avatar
Sage Weil committed
888
	struct ceph_inode_info *ci = ceph_inode(inode);
889
	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
890
	struct ceph_vino vino;
Sage Weil's avatar
Sage Weil committed
891 892
	struct ceph_osd_request *req;
	struct page **pages;
Yan, Zheng's avatar
Yan, Zheng committed
893 894
	struct ceph_aio_request *aio_req = NULL;
	int num_pages = 0;
Sage Weil's avatar
Sage Weil committed
895 896
	int flags;
	int ret;
897
	struct timespec mtime = current_time(inode);
Yan, Zheng's avatar
Yan, Zheng committed
898 899 900
	size_t count = iov_iter_count(iter);
	loff_t pos = iocb->ki_pos;
	bool write = iov_iter_rw(iter) == WRITE;
Sage Weil's avatar
Sage Weil committed
901

Yan, Zheng's avatar
Yan, Zheng committed
902
	if (write && ceph_snap(file_inode(file)) != CEPH_NOSNAP)
Sage Weil's avatar
Sage Weil committed
903 904
		return -EROFS;

Yan, Zheng's avatar
Yan, Zheng committed
905 906
	dout("sync_direct_read_write (%s) on file %p %lld~%u\n",
	     (write ? "write" : "read"), file, pos, (unsigned)count);
Sage Weil's avatar
Sage Weil committed
907

908
	ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
909 910 911
	if (ret < 0)
		return ret;

Yan, Zheng's avatar
Yan, Zheng committed
912
	if (write) {
913
		int ret2 = invalidate_inode_pages2_range(inode->i_mapping,
914 915
					pos >> PAGE_SHIFT,
					(pos + count) >> PAGE_SHIFT);
916
		if (ret2 < 0)
917
			dout("invalidate_inode_pages2_range returned %d\n", ret2);
918

Yan, Zheng's avatar
Yan, Zheng committed
919 920 921 922 923 924
		flags = CEPH_OSD_FLAG_ORDERSNAP |
			CEPH_OSD_FLAG_ONDISK |
			CEPH_OSD_FLAG_WRITE;
	} else {
		flags = CEPH_OSD_FLAG_READ;
	}
Sage Weil's avatar
Sage Weil committed
925

Yan, Zheng's avatar
Yan, Zheng committed
926 927 928 929
	while (iov_iter_count(iter) > 0) {
		u64 size = dio_get_pagev_size(iter);
		size_t start = 0;
		ssize_t len;
930 931 932

		vino = ceph_vino(inode);
		req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
Yan, Zheng's avatar
Yan, Zheng committed
933 934 935 936 937 938
					    vino, pos, &size, 0,
					    /*include a 'startsync' command*/
					    write ? 2 : 1,
					    write ? CEPH_OSD_OP_WRITE :
						    CEPH_OSD_OP_READ,
					    flags, snapc,
939 940 941 942 943
					    ci->i_truncate_seq,
					    ci->i_truncate_size,
					    false);
		if (IS_ERR(req)) {
			ret = PTR_ERR(req);
944
			break;
945
		}
Sage Weil's avatar
Sage Weil committed
946

Yan, Zheng's avatar
Yan, Zheng committed
947 948
		len = size;
		pages = dio_get_pages_alloc(iter, len, &start, &num_pages);
949
		if (IS_ERR(pages)) {
950
			ceph_osdc_put_request(req);
951
			ret = PTR_ERR(pages);
952
			break;
Sage Weil's avatar
Sage Weil committed
953 954 955
		}

		/*
Yan, Zheng's avatar
Yan, Zheng committed
956 957
		 * To simplify error handling, allow AIO when IO within i_size
		 * or IO can be satisfied by single OSD request.
Sage Weil's avatar
Sage Weil committed
958
		 */
Yan, Zheng's avatar
Yan, Zheng committed
959 960 961 962 963 964 965 966
		if (pos == iocb->ki_pos && !is_sync_kiocb(iocb) &&
		    (len == count || pos + count <= i_size_read(inode))) {
			aio_req = kzalloc(sizeof(*aio_req), GFP_KERNEL);
			if (aio_req) {
				aio_req->iocb = iocb;
				aio_req->write = write;
				INIT_LIST_HEAD(&aio_req->osd_reqs);
				if (write) {
967
					aio_req->mtime = mtime;
Yan, Zheng's avatar
Yan, Zheng committed
968 969 970 971 972 973 974 975 976 977 978 979
					swap(aio_req->prealloc_cf, *pcf);
				}
			}
			/* ignore error */
		}

		if (write) {
			/*
			 * throw out any page cache pages in this range. this
			 * may block.
			 */
			truncate_inode_pages_range(inode->i_mapping, pos,
980
					(pos+len) | (PAGE_SIZE - 1));
Yan, Zheng's avatar
Yan, Zheng committed
981 982

			osd_req_op_init(req, 1, CEPH_OSD_OP_STARTSYNC, 0);
983
			req->r_mtime = mtime;
Yan, Zheng's avatar
Yan, Zheng committed
984 985 986 987
		}

		osd_req_op_extent_osd_data_pages(req, 0, pages, len, start,
						 false, false);
988

Yan, Zheng's avatar
Yan, Zheng committed
989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004
		if (aio_req) {
			aio_req->total_len += len;
			aio_req->num_reqs++;
			atomic_inc(&aio_req->pending_reqs);

			req->r_callback = ceph_aio_complete_req;
			req->r_inode = inode;
			req->r_priv = aio_req;
			list_add_tail(&req->r_unsafe_item, &aio_req->osd_reqs);

			pos += len;
			iov_iter_advance(iter, len);
			continue;
		}

		ret = ceph_osdc_start_request(req->r_osdc, req, false);
1005 1006 1007
		if (!ret)
			ret = ceph_osdc_wait_request(&fsc->client->osdc, req);

Yan, Zheng's avatar
Yan, Zheng committed
1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022
		size = i_size_read(inode);
		if (!write) {
			if (ret == -ENOENT)
				ret = 0;
			if (ret >= 0 && ret < len && pos + ret < size) {
				int zlen = min_t(size_t, len - ret,
						 size - pos - ret);
				ceph_zero_page_vector_range(start + ret, zlen,
							    pages);
				ret += zlen;
			}
			if (ret >= 0)
				len = ret;
		}

1023
		ceph_put_page_vector(pages, num_pages, !write);
1024 1025

		ceph_osdc_put_request(req);
Yan, Zheng's avatar
Yan, Zheng committed
1026
		if (ret < 0)
1027
			break;
1028

Yan, Zheng's avatar
Yan, Zheng committed
1029 1030 1031 1032
		pos += len;
		iov_iter_advance(iter, len);

		if (!write && pos >= size)
1033
			break;
1034

Yan, Zheng's avatar
Yan, Zheng committed
1035 1036
		if (write && pos > size) {
			if (ceph_inode_set_size(inode, pos))
1037 1038 1039 1040
				ceph_check_caps(ceph_inode(inode),
						CHECK_CAPS_AUTHONLY,
						NULL);
		}
1041 1042
	}

Yan, Zheng's avatar
Yan, Zheng committed
1043
	if (aio_req) {
1044 1045
		LIST_HEAD(osd_reqs);

Yan, Zheng's avatar
Yan, Zheng committed
1046 1047 1048 1049 1050 1051 1052 1053
		if (aio_req->num_reqs == 0) {
			kfree(aio_req);
			return ret;
		}

		ceph_get_cap_refs(ci, write ? CEPH_CAP_FILE_WR :
					      CEPH_CAP_FILE_RD);

1054 1055 1056
		list_splice(&aio_req->osd_reqs, &osd_reqs);
		while (!list_empty(&osd_reqs)) {
			req = list_first_entry(&osd_reqs,
Yan, Zheng's avatar
Yan, Zheng committed
1057 1058 1059 1060 1061 1062 1063 1064
					       struct ceph_osd_request,
					       r_unsafe_item);
			list_del_init(&req->r_unsafe_item);
			if (ret >= 0)
				ret = ceph_osdc_start_request(req->r_osdc,
							      req, false);
			if (ret < 0) {
				req->r_result = ret;
1065
				ceph_aio_complete_req(req);
Yan, Zheng's avatar
Yan, Zheng committed
1066 1067 1068 1069 1070 1071 1072
			}
		}
		return -EIOCBQUEUED;
	}

	if (ret != -EOLDSNAPC && pos > iocb->ki_pos) {
		ret = pos - iocb->ki_pos;
1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084
		iocb->ki_pos = pos;
	}
	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.)
 */
Yan, Zheng's avatar
Yan, Zheng committed
1085
static ssize_t
1086 1087
ceph_sync_write(struct kiocb *iocb, struct iov_iter *from, loff_t pos,
		struct ceph_snap_context *snapc)
1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101
{
	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_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;
1102
	struct timespec mtime = current_time(inode);
Al Viro's avatar
Al Viro committed
1103
	size_t count = iov_iter_count(from);
1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114

	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,
1115 1116
					    pos >> PAGE_SHIFT,
					    (pos + count) >> PAGE_SHIFT);
1117 1118 1119 1120 1121 1122 1123 1124
	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;

Al Viro's avatar
Al Viro committed
1125
	while ((len = iov_iter_count(from)) > 0) {
1126 1127 1128 1129 1130
		size_t left;
		int n;

		vino = ceph_vino(inode);
		req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1131
					    vino, pos, &len, 0, 1,
1132 1133 1134 1135 1136 1137
					    CEPH_OSD_OP_WRITE, flags, snapc,
					    ci->i_truncate_seq,
					    ci->i_truncate_size,
					    false);
		if (IS_ERR(req)) {
			ret = PTR_ERR(req);
1138
			break;
1139 1140 1141 1142 1143 1144
		}

		/*
		 * write from beginning of first page,
		 * regardless of io alignment
		 */
1145
		num_pages = (len + PAGE_SIZE - 1) >> PAGE_SHIFT;
1146

1147
		pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
Sage Weil's avatar
Sage Weil committed
1148 1149 1150 1151
		if (IS_ERR(pages)) {
			ret = PTR_ERR(pages);
			goto out;
		}
1152 1153 1154

		left = len;
		for (n = 0; n < num_pages; n++) {
1155
			size_t plen = min_t(size_t, left, PAGE_SIZE);
Al Viro's avatar
Al Viro committed
1156
			ret = copy_page_from_iter(pages[n], 0, plen, from);
1157 1158 1159 1160 1161 1162 1163
			if (ret != plen) {
				ret = -EFAULT;
				break;
			}
			left -= ret;
		}

Sage Weil's avatar
Sage Weil committed
1164 1165 1166 1167 1168
		if (ret < 0) {
			ceph_release_page_vector(pages, num_pages);
			goto out;
		}

1169 1170 1171
		/* get a second commit callback */
		req->r_unsafe_callback = ceph_sync_write_unsafe;
		req->r_inode = inode;
Sage Weil's avatar
Sage Weil committed
1172

1173 1174
		osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
						false, true);
1175

1176
		req->r_mtime = mtime;
1177 1178 1179
		ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
		if (!ret)
			ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
Sage Weil's avatar
Sage Weil committed
1180 1181

out:
1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196
		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;
	}
Sage Weil's avatar
Sage Weil committed
1197

1198
	if (ret != -EOLDSNAPC && written > 0) {
Sage Weil's avatar
Sage Weil committed
1199
		ret = written;
1200
		iocb->ki_pos = pos;
Sage Weil's avatar
Sage Weil committed
1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211
	}
	return ret;
}

/*
 * Wrap generic_file_aio_read with checks for cap bits on the inode.
 * Atomically grab references, so that those bits are not released
 * back to the MDS mid-read.
 *
 * Hmm, the sync read case isn't actually async... should it be?
 */
Al Viro's avatar
Al Viro committed
1212
static ssize_t ceph_read_iter(struct kiocb *iocb, struct iov_iter *to)
Sage Weil's avatar
Sage Weil committed
1213 1214
{
	struct file *filp = iocb->ki_filp;
1215
	struct ceph_file_info *fi = filp->private_data;
Christoph Hellwig's avatar
Christoph Hellwig committed
1216
	size_t len = iov_iter_count(to);
Al Viro's avatar
Al Viro committed
1217
	struct inode *inode = file_inode(filp);
Sage Weil's avatar
Sage Weil committed
1218
	struct ceph_inode_info *ci = ceph_inode(inode);
1219
	struct page *pinned_page = NULL;
Sage Weil's avatar
Sage Weil committed
1220
	ssize_t ret;
1221
	int want, got = 0;
Yan, Zheng's avatar
Yan, Zheng committed
1222
	int retry_op = 0, read = 0;
Sage Weil's avatar
Sage Weil committed
1223

1224
again:
1225 1226 1227
	dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
	     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, inode);

1228 1229 1230 1231
	if (fi->fmode & CEPH_FILE_MODE_LAZY)
		want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
	else
		want = CEPH_CAP_FILE_CACHE;
1232
	ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, -1, &got, &pinned_page);
Sage Weil's avatar
Sage Weil committed
1233
	if (ret < 0)
1234
		return ret;
Sage Weil's avatar
Sage Weil committed
1235

1236
	if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
1237
	    (iocb->ki_flags & IOCB_DIRECT) ||
1238 1239 1240 1241 1242 1243
	    (fi->flags & CEPH_F_SYNC)) {

		dout("aio_sync_read %p %llx.%llx %llu~%u got cap refs on %s\n",
		     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
		     ceph_cap_string(got));

Yan, Zheng's avatar
Yan, Zheng committed
1244
		if (ci->i_inline_version == CEPH_INLINE_NONE) {
Yan, Zheng's avatar
Yan, Zheng committed
1245 1246 1247 1248 1249 1250 1251 1252
			if (!retry_op && (iocb->ki_flags & IOCB_DIRECT)) {
				ret = ceph_direct_read_write(iocb, to,
							     NULL, NULL);
				if (ret >= 0 && ret < len)
					retry_op = CHECK_EOF;
			} else {
				ret = ceph_sync_read(iocb, to, &retry_op);
			}
Yan, Zheng's avatar
Yan, Zheng committed
1253 1254 1255
		} else {
			retry_op = READ_INLINE;
		}
1256 1257
	} else {
		dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
Al Viro's avatar
Al Viro committed
1258
		     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
1259
		     ceph_cap_string(got));
1260
		current->journal_info = filp;
Al Viro's avatar
Al Viro committed
1261
		ret = generic_file_read_iter(iocb, to);
1262
		current->journal_info = NULL;
1263
	}
Sage Weil's avatar
Sage Weil committed
1264 1265
	dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
	     inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
1266
	if (pinned_page) {
1267
		put_page(pinned_page);
1268 1269
		pinned_page = NULL;
	}
Sage Weil's avatar
Sage Weil committed
1270
	ceph_put_cap_refs(ci, got);
Yan, Zheng's avatar
Yan, Zheng committed
1271
	if (retry_op > HAVE_RETRIED && ret >= 0) {
Yan, Zheng's avatar
Yan, Zheng committed
1272 1273 1274 1275
		int statret;
		struct page *page = NULL;
		loff_t i_size;
		if (retry_op == READ_INLINE) {
1276
			page = __page_cache_alloc(GFP_KERNEL);
Yan, Zheng's avatar
Yan, Zheng committed
1277 1278 1279
			if (!page)
				return -ENOMEM;
		}
1280

Yan, Zheng's avatar
Yan, Zheng committed
1281 1282 1283
		statret = __ceph_do_getattr(inode, page,
					    CEPH_STAT_CAP_INLINE_DATA, !!page);
		if (statret < 0) {
1284 1285
			if (page)
				__free_page(page);
Yan, Zheng's avatar
Yan, Zheng committed
1286 1287 1288 1289 1290 1291
			if (statret == -ENODATA) {
				BUG_ON(retry_op != READ_INLINE);
				goto again;
			}
			return statret;
		}
1292

Yan, Zheng's avatar
Yan, Zheng committed
1293 1294
		i_size = i_size_read(inode);
		if (retry_op == READ_INLINE) {
1295 1296
			BUG_ON(ret > 0 || read > 0);
			if (iocb->ki_pos < i_size &&
1297
			    iocb->ki_pos < PAGE_SIZE) {
Yan, Zheng's avatar
Yan, Zheng committed
1298 1299
				loff_t end = min_t(loff_t, i_size,
						   iocb->ki_pos + len);
1300
				end = min_t(loff_t, end, PAGE_SIZE);
Yan, Zheng's avatar
Yan, Zheng committed
1301 1302 1303 1304 1305 1306
				if (statret < end)
					zero_user_segment(page, statret, end);
				ret = copy_page_to_iter(page,
						iocb->ki_pos & ~PAGE_MASK,
						end - iocb->ki_pos, to);
				iocb->ki_pos += ret;
1307 1308 1309 1310 1311 1312 1313 1314
				read += ret;
			}
			if (iocb->ki_pos < i_size && read < len) {
				size_t zlen = min_t(size_t, len - read,
						    i_size - iocb->ki_pos);
				ret = iov_iter_zero(zlen, to);
				iocb->ki_pos += ret;
				read += ret;
Yan, Zheng's avatar
Yan, Zheng committed
1315 1316
			}
			__free_pages(page, 0);
1317
			return read;
Yan, Zheng's avatar
Yan, Zheng committed
1318
		}
1319 1320

		/* hit EOF or hole? */
Yan, Zheng's avatar
Yan, Zheng committed
1321
		if (retry_op == CHECK_EOF && iocb->ki_pos < i_size &&
1322
		    ret < len) {
1323
			dout("sync_read hit hole, ppos %lld < size %lld"
1324
			     ", reading more\n", iocb->ki_pos, i_size);
1325

1326 1327
			read += ret;
			len -= ret;
Yan, Zheng's avatar
Yan, Zheng committed
1328
			retry_op = HAVE_RETRIED;
1329 1330 1331
			goto again;
		}
	}
1332

1333 1334 1335
	if (ret >= 0)
		ret += read;

Sage Weil's avatar
Sage Weil committed
1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
	return ret;
}

/*
 * Take cap references to avoid releasing caps to MDS mid-write.
 *
 * If we are synchronous, and write with an old snap context, the OSD
 * may return EOLDSNAPC.  In that case, retry the write.. _after_
 * dropping our cap refs and allowing the pending snap to logically
 * complete _before_ this write occurs.
 *
 * If we are near ENOSPC, write synchronously.
 */
Al Viro's avatar
Al Viro committed
1349
static ssize_t ceph_write_iter(struct kiocb *iocb, struct iov_iter *from)
Sage Weil's avatar
Sage Weil committed
1350 1351
{
	struct file *file = iocb->ki_filp;
1352
	struct ceph_file_info *fi = file->private_data;
Al Viro's avatar
Al Viro committed
1353
	struct inode *inode = file_inode(file);
Sage Weil's avatar
Sage Weil committed
1354
	struct ceph_inode_info *ci = ceph_inode(inode);
1355 1356
	struct ceph_osd_client *osdc =
		&ceph_sb_to_client(inode->i_sb)->client->osdc;
1357
	struct ceph_cap_flush *prealloc_cf;
1358
	ssize_t count, written = 0;
1359
	int err, want, got;
1360
	loff_t pos;
Sage Weil's avatar
Sage Weil committed
1361 1362 1363 1364

	if (ceph_snap(inode) != CEPH_NOSNAP)
		return -EROFS;

1365 1366 1367 1368
	prealloc_cf = ceph_alloc_cap_flush();
	if (!prealloc_cf)
		return -ENOMEM;

Al Viro's avatar
Al Viro committed
1369
	inode_lock(inode);