Whamcloud - gitweb
83f2528a7dae317257c07a277a26ec0ede917d97
[fs/lustre-release.git] / lustre / quota / qsd_request.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, write to the
18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
19  * Boston, MA 021110-1307, USA
20  *
21  * GPL HEADER END
22  */
23 /*
24  * Copyright (c) 2012, 2013, Intel Corporation.
25  * Use is subject to license terms.
26  *
27  * Author: Johann Lombardi <johann.lombardi@intel.com>
28  * Author: Niu    Yawei    <yawei.niu@intel.com>
29  */
30
31 #define DEBUG_SUBSYSTEM S_LQUOTA
32
33 #include <lustre_net.h>
34 #include <lustre_import.h>
35 #include <lustre_dlm.h>
36 #include <obd_class.h>
37
38 #include "qsd_internal.h"
39
40 struct qsd_async_args {
41         struct obd_export     *aa_exp;
42         struct qsd_qtype_info *aa_qqi;
43         void                  *aa_arg;
44         struct lquota_lvb     *aa_lvb;
45         struct lustre_handle   aa_lockh;
46         qsd_req_completion_t   aa_completion;
47 };
48
49 /*
50  * non-intent quota request interpret callback.
51  *
52  * \param env    - the environment passed by the caller
53  * \param req    - the non-intent quota request
54  * \param arg    - qsd_async_args
55  * \param rc     - request status
56  *
57  * \retval 0     - success
58  * \retval -ve   - appropriate errors
59  */
60 static int qsd_dqacq_interpret(const struct lu_env *env,
61                                struct ptlrpc_request *req, void *arg, int rc)
62 {
63         struct quota_body     *rep_qbody = NULL, *req_qbody;
64         struct qsd_async_args *aa = (struct qsd_async_args *)arg;
65         ENTRY;
66
67         req_qbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
68         if (rc == 0 || rc == -EDQUOT || rc == -EINPROGRESS)
69                 rep_qbody = req_capsule_server_get(&req->rq_pill,
70                                                    &RMF_QUOTA_BODY);
71         aa->aa_completion(env, aa->aa_qqi, req_qbody, rep_qbody, &aa->aa_lockh,
72                           NULL, aa->aa_arg, rc);
73         RETURN(rc);
74 }
75
76 /*
77  * Send non-intent quota request to master.
78  *
79  * \param env    - the environment passed by the caller
80  * \param exp    - is the export to use to send the acquire RPC
81  * \param qbody  - quota body to be packed in request
82  * \param sync   - synchronous or asynchronous
83  * \param completion - completion callback
84  * \param qqi    - is the qsd_qtype_info structure to pass to the completion
85  *                 function
86  * \param lqe    - is the qid entry to be processed
87  *
88  * \retval 0     - success
89  * \retval -ve   - appropriate errors
90  */
91 int qsd_send_dqacq(const struct lu_env *env, struct obd_export *exp,
92                    struct quota_body *qbody, bool sync,
93                    qsd_req_completion_t completion, struct qsd_qtype_info *qqi,
94                    struct lustre_handle *lockh, struct lquota_entry *lqe)
95 {
96         struct ptlrpc_request   *req;
97         struct quota_body       *req_qbody;
98         struct qsd_async_args   *aa;
99         int                      rc;
100         ENTRY;
101
102         LASSERT(exp);
103
104         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_QUOTA_DQACQ);
105         if (req == NULL)
106                 GOTO(out, rc = -ENOMEM);
107
108         req->rq_no_resend = req->rq_no_delay = 1;
109         req->rq_no_retry_einprogress = 1;
110         rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, QUOTA_DQACQ);
111         if (rc) {
112                 ptlrpc_request_free(req);
113                 GOTO(out, rc);
114         }
115
116         req_qbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
117         *req_qbody = *qbody;
118
119         ptlrpc_request_set_replen(req);
120
121         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
122         aa = ptlrpc_req_async_args(req);
123         aa->aa_exp = exp;
124         aa->aa_qqi = qqi;
125         aa->aa_arg = (void *)lqe;
126         aa->aa_completion = completion;
127         lustre_handle_copy(&aa->aa_lockh, lockh);
128
129         if (sync) {
130                 rc = ptlrpc_queue_wait(req);
131                 rc = qsd_dqacq_interpret(env, req, aa, rc);
132                 ptlrpc_req_finished(req);
133         } else {
134                 req->rq_interpret_reply = qsd_dqacq_interpret;
135                 ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
136         }
137
138         RETURN(rc);
139 out:
140         completion(env, qqi, qbody, NULL, lockh, NULL, lqe, rc);
141         return rc;
142 }
143
144 /*
145  * intent quota request interpret callback.
146  *
147  * \param env    - the environment passed by the caller
148  * \param req    - the intent quota request
149  * \param arg    - qsd_async_args
150  * \param rc     - request status
151  *
152  * \retval 0     - success
153  * \retval -ve   - appropriate errors
154  */
155 static int qsd_intent_interpret(const struct lu_env *env,
156                                 struct ptlrpc_request *req, void *arg, int rc)
157 {
158         struct lustre_handle     *lockh;
159         struct quota_body        *rep_qbody = NULL, *req_qbody;
160         struct ldlm_intent       *lit;
161         struct qsd_async_args    *aa = (struct qsd_async_args *)arg;
162         __u64                     flags = LDLM_FL_HAS_INTENT;
163         ENTRY;
164
165         LASSERT(aa->aa_exp);
166         lockh = &aa->aa_lockh;
167         req_qbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
168         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
169
170         rc = ldlm_cli_enqueue_fini(aa->aa_exp, req, LDLM_PLAIN, 0, LCK_CR,
171                                    &flags, (void *)aa->aa_lvb,
172                                    sizeof(struct lquota_lvb), lockh, rc);
173         if (rc < 0)
174                 /* the lock has been destroyed, forget about the lock handle */
175                 memset(lockh, 0, sizeof(*lockh));
176
177         if (rc == 0 || rc == -EDQUOT || rc == -EINPROGRESS)
178                 rep_qbody = req_capsule_server_get(&req->rq_pill,
179                                                    &RMF_QUOTA_BODY);
180
181         aa->aa_completion(env, aa->aa_qqi, req_qbody, rep_qbody, lockh,
182                           aa->aa_lvb, aa->aa_arg, rc);
183         RETURN(rc);
184 }
185
186 /*
187  * Get intent per-ID lock or global-index lock from master.
188  *
189  * \param env    - the environment passed by the caller
190  * \param exp    - is the export to use to send the intent RPC
191  * \param qbody  - quota body to be packed in request
192  * \param sync   - synchronous or asynchronous (pre-acquire)
193  * \param it_op  - IT_QUOTA_DQACQ or IT_QUOTA_CONN
194  * \param completion - completion callback
195  * \param qqi    - is the qsd_qtype_info structure to pass to the completion
196  *                 function
197  * \param lvb    - is the lvb associated with the lock and returned by the
198  *                 server
199  * \param arg    - is an opaq argument passed to the completion callback
200  *
201  * \retval 0     - success
202  * \retval -ve   - appropriate errors
203  */
204 int qsd_intent_lock(const struct lu_env *env, struct obd_export *exp,
205                     struct quota_body *qbody, bool sync, int it_op,
206                     qsd_req_completion_t completion, struct qsd_qtype_info *qqi,
207                     struct lquota_lvb *lvb, void *arg)
208 {
209         struct qsd_thread_info  *qti = qsd_info(env);
210         struct ptlrpc_request   *req;
211         struct qsd_async_args   *aa = NULL;
212         struct ldlm_intent      *lit;
213         struct quota_body       *req_qbody;
214         __u64                    flags = LDLM_FL_HAS_INTENT;
215         int                      rc;
216         ENTRY;
217
218         LASSERT(exp != NULL);
219         LASSERT(!lustre_handle_is_used(&qbody->qb_lockh));
220
221         memset(&qti->qti_lockh, 0, sizeof(qti->qti_lockh));
222
223         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
224                                    &RQF_LDLM_INTENT_QUOTA);
225         if (req == NULL)
226                 GOTO(out, rc = -ENOMEM);
227
228         req->rq_no_retry_einprogress = 1;
229         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
230         if (rc) {
231                 ptlrpc_request_free(req);
232                 GOTO(out, rc);
233         }
234
235         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
236         lit->opc = (__u64)it_op;
237
238         req_qbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
239         *req_qbody = *qbody;
240
241         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
242                              sizeof(*lvb));
243         ptlrpc_request_set_replen(req);
244
245         switch(it_op) {
246         case IT_QUOTA_CONN:
247                 /* build resource name associated with global index */
248                 fid_build_reg_res_name(&qbody->qb_fid, &qti->qti_resid);
249
250                 /* copy einfo template and fill ei_cbdata with qqi pointer */
251                 memcpy(&qti->qti_einfo, &qsd_glb_einfo, sizeof(qti->qti_einfo));
252                 qti->qti_einfo.ei_cbdata = qqi;
253
254                 /* don't cancel global lock on memory pressure */
255                 flags |= LDLM_FL_NO_LRU;
256                 break;
257         case IT_QUOTA_DQACQ:
258                 /* build resource name associated for per-ID quota lock */
259                 fid_build_quota_res_name(&qbody->qb_fid, &qbody->qb_id,
260                                          &qti->qti_resid);
261
262                 /* copy einfo template and fill ei_cbdata with lqe pointer */
263                 memcpy(&qti->qti_einfo, &qsd_id_einfo, sizeof(qti->qti_einfo));
264                 qti->qti_einfo.ei_cbdata = arg;
265                 break;
266         default:
267                 LASSERTF(0, "invalid it_op %d\n", it_op);
268         }
269
270         /* build lock enqueue request */
271         rc = ldlm_cli_enqueue(exp, &req, &qti->qti_einfo, &qti->qti_resid, NULL,
272                               &flags, (void *)lvb, sizeof(*lvb), LVB_T_LQUOTA,
273                               &qti->qti_lockh, 1);
274         if (rc < 0) {
275                 ptlrpc_req_finished(req);
276                 GOTO(out, rc);
277         }
278
279         /* grab reference on backend structure for the new lock */
280         switch(it_op) {
281         case IT_QUOTA_CONN:
282                 /* grab reference on qqi for new lock */
283 #ifdef USE_LU_REF
284         {
285                 struct ldlm_lock        *lock;
286
287                 lock = ldlm_handle2lock(&qti->qti_lockh);
288                 if (lock == NULL) {
289                         ptlrpc_req_finished(req);
290                         GOTO(out, rc = -ENOLCK);
291                 }
292                 lu_ref_add(&qqi->qqi_reference, "glb_lock", lock);
293                 LDLM_LOCK_PUT(lock);
294         }
295 #endif
296                 qqi_getref(qqi);
297                 break;
298         case IT_QUOTA_DQACQ:
299                 /* grab reference on lqe for new lock */
300                 lqe_getref((struct lquota_entry *)arg);
301                 /* all acquire/release request are sent with no_resend and
302                  * no_delay flag */
303                 req->rq_no_resend = req->rq_no_delay = 1;
304                 break;
305         default:
306                 break;
307         }
308
309         CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
310         aa = ptlrpc_req_async_args(req);
311         aa->aa_exp = exp;
312         aa->aa_qqi = qqi;
313         aa->aa_arg = arg;
314         aa->aa_lvb = lvb;
315         aa->aa_completion = completion;
316         lustre_handle_copy(&aa->aa_lockh, &qti->qti_lockh);
317
318         if (sync) {
319                 /* send lock enqueue request and wait for completion */
320                 rc = ptlrpc_queue_wait(req);
321                 rc = qsd_intent_interpret(env, req, aa, rc);
322                 ptlrpc_req_finished(req);
323         } else {
324                 /* queue lock request and return */
325                 req->rq_interpret_reply = qsd_intent_interpret;
326                 ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
327         }
328
329         RETURN(rc);
330 out:
331         completion(env, qqi, qbody, NULL, &qti->qti_lockh, lvb, arg, rc);
332         return rc;
333 }
334
335 /*
336  * Fetch a global or slave index from the QMT.
337  *
338  * \param env    - the environment passed by the caller
339  * \param exp    - is the export to use to issue the OBD_IDX_READ RPC
340  * \param ii     - is the index information to be packed in the request
341  *                 on success, the index information returned by the server
342  *                 is copied there.
343  * \param npages - is the number of pages in the pages array
344  * \param pages  - is an array of @npages pages
345  *
346  * \retval 0     - success
347  * \retval -ve   - appropriate errors
348  */
349 int qsd_fetch_index(const struct lu_env *env, struct obd_export *exp,
350                     struct idx_info *ii, unsigned int npages,
351                     struct page **pages, bool *need_swab)
352 {
353         struct ptlrpc_request   *req;
354         struct idx_info         *req_ii;
355         struct ptlrpc_bulk_desc *desc;
356         int                      rc, i;
357         ENTRY;
358
359         LASSERT(exp);
360
361         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_OBD_IDX_READ);
362         if (req == NULL)
363                 RETURN(-ENOMEM);
364
365         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
366         if (rc) {
367                 ptlrpc_request_free(req);
368                 RETURN(rc);
369         }
370
371         req->rq_request_portal = MDS_READPAGE_PORTAL;
372         ptlrpc_at_set_req_timeout(req);
373
374         /* allocate bulk descriptor */
375         desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
376                                     MDS_BULK_PORTAL);
377         if (desc == NULL) {
378                 ptlrpc_request_free(req);
379                 RETURN(-ENOMEM);
380         }
381
382         /* req now owns desc and will free it when it gets freed */
383         for (i = 0; i < npages; i++)
384                 ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
385
386         /* pack index information in request */
387         req_ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
388         *req_ii = *ii;
389
390         ptlrpc_request_set_replen(req);
391
392         /* send request to master and wait for RPC to complete */
393         rc = ptlrpc_queue_wait(req);
394         if (rc)
395                 GOTO(out, rc);
396
397         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
398                                           req->rq_bulk->bd_nob_transferred);
399         if (rc < 0)
400                 GOTO(out, rc);
401         else
402                 /* sptlrpc_cli_unwrap_bulk_read() returns the number of bytes
403                  * transferred*/
404                 rc = 0;
405
406         req_ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
407         *ii = *req_ii;
408
409         *need_swab = ptlrpc_rep_need_swab(req);
410
411         EXIT;
412 out:
413         ptlrpc_req_finished(req);
414         return rc;
415 }