[deliverable/linux.git] / fs / nfs / pnfs_nfs.c

/*
 * Common NFS I/O  operations for the pnfs file based
 * layout drivers.
 *
 * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
 *
 * Tom Haynes <loghyr@primarydata.com>
 */

#include <linux/nfs_fs.h>
#include <linux/nfs_page.h>

#include "internal.h"
#include "pnfs.h"

static void pnfs_generic_fenceme(struct inode *inode,
				 struct pnfs_layout_hdr *lo)
{
	if (!test_and_clear_bit(NFS_LAYOUT_RETURN, &lo->plh_flags))
		return;
	pnfs_return_layout(inode);
}

void pnfs_generic_rw_release(void *data)
{
	struct nfs_pgio_header *hdr = data;
	struct pnfs_layout_hdr *lo = hdr->lseg->pls_layout;

	pnfs_generic_fenceme(lo->plh_inode, lo);
	nfs_put_client(hdr->ds_clp);
	hdr->mds_ops->rpc_release(data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_rw_release);

/* Fake up some data that will cause nfs_commit_release to retry the writes. */
void pnfs_generic_prepare_to_resend_writes(struct nfs_commit_data *data)
{
	struct nfs_page *first = nfs_list_entry(data->pages.next);

	data->task.tk_status = 0;
	memcpy(&data->verf.verifier, &first->wb_verf,
	       sizeof(data->verf.verifier));
	data->verf.verifier.data[0]++; /* ensure verifier mismatch */
}
EXPORT_SYMBOL_GPL(pnfs_generic_prepare_to_resend_writes);

void pnfs_generic_write_commit_done(struct rpc_task *task, void *data)
{
	struct nfs_commit_data *wdata = data;

	/* Note this may cause RPC to be resent */
	wdata->mds_ops->rpc_call_done(task, data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_write_commit_done);

void pnfs_generic_commit_release(void *calldata)
{
	struct nfs_commit_data *data = calldata;

	data->completion_ops->completion(data);
	pnfs_put_lseg(data->lseg);
	nfs_put_client(data->ds_clp);
	nfs_commitdata_release(data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_commit_release);

/* The generic layer is about to remove the req from the commit list.
 * If this will make the bucket empty, it will need to put the lseg reference.
 * Note this is must be called holding the inode (/cinfo) lock
 */
void
pnfs_generic_clear_request_commit(struct nfs_page *req,
				  struct nfs_commit_info *cinfo)
{
	struct pnfs_layout_segment *freeme = NULL;

	if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
		goto out;
	cinfo->ds->nwritten--;
	if (list_is_singular(&req->wb_list)) {
		struct pnfs_commit_bucket *bucket;

		bucket = list_first_entry(&req->wb_list,
					  struct pnfs_commit_bucket,
					  written);
		freeme = bucket->wlseg;
		bucket->wlseg = NULL;
	}
out:
	nfs_request_remove_commit_list(req, cinfo);
	pnfs_put_lseg_locked(freeme);
}
EXPORT_SYMBOL_GPL(pnfs_generic_clear_request_commit);

static int
pnfs_generic_transfer_commit_list(struct list_head *src, struct list_head *dst,
				  struct nfs_commit_info *cinfo, int max)
{
	struct nfs_page *req, *tmp;
	int ret = 0;

	list_for_each_entry_safe(req, tmp, src, wb_list) {
		if (!nfs_lock_request(req))
			continue;
		kref_get(&req->wb_kref);
		if (cond_resched_lock(cinfo->lock))
			list_safe_reset_next(req, tmp, wb_list);
		nfs_request_remove_commit_list(req, cinfo);
		clear_bit(PG_COMMIT_TO_DS, &req->wb_flags);
		nfs_list_add_request(req, dst);
		ret++;
		if ((ret == max) && !cinfo->dreq)
			break;
	}
	return ret;
}

/* Note called with cinfo->lock held. */
static int
pnfs_generic_scan_ds_commit_list(struct pnfs_commit_bucket *bucket,
				 struct nfs_commit_info *cinfo,
				 int max)
{
	struct list_head *src = &bucket->written;
	struct list_head *dst = &bucket->committing;
	int ret;

	ret = pnfs_generic_transfer_commit_list(src, dst, cinfo, max);
	if (ret) {
		cinfo->ds->nwritten -= ret;
		cinfo->ds->ncommitting += ret;
		bucket->clseg = bucket->wlseg;
		if (list_empty(src))
			bucket->wlseg = NULL;
		else
			pnfs_get_lseg(bucket->clseg);
	}
	return ret;
}

/* Move reqs from written to committing lists, returning count of number moved.
 * Note called with cinfo->lock held.
 */
int pnfs_generic_scan_commit_lists(struct nfs_commit_info *cinfo,
				   int max)
{
	int i, rv = 0, cnt;

	for (i = 0; i < cinfo->ds->nbuckets && max != 0; i++) {
		cnt = pnfs_generic_scan_ds_commit_list(&cinfo->ds->buckets[i],
						       cinfo, max);
		max -= cnt;
		rv += cnt;
	}
	return rv;
}
EXPORT_SYMBOL_GPL(pnfs_generic_scan_commit_lists);

/* Pull everything off the committing lists and dump into @dst */
void pnfs_generic_recover_commit_reqs(struct list_head *dst,
				      struct nfs_commit_info *cinfo)
{
	struct pnfs_commit_bucket *b;
	struct pnfs_layout_segment *freeme;
	int i;

restart:
	spin_lock(cinfo->lock);
	for (i = 0, b = cinfo->ds->buckets; i < cinfo->ds->nbuckets; i++, b++) {
		if (pnfs_generic_transfer_commit_list(&b->written, dst,
						      cinfo, 0)) {
			freeme = b->wlseg;
			b->wlseg = NULL;
			spin_unlock(cinfo->lock);
			pnfs_put_lseg(freeme);
			goto restart;
		}
	}
	cinfo->ds->nwritten = 0;
	spin_unlock(cinfo->lock);
}
EXPORT_SYMBOL_GPL(pnfs_generic_recover_commit_reqs);

static void pnfs_generic_retry_commit(struct nfs_commit_info *cinfo, int idx)
{
	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
	struct pnfs_commit_bucket *bucket;
	struct pnfs_layout_segment *freeme;
	int i;

	for (i = idx; i < fl_cinfo->nbuckets; i++) {
		bucket = &fl_cinfo->buckets[i];
		if (list_empty(&bucket->committing))
			continue;
		nfs_retry_commit(&bucket->committing, bucket->clseg, cinfo);
		spin_lock(cinfo->lock);
		freeme = bucket->clseg;
		bucket->clseg = NULL;
		spin_unlock(cinfo->lock);
		pnfs_put_lseg(freeme);
	}
}

static unsigned int
pnfs_generic_alloc_ds_commits(struct nfs_commit_info *cinfo,
			      struct list_head *list)
{
	struct pnfs_ds_commit_info *fl_cinfo;
	struct pnfs_commit_bucket *bucket;
	struct nfs_commit_data *data;
	int i;
	unsigned int nreq = 0;

	fl_cinfo = cinfo->ds;
	bucket = fl_cinfo->buckets;
	for (i = 0; i < fl_cinfo->nbuckets; i++, bucket++) {
		if (list_empty(&bucket->committing))
			continue;
		data = nfs_commitdata_alloc();
		if (!data)
			break;
		data->ds_commit_index = i;
		spin_lock(cinfo->lock);
		data->lseg = bucket->clseg;
		bucket->clseg = NULL;
		spin_unlock(cinfo->lock);
		list_add(&data->pages, list);
		nreq++;
	}

	/* Clean up on error */
	pnfs_generic_retry_commit(cinfo, i);
	return nreq;
}

/* This follows nfs_commit_list pretty closely */
int
pnfs_generic_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
			     int how, struct nfs_commit_info *cinfo,
			     int (*initiate_commit)(struct nfs_commit_data *data,
						    int how))
{
	struct nfs_commit_data *data, *tmp;
	LIST_HEAD(list);
	unsigned int nreq = 0;

	if (!list_empty(mds_pages)) {
		data = nfs_commitdata_alloc();
		if (data != NULL) {
			data->lseg = NULL;
			list_add(&data->pages, &list);
			nreq++;
		} else {
			nfs_retry_commit(mds_pages, NULL, cinfo);
			pnfs_generic_retry_commit(cinfo, 0);
			cinfo->completion_ops->error_cleanup(NFS_I(inode));
			return -ENOMEM;
		}
	}

	nreq += pnfs_generic_alloc_ds_commits(cinfo, &list);

	if (nreq == 0) {
		cinfo->completion_ops->error_cleanup(NFS_I(inode));
		goto out;
	}

	atomic_add(nreq, &cinfo->mds->rpcs_out);

	list_for_each_entry_safe(data, tmp, &list, pages) {
		list_del_init(&data->pages);
		if (!data->lseg) {
			nfs_init_commit(data, mds_pages, NULL, cinfo);
			nfs_initiate_commit(NFS_CLIENT(inode), data,
					    data->mds_ops, how, 0);
		} else {
			struct pnfs_commit_bucket *buckets;

			buckets = cinfo->ds->buckets;
			nfs_init_commit(data,
					&buckets[data->ds_commit_index].committing,
					data->lseg,
					cinfo);
			initiate_commit(data, how);
		}
	}
out:
	cinfo->ds->ncommitting = 0;
	return PNFS_ATTEMPTED;
}
EXPORT_SYMBOL_GPL(pnfs_generic_commit_pagelist);
Commit	Line	Data
f54bcf2e TH	1	/*
	2	* Common NFS I/O operations for the pnfs file based
	3	* layout drivers.
	4	*
	5	* Copyright (c) 2014, Primary Data, Inc. All rights reserved.
	6	*
	7	* Tom Haynes <loghyr@primarydata.com>
	8	*/
	9
	10	#include <linux/nfs_fs.h>
	11	#include <linux/nfs_page.h>
	12
	13	#include "internal.h"
	14	#include "pnfs.h"
	15
	16	static void pnfs_generic_fenceme(struct inode *inode,
	17	struct pnfs_layout_hdr *lo)
	18	{
	19	if (!test_and_clear_bit(NFS_LAYOUT_RETURN, &lo->plh_flags))
	20	return;
	21	pnfs_return_layout(inode);
	22	}
	23
	24	void pnfs_generic_rw_release(void *data)
	25	{
	26	struct nfs_pgio_header *hdr = data;
	27	struct pnfs_layout_hdr *lo = hdr->lseg->pls_layout;
	28
	29	pnfs_generic_fenceme(lo->plh_inode, lo);
	30	nfs_put_client(hdr->ds_clp);
	31	hdr->mds_ops->rpc_release(data);
	32	}
	33	EXPORT_SYMBOL_GPL(pnfs_generic_rw_release);
	34
	35	/* Fake up some data that will cause nfs_commit_release to retry the writes. */
	36	void pnfs_generic_prepare_to_resend_writes(struct nfs_commit_data *data)
	37	{
	38	struct nfs_page *first = nfs_list_entry(data->pages.next);
	39
	40	data->task.tk_status = 0;
	41	memcpy(&data->verf.verifier, &first->wb_verf,
	42	sizeof(data->verf.verifier));
	43	data->verf.verifier.data[0]++; /* ensure verifier mismatch */
	44	}
	45	EXPORT_SYMBOL_GPL(pnfs_generic_prepare_to_resend_writes);
	46
	47	void pnfs_generic_write_commit_done(struct rpc_task task, void data)
	48	{
	49	struct nfs_commit_data *wdata = data;
	50
	51	/* Note this may cause RPC to be resent */
	52	wdata->mds_ops->rpc_call_done(task, data);
	53	}
	54	EXPORT_SYMBOL_GPL(pnfs_generic_write_commit_done);
	55
	56	void pnfs_generic_commit_release(void *calldata)
	57	{
	58	struct nfs_commit_data *data = calldata;
	59
	60	data->completion_ops->completion(data);
	61	pnfs_put_lseg(data->lseg);
	62	nfs_put_client(data->ds_clp);
	63	nfs_commitdata_release(data);
	64	}
65	EXPORT_SYMBOL_GPL(pnfs_generic_commit_release);
66
67	/* The generic layer is about to remove the req from the commit list.
68	* If this will make the bucket empty, it will need to put the lseg reference.
69	* Note this is must be called holding the inode (/cinfo) lock
70	*/
71	void
72	pnfs_generic_clear_request_commit(struct nfs_page *req,
73	struct nfs_commit_info *cinfo)
74	{
75	struct pnfs_layout_segment *freeme = NULL;
76
77	if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
78	goto out;
79	cinfo->ds->nwritten--;
80	if (list_is_singular(&req->wb_list)) {
81	struct pnfs_commit_bucket *bucket;
82
83	bucket = list_first_entry(&req->wb_list,
84	struct pnfs_commit_bucket,
85	written);
86	freeme = bucket->wlseg;
87	bucket->wlseg = NULL;
88	}
89	out:
90	nfs_request_remove_commit_list(req, cinfo);
91	pnfs_put_lseg_locked(freeme);
92	}
93	EXPORT_SYMBOL_GPL(pnfs_generic_clear_request_commit);
94
95	static int
96	pnfs_generic_transfer_commit_list(struct list_head src, struct list_head dst,
97	struct nfs_commit_info *cinfo, int max)
98	{
99	struct nfs_page req, tmp;
100	int ret = 0;
101
102	list_for_each_entry_safe(req, tmp, src, wb_list) {
103	if (!nfs_lock_request(req))
104	continue;
105	kref_get(&req->wb_kref);
106	if (cond_resched_lock(cinfo->lock))
107	list_safe_reset_next(req, tmp, wb_list);
108	nfs_request_remove_commit_list(req, cinfo);
109	clear_bit(PG_COMMIT_TO_DS, &req->wb_flags);
110	nfs_list_add_request(req, dst);
111	ret++;
112	if ((ret == max) && !cinfo->dreq)
113	break;
114	}
115	return ret;
116	}
117
118	/* Note called with cinfo->lock held. */
119	static int
120	pnfs_generic_scan_ds_commit_list(struct pnfs_commit_bucket *bucket,
121	struct nfs_commit_info *cinfo,
122	int max)
123	{
124	struct list_head *src = &bucket->written;
125	struct list_head *dst = &bucket->committing;
126	int ret;
127
128	ret = pnfs_generic_transfer_commit_list(src, dst, cinfo, max);
129	if (ret) {
130	cinfo->ds->nwritten -= ret;
131	cinfo->ds->ncommitting += ret;
132	bucket->clseg = bucket->wlseg;
133	if (list_empty(src))
134	bucket->wlseg = NULL;
135	else
136	pnfs_get_lseg(bucket->clseg);
137	}
138	return ret;
139	}
140
141	/* Move reqs from written to committing lists, returning count of number moved.
142	* Note called with cinfo->lock held.
143	*/
144	int pnfs_generic_scan_commit_lists(struct nfs_commit_info *cinfo,
145	int max)
146	{
147	int i, rv = 0, cnt;
148
149	for (i = 0; i < cinfo->ds->nbuckets && max != 0; i++) {
150	cnt = pnfs_generic_scan_ds_commit_list(&cinfo->ds->buckets[i],
151	cinfo, max);
152	max -= cnt;
153	rv += cnt;
154	}
155	return rv;
156	}
157	EXPORT_SYMBOL_GPL(pnfs_generic_scan_commit_lists);
158
159	/* Pull everything off the committing lists and dump into @dst */
160	void pnfs_generic_recover_commit_reqs(struct list_head *dst,
161	struct nfs_commit_info *cinfo)
162	{
163	struct pnfs_commit_bucket *b;
164	struct pnfs_layout_segment *freeme;
165	int i;
166
167	restart:
168	spin_lock(cinfo->lock);
169	for (i = 0, b = cinfo->ds->buckets; i < cinfo->ds->nbuckets; i++, b++) {
170	if (pnfs_generic_transfer_commit_list(&b->written, dst,
171	cinfo, 0)) {
172	freeme = b->wlseg;
173	b->wlseg = NULL;
174	spin_unlock(cinfo->lock);
175	pnfs_put_lseg(freeme);
176	goto restart;
177	}
178	}
179	cinfo->ds->nwritten = 0;
180	spin_unlock(cinfo->lock);
181	}
182	EXPORT_SYMBOL_GPL(pnfs_generic_recover_commit_reqs);
183
184	static void pnfs_generic_retry_commit(struct nfs_commit_info *cinfo, int idx)
185	{
186	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
187	struct pnfs_commit_bucket *bucket;
188	struct pnfs_layout_segment *freeme;
189	int i;
190
191	for (i = idx; i < fl_cinfo->nbuckets; i++) {
192	bucket = &fl_cinfo->buckets[i];
193	if (list_empty(&bucket->committing))
194	continue;
195	nfs_retry_commit(&bucket->committing, bucket->clseg, cinfo);
196	spin_lock(cinfo->lock);
197	freeme = bucket->clseg;
198	bucket->clseg = NULL;
199	spin_unlock(cinfo->lock);
200	pnfs_put_lseg(freeme);
201	}
202	}
203
204	static unsigned int
205	pnfs_generic_alloc_ds_commits(struct nfs_commit_info *cinfo,
206	struct list_head *list)
207	{
208	struct pnfs_ds_commit_info *fl_cinfo;
209	struct pnfs_commit_bucket *bucket;
210	struct nfs_commit_data *data;
211	int i;
212	unsigned int nreq = 0;
213
214	fl_cinfo = cinfo->ds;
215	bucket = fl_cinfo->buckets;
216	for (i = 0; i < fl_cinfo->nbuckets; i++, bucket++) {
217	if (list_empty(&bucket->committing))
218	continue;
219	data = nfs_commitdata_alloc();
220	if (!data)
221	break;
222	data->ds_commit_index = i;
223	spin_lock(cinfo->lock);
224	data->lseg = bucket->clseg;
225	bucket->clseg = NULL;
226	spin_unlock(cinfo->lock);
227	list_add(&data->pages, list);
228	nreq++;
229	}
230
231	/* Clean up on error */
232	pnfs_generic_retry_commit(cinfo, i);
233	return nreq;
234	}
235
236	/* This follows nfs_commit_list pretty closely */
237	int
238	pnfs_generic_commit_pagelist(struct inode inode, struct list_head mds_pages,
239	int how, struct nfs_commit_info *cinfo,
240	int (initiate_commit)(struct nfs_commit_data data,
241	int how))
242	{
243	struct nfs_commit_data data, tmp;
244	LIST_HEAD(list);
245	unsigned int nreq = 0;
246
247	if (!list_empty(mds_pages)) {
248	data = nfs_commitdata_alloc();
249	if (data != NULL) {
250	data->lseg = NULL;
251	list_add(&data->pages, &list);
252	nreq++;
253	} else {
254	nfs_retry_commit(mds_pages, NULL, cinfo);
255	pnfs_generic_retry_commit(cinfo, 0);
256	cinfo->completion_ops->error_cleanup(NFS_I(inode));
257	return -ENOMEM;
258	}
259	}
260
261	nreq += pnfs_generic_alloc_ds_commits(cinfo, &list);
262
263	if (nreq == 0) {
264	cinfo->completion_ops->error_cleanup(NFS_I(inode));
265	goto out;
266	}
267
268	atomic_add(nreq, &cinfo->mds->rpcs_out);
269
270	list_for_each_entry_safe(data, tmp, &list, pages) {
271	list_del_init(&data->pages);
272	if (!data->lseg) {
273	nfs_init_commit(data, mds_pages, NULL, cinfo);
274	nfs_initiate_commit(NFS_CLIENT(inode), data,
275	data->mds_ops, how, 0);
276	} else {
277	struct pnfs_commit_bucket *buckets;
278
279	buckets = cinfo->ds->buckets;
280	nfs_init_commit(data,
281	&buckets[data->ds_commit_index].committing,
282	data->lseg,
283	cinfo);
284	initiate_commit(data, how);
285	}
286	}
287	out:
288	cinfo->ds->ncommitting = 0;
289	return PNFS_ATTEMPTED;
290	}
291	EXPORT_SYMBOL_GPL(pnfs_generic_commit_pagelist);