Whamcloud - gitweb
LU-8066 obd: collect all resource releasing for obj_type.
[fs/lustre-release.git] / lustre / osp / osp_object.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, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2012, 2017, Intel Corporation.
27  */
28 /*
29  * lustre/osp/osp_object.c
30  *
31  * Lustre OST Proxy Device (OSP) is the agent on the local MDT for the OST
32  * or remote MDT.
33  *
34  * OSP object attributes cache
35  * ---------------------------
36  * OSP object is the stub of the remote OST-object or MDT-object. Both the
37  * attribute and the extended attributes are stored on the peer side remotely.
38  * It is inefficient to send RPC to peer to fetch those attributes when every
39  * get_attr()/get_xattr() called. For a large system, the LFSCK synchronous
40  * mode scanning is prohibitively inefficient.
41  *
42  * So the OSP maintains the OSP object attributes cache to cache some
43  * attributes on the local MDT. The cache is organized against the OSP
44  * object as follows:
45  *
46  * struct osp_xattr_entry {
47  *      struct list_head         oxe_list;
48  *      atomic_t                 oxe_ref;
49  *      void                    *oxe_value;
50  *      int                      oxe_buflen;
51  *      int                      oxe_namelen;
52  *      int                      oxe_vallen;
53  *      unsigned int             oxe_exist:1,
54  *                               oxe_ready:1;
55  *      char                     oxe_buf[0];
56  * };
57  *
58  * struct osp_object {
59  *      ...
60  *      struct lu_attr          opo_attr;
61  *      struct list_head        opo_xattr_list;
62  *      spinlock_t              opo_lock;
63  *      ...
64  * };
65  *
66  * The basic attributes, such as owner/mode/flags, are stored in the
67  * osp_object::opo_attr. The extended attributes will be stored
68  * as osp_xattr_entry. Every extended attribute has an independent
69  * osp_xattr_entry, and all the osp_xattr_entry are linked into the
70  * osp_object::opo_xattr_list. The OSP object attributes cache
71  * is protected by the osp_object::opo_lock.
72  *
73  * Not all OSP objects have an attributes cache because maintaining
74  * the cache requires some resources. Currently, the OSP object
75  * attributes cache will be initialized when the attributes or the
76  * extended attributes are pre-fetched via osp_declare_attr_get()
77  * or osp_declare_xattr_get(). That is usually for LFSCK purpose,
78  * but it also can be shared by others.
79  *
80  *
81  * XXX: NOT prepare out RPC for remote transaction. ((please refer to the
82  *      comment of osp_trans_create() for remote transaction)
83  *
84  * According to our current transaction/dt_object_lock framework (to make
85  * the cross-MDTs modification for DNE1 to be workable), the transaction
86  * sponsor will start the transaction firstly, then try to acquire related
87  * dt_object_lock if needed. Under such rules, if we want to prepare the
88  * OUT RPC in the transaction declare phase, then related attr/xattr
89  * should be known without dt_object_lock. But such condition maybe not
90  * true for some remote transaction case. For example:
91  *
92  * For linkEA repairing (by LFSCK) case, before the LFSCK thread obtained
93  * the dt_object_lock on the target MDT-object, it cannot know whether
94  * the MDT-object has linkEA or not, neither invalid or not.
95  *
96  * Since the LFSCK thread cannot hold dt_object_lock before the remote
97  * transaction start (otherwise there will be some potential deadlock),
98  * it cannot prepare related OUT RPC for repairing during the declare
99  * phase as other normal transactions do.
100  *
101  * To resolve the trouble, we will make OSP to prepare related OUT RPC
102  * after remote transaction started, and trigger the remote updating
103  * (send RPC) when trans_stop. Then the up layer users, such as LFSCK,
104  * can follow the general rule to handle trans_start/dt_object_lock
105  * for repairing linkEA inconsistency without distinguishing remote
106  * MDT-object.
107  *
108  * In fact, above solution for remote transaction should be the normal
109  * model without considering DNE1. The trouble brought by DNE1 will be
110  * resolved in DNE2. At that time, this patch can be removed.
111  *
112  *
113  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
114  * Author: Mikhail Pershin <mike.tappro@intel.com>
115  */
116
117 #define DEBUG_SUBSYSTEM S_MDS
118
119 #include <lustre_obdo.h>
120 #include <lustre_swab.h>
121
122 #include "osp_internal.h"
123
124 static inline __u32 osp_dev2node(struct osp_device *osp)
125 {
126         return osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
127 }
128
129 static inline const char *osp_dto2name(struct osp_object *obj)
130 {
131         return obj->opo_obj.do_lu.lo_dev->ld_obd->obd_name;
132 }
133
134 static inline bool is_ost_obj(struct lu_object *lo)
135 {
136         return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt;
137 }
138
139 static inline void __osp_oac_xattr_assignment(struct osp_object *obj,
140                                               struct osp_xattr_entry *oxe,
141                                               const struct lu_buf *buf)
142 {
143         if (buf->lb_len > 0)
144                 memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
145
146         oxe->oxe_vallen = buf->lb_len;
147         oxe->oxe_exist = 1;
148         oxe->oxe_ready = 1;
149 }
150
151 /**
152  * Assign FID to the OST object.
153  *
154  * This function will assign the FID to the OST object of a striped file.
155  *
156  * \param[in] env       pointer to the thread context
157  * \param[in] d         pointer to the OSP device
158  * \param[in] o         pointer to the OSP object that the FID will be
159  *                      assigned to
160  */
161 static void osp_object_assign_fid(const struct lu_env *env,
162                                   struct osp_device *d, struct osp_object *o)
163 {
164         struct osp_thread_info *osi = osp_env_info(env);
165
166         LASSERT(fid_is_zero(lu_object_fid(&o->opo_obj.do_lu)));
167         LASSERT(o->opo_reserved);
168         o->opo_reserved = 0;
169
170         osp_precreate_get_fid(env, d, &osi->osi_fid);
171
172         lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid);
173 }
174
175 #define OXE_DEFAULT_LEN 16
176
177 /**
178  * Release reference from the OSP object extended attribute entry.
179  *
180  * If it is the last reference, then free the entry.
181  *
182  * \param[in] oxe       pointer to the OSP object extended attribute entry.
183  */
184 static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe)
185 {
186         if (atomic_dec_and_test(&oxe->oxe_ref)) {
187                 LASSERT(list_empty(&oxe->oxe_list));
188
189                 OBD_FREE(oxe, oxe->oxe_buflen);
190         }
191 }
192
193 /**
194  * Find the named extended attribute in the OSP object attributes cache.
195  *
196  * The caller should take the osp_object::opo_lock before calling
197  * this function.
198  *
199  * \param[in] obj       pointer to the OSP object
200  * \param[in] name      the name of the extended attribute
201  * \param[in] namelen   the name length of the extended attribute
202  *
203  * \retval              pointer to the found extended attribute entry
204  * \retval              NULL if the specified extended attribute is not
205  *                      in the cache
206  */
207 static struct osp_xattr_entry *
208 osp_oac_xattr_find_locked(struct osp_object *obj, const char *name,
209                           size_t namelen)
210 {
211         struct osp_xattr_entry *oxe;
212
213         list_for_each_entry(oxe, &obj->opo_xattr_list, oxe_list) {
214                 if (namelen == oxe->oxe_namelen &&
215                     strncmp(name, oxe->oxe_buf, namelen) == 0)
216                         return oxe;
217         }
218
219         return NULL;
220 }
221
222 /**
223  * Find the named extended attribute in the OSP object attributes cache.
224  *
225  * Call osp_oac_xattr_find_locked() with the osp_object::opo_lock held.
226  *
227  * \param[in] obj       pointer to the OSP object
228  * \param[in] name      the name of the extended attribute
229  * \param[in] unlink    true if the extended attribute entry is to be removed
230  *                      from the cache
231  *
232  * \retval              pointer to the found extended attribute entry
233  * \retval              NULL if the specified extended attribute is not
234  *                      in the cache
235  */
236 static struct osp_xattr_entry *osp_oac_xattr_find(struct osp_object *obj,
237                                                   const char *name, bool unlink)
238 {
239         struct osp_xattr_entry *oxe = NULL;
240
241         spin_lock(&obj->opo_lock);
242         oxe = osp_oac_xattr_find_locked(obj, name, strlen(name));
243         if (oxe) {
244                 if (unlink)
245                         list_del_init(&oxe->oxe_list);
246                 else
247                         atomic_inc(&oxe->oxe_ref);
248         }
249         spin_unlock(&obj->opo_lock);
250
251         return oxe;
252 }
253
254 /**
255  * Find the named extended attribute in the OSP object attributes cache.
256  *
257  * If it is not in the cache, then add an empty entry (that will be
258  * filled later) to cache with the given name.
259  *
260  * \param[in] obj       pointer to the OSP object
261  * \param[in] name      the name of the extended attribute
262  * \param[in] len       the length of the extended attribute value
263  *
264  * \retval              pointer to the found or new-created extended
265  *                      attribute entry
266  * \retval              NULL if the specified extended attribute is not in the
267  *                      cache or fail to add new empty entry to the cache.
268  */
269 static struct osp_xattr_entry *
270 osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
271 {
272         struct osp_xattr_entry *oxe;
273         struct osp_xattr_entry *tmp = NULL;
274         size_t namelen = strlen(name);
275         size_t size = sizeof(*oxe) + namelen + 1 +
276                       (len ? len : OXE_DEFAULT_LEN);
277
278         oxe = osp_oac_xattr_find(obj, name, false);
279         if (oxe)
280                 return oxe;
281
282         OBD_ALLOC(oxe, size);
283         if (unlikely(!oxe))
284                 return NULL;
285
286         INIT_LIST_HEAD(&oxe->oxe_list);
287         oxe->oxe_buflen = size;
288         oxe->oxe_namelen = namelen;
289         memcpy(oxe->oxe_buf, name, namelen);
290         oxe->oxe_value = oxe->oxe_buf + namelen + 1;
291         /* One ref is for the caller, the other is for the entry on the list. */
292         atomic_set(&oxe->oxe_ref, 2);
293
294         spin_lock(&obj->opo_lock);
295         tmp = osp_oac_xattr_find_locked(obj, name, namelen);
296         if (!tmp)
297                 list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list);
298         else
299                 atomic_inc(&tmp->oxe_ref);
300         spin_unlock(&obj->opo_lock);
301
302         if (tmp) {
303                 OBD_FREE(oxe, size);
304                 oxe = tmp;
305         }
306
307         return oxe;
308 }
309
310 /**
311  * Assign the cached OST-object's EA with the given value.
312  *
313  * If the current EA entry in cache has not enough space to hold the new
314  * value, remove it, create a new one, then assign with the given value.
315  *
316  * \param[in] obj       pointer to the OSP object
317  * \param[in] oxe       pointer to the cached EA entry to be assigned
318  * \param[in] buf       pointer to the buffer with new EA value
319  *
320  * \retval              pointer to the new created EA entry in cache if
321  *                      current entry is not big enough; otherwise, the
322  *                      input 'oxe' will be returned.
323  */
324 static struct osp_xattr_entry *
325 osp_oac_xattr_assignment(struct osp_object *obj, struct osp_xattr_entry *oxe,
326                          const struct lu_buf *buf)
327 {
328         struct osp_xattr_entry *new = NULL;
329         struct osp_xattr_entry *old = NULL;
330         int namelen = oxe->oxe_namelen;
331         size_t size = sizeof(*oxe) + namelen + 1 + buf->lb_len;
332         bool unlink_only = false;
333
334         if (oxe->oxe_buflen < size) {
335                 OBD_ALLOC(new, size);
336                 if (likely(new)) {
337                         INIT_LIST_HEAD(&new->oxe_list);
338                         new->oxe_buflen = size;
339                         new->oxe_namelen = namelen;
340                         memcpy(new->oxe_buf, oxe->oxe_buf, namelen);
341                         new->oxe_value = new->oxe_buf + namelen + 1;
342                         /* One ref is for the caller,
343                          * the other is for the entry on the list. */
344                         atomic_set(&new->oxe_ref, 2);
345                         __osp_oac_xattr_assignment(obj, new, buf);
346                 } else {
347                         unlink_only = true;
348                         CWARN("%s: cannot update cached xattr %.*s of "DFID"\n",
349                               osp_dto2name(obj), namelen, oxe->oxe_buf,
350                               PFID(lu_object_fid(&obj->opo_obj.do_lu)));
351                 }
352         }
353
354         spin_lock(&obj->opo_lock);
355         old = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen);
356         if (likely(old)) {
357                 if (new) {
358                         /* Unlink the 'old'. */
359                         list_del_init(&old->oxe_list);
360
361                         /* Drop the ref for 'old' on list. */
362                         osp_oac_xattr_put(old);
363
364                         /* Drop the ref for current using. */
365                         osp_oac_xattr_put(oxe);
366                         oxe = new;
367
368                         /* Insert 'new' into list. */
369                         list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
370                 } else if (unlink_only) {
371                         /* Unlink the 'old'. */
372                         list_del_init(&old->oxe_list);
373
374                         /* Drop the ref for 'old' on list. */
375                         osp_oac_xattr_put(old);
376                 } else {
377                         __osp_oac_xattr_assignment(obj, oxe, buf);
378                 }
379         } else if (new) {
380                 /* Drop the ref for current using. */
381                 osp_oac_xattr_put(oxe);
382                 oxe = new;
383
384                 /* Someone unlinked the 'old' by race,
385                  * insert the 'new' one into list. */
386                 list_add_tail(&new->oxe_list, &obj->opo_xattr_list);
387         }
388         spin_unlock(&obj->opo_lock);
389
390         return oxe;
391 }
392
393 /**
394  * Parse the OSP object attribute from the RPC reply.
395  *
396  * If the attribute is valid, then it will be added to the OSP object
397  * attributes cache.
398  *
399  * \param[in] env       pointer to the thread context
400  * \param[in] reply     pointer to the RPC reply
401  * \param[in] req       pointer to the RPC request
402  * \param[out] attr     pointer to buffer to hold the output attribute
403  * \param[in] obj       pointer to the OSP object
404  * \param[in] index     the index of the attribute buffer in the reply
405  *
406  * \retval              0 for success
407  * \retval              negative error number on failure
408  */
409 static int osp_get_attr_from_reply(const struct lu_env *env,
410                                    struct object_update_reply *reply,
411                                    struct ptlrpc_request *req,
412                                    struct lu_attr *attr,
413                                    struct osp_object *obj, int index)
414 {
415         struct osp_thread_info  *osi    = osp_env_info(env);
416         struct lu_buf           *rbuf   = &osi->osi_lb2;
417         struct obdo             *lobdo  = &osi->osi_obdo;
418         struct obdo             *wobdo;
419         int                     rc;
420
421         rc = object_update_result_data_get(reply, rbuf, index);
422         if (rc < 0)
423                 return rc;
424
425         wobdo = rbuf->lb_buf;
426         if (rbuf->lb_len != sizeof(*wobdo))
427                 return -EPROTO;
428
429         LASSERT(req != NULL);
430         if (ptlrpc_req_need_swab(req))
431                 lustre_swab_obdo(wobdo);
432
433         lustre_get_wire_obdo(NULL, lobdo, wobdo);
434         if (obj) {
435                 spin_lock(&obj->opo_lock);
436                 la_from_obdo(&obj->opo_attr, lobdo, lobdo->o_valid);
437                 spin_unlock(&obj->opo_lock);
438         }
439         if (attr)
440                 la_from_obdo(attr, lobdo, lobdo->o_valid);
441
442         return 0;
443 }
444
445 /**
446  * Interpreter function for getting OSP object attribute asynchronously.
447  *
448  * Called to interpret the result of an async mode RPC for getting the
449  * OSP object attribute.
450  *
451  * \param[in] env       pointer to the thread context
452  * \param[in] reply     pointer to the RPC reply
453  * \param[in] req       pointer to the RPC request
454  * \param[in] obj       pointer to the OSP object
455  * \param[out] data     pointer to buffer to hold the output attribute
456  * \param[in] index     the index of the attribute buffer in the reply
457  * \param[in] rc        the result for handling the RPC
458  *
459  * \retval              0 for success
460  * \retval              negative error number on failure
461  */
462 static int osp_attr_get_interpterer(const struct lu_env *env,
463                                     struct object_update_reply *reply,
464                                     struct ptlrpc_request *req,
465                                     struct osp_object *obj,
466                                     void *data, int index, int rc)
467 {
468         struct lu_attr *attr = data;
469
470         if (rc == 0) {
471                 osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
472                 obj->opo_non_exist = 0;
473
474                 return osp_get_attr_from_reply(env, reply, req, NULL, obj,
475                                                index);
476         } else {
477                 if (rc == -ENOENT) {
478                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
479                         obj->opo_non_exist = 1;
480                 }
481
482                 spin_lock(&obj->opo_lock);
483                 attr->la_valid = 0;
484                 spin_unlock(&obj->opo_lock);
485         }
486
487         return 0;
488 }
489
490 /**
491  * Implement OSP layer dt_object_operations::do_declare_attr_get() interface.
492  *
493  * Declare that the caller will get attribute from the specified OST object.
494  *
495  * This function adds an Object Unified Target (OUT) sub-request to the per-OSP
496  * based shared asynchronous request queue. The osp_attr_get_interpterer()
497  * is registered as the interpreter function to handle the result of this
498  * sub-request.
499  *
500  * \param[in] env       pointer to the thread context
501  * \param[in] dt        pointer to the OSP layer dt_object
502  *
503  * \retval              0 for success
504  * \retval              negative error number on failure
505  */
506 static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt)
507 {
508         struct osp_object       *obj    = dt2osp_obj(dt);
509         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
510         int                      rc     = 0;
511
512         mutex_lock(&osp->opd_async_requests_mutex);
513         rc = osp_insert_async_request(env, OUT_ATTR_GET, obj, 0, NULL, NULL,
514                                       &obj->opo_attr, sizeof(struct obdo),
515                                       osp_attr_get_interpterer);
516         mutex_unlock(&osp->opd_async_requests_mutex);
517
518         return rc;
519 }
520
521 /**
522  * Implement OSP layer dt_object_operations::do_attr_get() interface.
523  *
524  * Get attribute from the specified MDT/OST object.
525  *
526  * If the attribute is in the OSP object attributes cache, then return
527  * the cached attribute directly. Otherwise it will trigger an OUT RPC
528  * to the peer to get the attribute synchronously, if successful, add it
529  * to the OSP attributes cache. (\see lustre/osp/osp_trans.c for OUT RPC.)
530  *
531  * \param[in] env       pointer to the thread context
532  * \param[in] dt        pointer to the OSP layer dt_object
533  * \param[out] attr     pointer to the buffer to hold the output attribute
534  *
535  * \retval              0 for success
536  * \retval              negative error number on failure
537  */
538 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
539                  struct lu_attr *attr)
540 {
541         struct osp_device               *osp = lu2osp_dev(dt->do_lu.lo_dev);
542         struct osp_object               *obj = dt2osp_obj(dt);
543         struct dt_device                *dev = &osp->opd_dt_dev;
544         struct osp_update_request       *update;
545         struct object_update_reply      *reply;
546         struct ptlrpc_request           *req = NULL;
547         int                             invalidated, cache = 0, rc = 0;
548         ENTRY;
549
550         if (is_ost_obj(&dt->do_lu) && obj->opo_non_exist)
551                 RETURN(-ENOENT);
552
553         spin_lock(&obj->opo_lock);
554         if (obj->opo_attr.la_valid != 0 && !obj->opo_stale) {
555                 *attr = obj->opo_attr;
556                 spin_unlock(&obj->opo_lock);
557
558                 RETURN(0);
559         }
560         spin_unlock(&obj->opo_lock);
561
562         update = osp_update_request_create(dev);
563         if (IS_ERR(update))
564                 RETURN(PTR_ERR(update));
565
566         rc = OSP_UPDATE_RPC_PACK(env, out_attr_get_pack, update,
567                                  lu_object_fid(&dt->do_lu));
568         if (rc != 0) {
569                 CERROR("%s: Insert update error "DFID": rc = %d\n",
570                        dev->dd_lu_dev.ld_obd->obd_name,
571                        PFID(lu_object_fid(&dt->do_lu)), rc);
572
573                 GOTO(out_req, rc);
574         }
575
576         invalidated = atomic_read(&obj->opo_invalidate_seq);
577
578         rc = osp_remote_sync(env, osp, update, &req);
579
580         down_read(&obj->opo_invalidate_sem);
581         if (invalidated == atomic_read(&obj->opo_invalidate_seq)) {
582                 /* no invalited has came so far, we can cache the attrs */
583                 cache = 1;
584         }
585
586         if (rc != 0) {
587                 if (rc == -ENOENT) {
588                         osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
589                         if (cache)
590                                 obj->opo_non_exist = 1;
591                 } else {
592                         CERROR("%s:osp_attr_get update error "DFID": rc = %d\n",
593                                dev->dd_lu_dev.ld_obd->obd_name,
594                                PFID(lu_object_fid(&dt->do_lu)), rc);
595                 }
596
597                 GOTO(out, rc);
598         }
599
600         osp2lu_obj(obj)->lo_header->loh_attr |= LOHA_EXISTS;
601         obj->opo_non_exist = 0;
602         reply = req_capsule_server_sized_get(&req->rq_pill,
603                                              &RMF_OUT_UPDATE_REPLY,
604                                              OUT_UPDATE_REPLY_SIZE);
605         if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
606                 GOTO(out, rc = -EPROTO);
607
608         rc = osp_get_attr_from_reply(env, reply, req, attr,
609                                      cache ? obj : NULL, 0);
610         if (rc != 0)
611                 GOTO(out, rc);
612
613         spin_lock(&obj->opo_lock);
614         if (cache)
615                 obj->opo_stale = 0;
616         spin_unlock(&obj->opo_lock);
617
618         GOTO(out, rc);
619
620 out:
621         up_read(&obj->opo_invalidate_sem);
622
623 out_req:
624         if (req != NULL)
625                 ptlrpc_req_finished(req);
626
627         osp_update_request_destroy(env, update);
628
629         return rc;
630 }
631
632 /**
633  * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
634  *
635  * If the transaction is not remote one, then declare the credits that will
636  * be used for the subsequent llog record for the object's attributes.
637  *
638  * \param[in] env       pointer to the thread context
639  * \param[in] dt        pointer to the OSP layer dt_object
640  * \param[in] attr      pointer to the attribute to be set
641  * \param[in] th        pointer to the transaction handler
642  *
643  * \retval              0 for success
644  * \retval              negative error number on failure
645  */
646 static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
647                                 const struct lu_attr *attr, struct thandle *th)
648 {
649         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
650         struct osp_object       *o = dt2osp_obj(dt);
651         int                     rc;
652
653         if (is_only_remote_trans(th))
654                 return osp_md_declare_attr_set(env, dt, attr, th);
655         /*
656          * Usually we don't allow server stack to manipulate size
657          * but there is a special case when striping is created
658          * late, after stripeless file got truncated to non-zero.
659          *
660          * In this case we do the following:
661          *
662          * 1) grab id in declare - this can lead to leaked OST objects
663          *    but we don't currently have proper mechanism and the only
664          *    options we have are to do truncate RPC holding transaction
665          *    open (very bad) or to grab id in declare at cost of leaked
666          *    OST object in same very rare unfortunate case (just bad)
667          *    notice 1.6-2.0 do assignment outside of running transaction
668          *    all the time, meaning many more chances for leaked objects.
669          *
670          * 2) send synchronous truncate RPC with just assigned id
671          */
672
673         /* there are few places in MDD code still passing NULL
674          * XXX: to be fixed soon */
675         if (attr == NULL)
676                 RETURN(0);
677
678         if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
679             fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
680                 LASSERT(!dt_object_exists(dt));
681                 osp_object_assign_fid(env, d, o);
682                 rc = osp_object_truncate(env, dt, attr->la_size);
683                 if (rc != 0)
684                         RETURN(rc);
685         }
686
687         if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
688                 RETURN(0);
689
690         /* track all UID/GID, projid, and layout version changes via llog */
691         rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
692
693         return 0;
694 }
695
696 /**
697  * Implement OSP layer dt_object_operations::do_attr_set() interface.
698  *
699  * Set attribute to the specified OST object.
700  *
701  * If the transaction is a remote one, then add OUT_ATTR_SET sub-request
702  * in the OUT RPC that will be flushed when the remote transaction stop.
703  * Otherwise, it will generate a MDS_SETATTR64_REC record in the llog that
704  * will be handled by a dedicated thread asynchronously.
705  *
706  * If the attribute entry exists in the OSP object attributes cache,
707  * then update the cached attribute according to given attribute.
708  *
709  * \param[in] env       pointer to the thread context
710  * \param[in] dt        pointer to the OSP layer dt_object
711  * \param[in] attr      pointer to the attribute to be set
712  * \param[in] th        pointer to the transaction handler
713  *
714  * \retval              0 for success
715  * \retval              negative error number on failure
716  */
717 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
718                         const struct lu_attr *attr, struct thandle *th)
719 {
720         struct osp_object       *o = dt2osp_obj(dt);
721         int                      rc = 0;
722         ENTRY;
723
724         /* we're interested in uid/gid/projid/layout version changes only */
725         if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
726                 RETURN(0);
727
728         if (!is_only_remote_trans(th)) {
729                 if (attr->la_flags & LUSTRE_SET_SYNC_FL) {
730                         struct ptlrpc_request *req = NULL;
731                         struct osp_update_request *update = NULL;
732                         struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
733
734                         update = osp_update_request_create(&osp->opd_dt_dev);
735                         if (IS_ERR(update))
736                                 RETURN(PTR_ERR(update));
737
738                         rc = OSP_UPDATE_RPC_PACK(env, out_attr_set_pack, update,
739                                                  lu_object_fid(&dt->do_lu),
740                                                  attr);
741                         if (rc != 0) {
742                                 CERROR("%s: update error "DFID": rc = %d\n",
743                                        osp->opd_obd->obd_name,
744                                        PFID(lu_object_fid(&dt->do_lu)), rc);
745
746                                 osp_update_request_destroy(env, update);
747                                 RETURN(rc);
748                         }
749
750                         rc = osp_remote_sync(env, osp, update, &req);
751                         if (req != NULL)
752                                 ptlrpc_req_finished(req);
753
754                         osp_update_request_destroy(env, update);
755                 } else {
756                         rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
757                         /* XXX: send new uid/gid to OST ASAP? */
758                 }
759         } else {
760                 struct lu_attr  *la;
761
762                 /* It is for OST-object attr_set directly without updating
763                  * local MDT-object attribute. It is usually used by LFSCK. */
764                 rc = osp_md_attr_set(env, dt, attr, th);
765                 CDEBUG(D_INFO, "(1) set attr "DFID": rc = %d\n",
766                        PFID(&dt->do_lu.lo_header->loh_fid), rc);
767
768                 if (rc != 0)
769                         RETURN(rc);
770
771                 /* Update the OSP object attributes cache. */
772                 la = &o->opo_attr;
773                 spin_lock(&o->opo_lock);
774                 if (attr->la_valid & LA_UID) {
775                         la->la_uid = attr->la_uid;
776                         la->la_valid |= LA_UID;
777                 }
778
779                 if (attr->la_valid & LA_GID) {
780                         la->la_gid = attr->la_gid;
781                         la->la_valid |= LA_GID;
782                 }
783                 if (attr->la_valid & LA_PROJID) {
784                         la->la_projid = attr->la_projid;
785                         la->la_valid |= LA_PROJID;
786                 }
787                 spin_unlock(&o->opo_lock);
788         }
789
790         RETURN(rc);
791 }
792
793 /**
794  * Interpreter function for getting OSP object extended attribute asynchronously
795  *
796  * Called to interpret the result of an async mode RPC for getting the
797  * OSP object extended attribute.
798  *
799  * \param[in] env       pointer to the thread context
800  * \param[in] reply     pointer to the RPC reply
801  * \param[in] req       pointer to the RPC request
802  * \param[in] obj       pointer to the OSP object
803  * \param[out] data     pointer to OSP object attributes cache
804  * \param[in] index     the index of the attribute buffer in the reply
805  * \param[in] rc        the result for handling the RPC
806  *
807  * \retval              0 for success
808  * \retval              negative error number on failure
809  */
810 static int osp_xattr_get_interpterer(const struct lu_env *env,
811                                      struct object_update_reply *reply,
812                                      struct ptlrpc_request *req,
813                                      struct osp_object *obj,
814                                      void *data, int index, int rc)
815 {
816         struct osp_xattr_entry *oxe = data;
817
818         spin_lock(&obj->opo_lock);
819         if (rc >= 0) {
820                 struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2;
821                 size_t len = sizeof(*oxe) + oxe->oxe_namelen + 1;
822
823                 rc = object_update_result_data_get(reply, rbuf, index);
824                 if (rc == -ENOENT || rc == -ENODATA || rc == 0) {
825                         oxe->oxe_exist = 0;
826                         oxe->oxe_ready = 1;
827                         goto unlock;
828                 }
829
830                 if (unlikely(rc < 0) ||
831                     rbuf->lb_len > (oxe->oxe_buflen - len)) {
832                         oxe->oxe_ready = 0;
833                         goto unlock;
834                 }
835
836                 __osp_oac_xattr_assignment(obj, oxe, rbuf);
837         } else if (rc == -ENOENT || rc == -ENODATA) {
838                 oxe->oxe_exist = 0;
839                 oxe->oxe_ready = 1;
840         } else {
841                 oxe->oxe_ready = 0;
842         }
843
844 unlock:
845         spin_unlock(&obj->opo_lock);
846
847         /* Put the reference obtained in the osp_declare_xattr_get(). */
848         osp_oac_xattr_put(oxe);
849
850         return 0;
851 }
852
853 /**
854  * Implement OSP dt_object_operations::do_declare_xattr_get() interface.
855  *
856  * Declare that the caller will get extended attribute from the specified
857  * OST object.
858  *
859  * This function will add an OUT_XATTR_GET sub-request to the per OSP
860  * based shared asynchronous request queue with the interpreter function:
861  * osp_xattr_get_interpterer().
862  *
863  * \param[in] env       pointer to the thread context
864  * \param[in] dt        pointer to the OSP layer dt_object
865  * \param[out] buf      pointer to the lu_buf to hold the extended attribute
866  * \param[in] name      the name for the expected extended attribute
867  *
868  * \retval              0 for success
869  * \retval              negative error number on failure
870  */
871 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
872                                  struct lu_buf *buf, const char *name)
873 {
874         struct osp_object       *obj     = dt2osp_obj(dt);
875         struct osp_device       *osp     = lu2osp_dev(dt->do_lu.lo_dev);
876         struct osp_xattr_entry  *oxe;
877         int                      rc      = 0;
878         __u16 len;
879
880         LASSERT(buf != NULL);
881         LASSERT(name != NULL);
882
883         if (unlikely(buf->lb_len == 0))
884                 return -EINVAL;
885
886         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
887         if (oxe == NULL)
888                 return -ENOMEM;
889
890         len = strlen(name) + 1;
891         mutex_lock(&osp->opd_async_requests_mutex);
892         rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
893                                       &len, (const void **)&name,
894                                       oxe, buf->lb_len,
895                                       osp_xattr_get_interpterer);
896         if (rc != 0) {
897                 mutex_unlock(&osp->opd_async_requests_mutex);
898                 osp_oac_xattr_put(oxe);
899         } else {
900                 struct osp_update_request *our;
901                 struct osp_update_request_sub *ours;
902
903                 /* XXX: Currently, we trigger the batched async OUT
904                  *      RPC via dt_declare_xattr_get(). It is not
905                  *      perfect solution, but works well now.
906                  *
907                  *      We will improve it in the future. */
908                 our = osp->opd_async_requests;
909                 ours = osp_current_object_update_request(our);
910                 if (ours != NULL && ours->ours_req != NULL &&
911                     ours->ours_req->ourq_count > 0) {
912                         osp->opd_async_requests = NULL;
913                         mutex_unlock(&osp->opd_async_requests_mutex);
914                         rc = osp_unplug_async_request(env, osp, our);
915                 } else {
916                         mutex_unlock(&osp->opd_async_requests_mutex);
917                 }
918         }
919
920         return rc;
921 }
922
923 /**
924  * Implement OSP layer dt_object_operations::do_xattr_get() interface.
925  *
926  * Get extended attribute from the specified MDT/OST object.
927  *
928  * If the extended attribute is in the OSP object attributes cache, then
929  * return the cached extended attribute directly. Otherwise it will get
930  * the extended attribute synchronously, if successful, add it to the OSP
931  * attributes cache. (\see lustre/osp/osp_trans.c for OUT RPC.)
932  *
933  * There is a race condition: some other thread has added the named extended
934  * attributed entry to the OSP object attributes cache during the current
935  * OUT_XATTR_GET handling. If such case happens, the OSP will replace the
936  * (just) existing extended attribute entry with the new replied one.
937  *
938  * \param[in] env       pointer to the thread context
939  * \param[in] dt        pointer to the OSP layer dt_object
940  * \param[out] buf      pointer to the lu_buf to hold the extended attribute
941  * \param[in] name      the name for the expected extended attribute
942  *
943  * \retval              0 for success
944  * \retval              negative error number on failure
945  */
946 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
947                   struct lu_buf *buf, const char *name)
948 {
949         struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
950         struct osp_object       *obj    = dt2osp_obj(dt);
951         struct dt_device        *dev    = &osp->opd_dt_dev;
952         struct lu_buf           *rbuf   = &osp_env_info(env)->osi_lb2;
953         struct osp_update_request *update = NULL;
954         struct ptlrpc_request   *req    = NULL;
955         struct object_update_reply *reply;
956         struct osp_xattr_entry  *oxe    = NULL;
957         const char *dname = osp_dto2name(obj);
958         int invalidated, rc = 0;
959         ENTRY;
960
961         LASSERT(buf != NULL);
962         LASSERT(name != NULL);
963
964         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_NETWORK) &&
965             osp->opd_index == cfs_fail_val) {
966                 if (is_ost_obj(&dt->do_lu)) {
967                         if (osp_dev2node(osp) == cfs_fail_val)
968                                 RETURN(-ENOTCONN);
969                 } else {
970                         if (strcmp(name, XATTR_NAME_LINK) == 0)
971                                 RETURN(-ENOTCONN);
972                 }
973         }
974
975         if (unlikely(obj->opo_non_exist))
976                 RETURN(-ENOENT);
977
978         invalidated = atomic_read(&obj->opo_invalidate_seq);
979
980         oxe = osp_oac_xattr_find(obj, name, false);
981         if (oxe != NULL) {
982                 spin_lock(&obj->opo_lock);
983                 if (oxe->oxe_ready) {
984                         if (!oxe->oxe_exist)
985                                 GOTO(unlock, rc = -ENODATA);
986
987                         if (buf->lb_buf == NULL)
988                                 GOTO(unlock, rc = oxe->oxe_vallen);
989
990                         if (buf->lb_len < oxe->oxe_vallen)
991                                 GOTO(unlock, rc = -ERANGE);
992
993                         memcpy(buf->lb_buf, oxe->oxe_value,
994                                oxe->oxe_vallen);
995
996                         GOTO(unlock, rc = oxe->oxe_vallen);
997
998 unlock:
999                         spin_unlock(&obj->opo_lock);
1000                         osp_oac_xattr_put(oxe);
1001
1002                         return rc;
1003                 }
1004                 spin_unlock(&obj->opo_lock);
1005         }
1006         update = osp_update_request_create(dev);
1007         if (IS_ERR(update))
1008                 GOTO(out_req, rc = PTR_ERR(update));
1009
1010         rc = OSP_UPDATE_RPC_PACK(env, out_xattr_get_pack, update,
1011                                  lu_object_fid(&dt->do_lu), name, buf->lb_len);
1012         if (rc != 0) {
1013                 CERROR("%s: Insert update error "DFID": rc = %d\n",
1014                        dname, PFID(lu_object_fid(&dt->do_lu)), rc);
1015                 GOTO(out_req, rc);
1016         }
1017
1018         rc = osp_remote_sync(env, osp, update, &req);
1019
1020         down_read(&obj->opo_invalidate_sem);
1021         if (invalidated != atomic_read(&obj->opo_invalidate_seq)) {
1022                 /* invalidated has been requested, we can't cache the result */
1023                 if (rc < 0) {
1024                         if (rc == -ENOENT)
1025                                 dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
1026                         GOTO(out, rc);
1027                 }
1028                 reply = req_capsule_server_sized_get(&req->rq_pill,
1029                                                      &RMF_OUT_UPDATE_REPLY,
1030                                 OUT_UPDATE_REPLY_SIZE);
1031                 if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
1032                         CERROR("%s: Wrong version %x expected %x "DFID
1033                                ": rc = %d\n", dname, reply->ourp_magic,
1034                                UPDATE_REPLY_MAGIC,
1035                                PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
1036                         GOTO(out, rc = -EPROTO);
1037                 }
1038                 rc = object_update_result_data_get(reply, rbuf, 0);
1039                 GOTO(out, rc);
1040         }
1041
1042         if (rc < 0) {
1043                 if (rc == -ENOENT) {
1044                         dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
1045                         obj->opo_non_exist = 1;
1046                 }
1047
1048                 if (oxe == NULL)
1049                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
1050
1051                 if (oxe == NULL) {
1052                         CWARN("%s: Fail to add xattr (%s) to cache for "
1053                               DFID" (1): rc = %d\n", dname, name,
1054                               PFID(lu_object_fid(&dt->do_lu)), rc);
1055
1056                         GOTO(out, rc);
1057                 }
1058
1059                 spin_lock(&obj->opo_lock);
1060                 if (rc == -ENOENT || rc == -ENODATA) {
1061                         oxe->oxe_exist = 0;
1062                         oxe->oxe_ready = 1;
1063                 } else {
1064                         oxe->oxe_ready = 0;
1065                 }
1066                 spin_unlock(&obj->opo_lock);
1067
1068                 GOTO(out, rc);
1069         }
1070
1071         reply = req_capsule_server_sized_get(&req->rq_pill,
1072                                              &RMF_OUT_UPDATE_REPLY,
1073                                              OUT_UPDATE_REPLY_SIZE);
1074         if (reply->ourp_magic != UPDATE_REPLY_MAGIC) {
1075                 CERROR("%s: Wrong version %x expected %x "DFID": rc = %d\n",
1076                        dname, reply->ourp_magic, UPDATE_REPLY_MAGIC,
1077                        PFID(lu_object_fid(&dt->do_lu)), -EPROTO);
1078
1079                 GOTO(out, rc = -EPROTO);
1080         }
1081
1082         rc = object_update_result_data_get(reply, rbuf, 0);
1083         if (rc < 0 || rbuf->lb_len == 0) {
1084                 if (oxe == NULL && rc == -ENODATA) {
1085                         oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len);
1086                         if (oxe == NULL) {
1087                                 rc = -ENOMEM;
1088                                 CWARN("%s: Fail to add xattr (%s) to cache for "
1089                                       DFID" (1): rc = %d\n", dname, name,
1090                                       PFID(lu_object_fid(&dt->do_lu)), rc);
1091                                 GOTO(out, rc);
1092                         }
1093                 }
1094
1095                 if (oxe) {
1096                         spin_lock(&obj->opo_lock);
1097                         if (unlikely(rc == -ENODATA)) {
1098                                 oxe->oxe_exist = 0;
1099                                 oxe->oxe_ready = 1;
1100                         } else {
1101                                 oxe->oxe_ready = 0;
1102                         }
1103                         spin_unlock(&obj->opo_lock);
1104                 }
1105
1106                 GOTO(out, rc);
1107         }
1108
1109         /* For detecting EA size. */
1110         if (!buf->lb_buf)
1111                 GOTO(out, rc);
1112
1113         if (!oxe) {
1114                 oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len);
1115                 if (!oxe) {
1116                         CWARN("%s: Fail to add xattr (%s) to "
1117                               "cache for "DFID" (2): rc = %d\n",
1118                               dname, name, PFID(lu_object_fid(&dt->do_lu)), rc);
1119
1120                         GOTO(out, rc);
1121                 }
1122         }
1123
1124         oxe = osp_oac_xattr_assignment(obj, oxe, rbuf);
1125
1126         GOTO(out, rc);
1127
1128 out:
1129         up_read(&obj->opo_invalidate_sem);
1130
1131 out_req:
1132         if (rc > 0 && buf->lb_buf) {
1133                 if (unlikely(buf->lb_len < rbuf->lb_len))
1134                         rc = -ERANGE;
1135                 else
1136                         memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
1137         }
1138
1139         if (req)
1140                 ptlrpc_req_finished(req);
1141
1142         if (update && !IS_ERR(update))
1143                 osp_update_request_destroy(env, update);
1144
1145         if (oxe)
1146                 osp_oac_xattr_put(oxe);
1147
1148         return rc;
1149 }
1150
1151 /**
1152  * Implement OSP layer dt_object_operations::do_declare_xattr_set() interface.
1153  *
1154  * Declare that the caller will set extended attribute to the specified
1155  * MDT/OST object.
1156  *
1157  * If it is non-remote transaction, it will add an OUT_XATTR_SET sub-request
1158  * to the OUT RPC that will be flushed when the transaction start. And if the
1159  * OSP attributes cache is initialized, then check whether the name extended
1160  * attribute entry exists in the cache or not. If yes, replace it; otherwise,
1161  * add the extended attribute to the cache.
1162  *
1163  * \param[in] env       pointer to the thread context
1164  * \param[in] dt        pointer to the OSP layer dt_object
1165  * \param[in] buf       pointer to the lu_buf to hold the extended attribute
1166  * \param[in] name      the name of the extended attribute to be set
1167  * \param[in] flag      to indicate the detailed set operation: LU_XATTR_CREATE
1168  *                      or LU_XATTR_REPLACE or others
1169  * \param[in] th        pointer to the transaction handler
1170  *
1171  * \retval              0 for success
1172  * \retval              negative error number on failure
1173  */
1174 int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
1175                           const struct lu_buf *buf, const char *name,
1176                           int flag, struct thandle *th)
1177 {
1178         return osp_trans_update_request_create(th);
1179 }
1180
1181 /**
1182  * Implement OSP layer dt_object_operations::do_xattr_set() interface.
1183  *
1184  * Set extended attribute to the specified MDT/OST object.
1185  *
1186  * Add an OUT_XATTR_SET sub-request into the OUT RPC that will be flushed in
1187  * the transaction stop. And if the OSP attributes cache is initialized, then
1188  * check whether the name extended attribute entry exists in the cache or not.
1189  * If yes, replace it; otherwise, add the extended attribute to the cache.
1190  *
1191  * \param[in] env       pointer to the thread context
1192  * \param[in] dt        pointer to the OSP layer dt_object
1193  * \param[in] buf       pointer to the lu_buf to hold the extended attribute
1194  * \param[in] name      the name of the extended attribute to be set
1195  * \param[in] fl        to indicate the detailed set operation: LU_XATTR_CREATE
1196  *                      or LU_XATTR_REPLACE or others
1197  * \param[in] th        pointer to the transaction handler
1198  *
1199  * \retval              0 for success
1200  * \retval              negative error number on failure
1201  */
1202 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
1203                   const struct lu_buf *buf, const char *name, int fl,
1204                   struct thandle *th)
1205 {
1206         struct osp_object *o = dt2osp_obj(dt);
1207         struct osp_update_request *update;
1208         struct osp_xattr_entry *oxe;
1209         int rc;
1210         ENTRY;
1211
1212         update = thandle_to_osp_update_request(th);
1213         LASSERT(update != NULL);
1214
1215         CDEBUG(D_INODE, DFID" set xattr '%s' with size %zd\n",
1216                PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
1217
1218         rc = OSP_UPDATE_RPC_PACK(env, out_xattr_set_pack, update,
1219                                  lu_object_fid(&dt->do_lu), buf, name, fl);
1220         if (rc != 0)
1221                 RETURN(rc);
1222
1223         /* Do not cache linkEA that may be self-adjusted by peers
1224          * under EA overflow case. */
1225         if (strcmp(name, XATTR_NAME_LINK) == 0) {
1226                 oxe = osp_oac_xattr_find(o, name, true);
1227                 if (oxe != NULL)
1228                         osp_oac_xattr_put(oxe);
1229
1230                 RETURN(0);
1231         }
1232
1233         oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
1234         if (oxe == NULL) {
1235                 CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
1236                       osp_dto2name(o), name, PFID(lu_object_fid(&dt->do_lu)));
1237
1238                 RETURN(0);
1239         }
1240
1241         oxe = osp_oac_xattr_assignment(o, oxe, buf);
1242         if (oxe)
1243                 osp_oac_xattr_put(oxe);
1244
1245         RETURN(0);
1246 }
1247
1248 /**
1249  * Implement OSP layer dt_object_operations::do_declare_xattr_del() interface.
1250  *
1251  * Declare that the caller will delete extended attribute on the specified
1252  * MDT/OST object.
1253  *
1254  * If it is non-remote transaction, it will add an OUT_XATTR_DEL sub-request
1255  * to the OUT RPC that will be flushed when the transaction start. And if the
1256  * name extended attribute entry exists in the OSP attributes cache, then remove
1257  * it from the cache.
1258  *
1259  * \param[in] env       pointer to the thread context
1260  * \param[in] dt        pointer to the OSP layer dt_object
1261  * \param[in] name      the name of the extended attribute to be set
1262  * \param[in] th        pointer to the transaction handler
1263  *
1264  * \retval              0 for success
1265  * \retval              negative error number on failure
1266  */
1267 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
1268                           const char *name, struct thandle *th)
1269 {
1270         return osp_trans_update_request_create(th);
1271 }
1272
1273 /**
1274  * Implement OSP layer dt_object_operations::do_xattr_del() interface.
1275  *
1276  * Delete extended attribute on the specified MDT/OST object.
1277  *
1278  * If it is remote transaction, it will add an OUT_XATTR_DEL sub-request into
1279  * the OUT RPC that will be flushed when the transaction stop. And if the name
1280  * extended attribute entry exists in the OSP attributes cache, then remove it
1281  * from the cache.
1282  *
1283  * \param[in] env       pointer to the thread context
1284  * \param[in] dt        pointer to the OSP layer dt_object
1285  * \param[in] name      the name of the extended attribute to be set
1286  * \param[in] th        pointer to the transaction handler
1287  *
1288  * \retval              0 for success
1289  * \retval              negative error number on failure
1290  */
1291 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
1292                   const char *name, struct thandle *th)
1293 {
1294         struct osp_update_request *update;
1295         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1296         struct osp_object *o = dt2osp_obj(dt);
1297         struct osp_xattr_entry *oxe;
1298         int rc;
1299
1300         update = thandle_to_osp_update_request(th);
1301         LASSERT(update != NULL);
1302
1303         rc = OSP_UPDATE_RPC_PACK(env, out_xattr_del_pack, update, fid, name);
1304         if (rc != 0)
1305                 return rc;
1306
1307         oxe = osp_oac_xattr_find(o, name, true);
1308         if (oxe != NULL)
1309                 /* Drop the ref for entry on list. */
1310                 osp_oac_xattr_put(oxe);
1311
1312         return 0;
1313 }
1314
1315 void osp_obj_invalidate_cache(struct osp_object *obj)
1316 {
1317         struct osp_xattr_entry *oxe;
1318         struct osp_xattr_entry *tmp;
1319
1320         spin_lock(&obj->opo_lock);
1321         list_for_each_entry_safe(oxe, tmp, &obj->opo_xattr_list, oxe_list) {
1322                 oxe->oxe_ready = 0;
1323                 list_del_init(&oxe->oxe_list);
1324                 osp_oac_xattr_put(oxe);
1325         }
1326         obj->opo_attr.la_valid = 0;
1327         spin_unlock(&obj->opo_lock);
1328 }
1329
1330 /**
1331  * Implement OSP layer dt_object_operations::do_invalidate() interface.
1332  *
1333  * Invalidate attributes cached on the specified MDT/OST object.
1334  *
1335  * \param[in] env       pointer to the thread context
1336  * \param[in] dt        pointer to the OSP layer dt_object
1337  *
1338  * \retval              0 for success
1339  * \retval              negative error number on failure
1340  */
1341 int osp_invalidate(const struct lu_env *env, struct dt_object *dt)
1342 {
1343         struct osp_object *obj = dt2osp_obj(dt);
1344         ENTRY;
1345
1346         CDEBUG(D_HA, "Invalidate osp_object "DFID"\n",
1347                PFID(lu_object_fid(&dt->do_lu)));
1348
1349         /* serialize attr/EA set vs. invalidation */
1350         down_write(&obj->opo_invalidate_sem);
1351
1352         /* this should invalidate all in-flights */
1353         atomic_inc(&obj->opo_invalidate_seq);
1354
1355         spin_lock(&obj->opo_lock);
1356         /* do not mark new objects stale */
1357         if (obj->opo_attr.la_valid)
1358                 obj->opo_stale = 1;
1359         obj->opo_non_exist = 0;
1360         spin_unlock(&obj->opo_lock);
1361
1362         osp_obj_invalidate_cache(obj);
1363
1364         up_write(&obj->opo_invalidate_sem);
1365
1366         RETURN(0);
1367 }
1368
1369 /**
1370  * Implement OSP layer dt_object_operations::do_declare_create() interface.
1371  *
1372  * Declare that the caller will create the OST object.
1373  *
1374  * If the transaction is a remote transaction and the FID for the OST-object
1375  * has been assigned already, then handle it as creating (remote) MDT object
1376  * via osp_md_declare_create(). This function is usually used for LFSCK
1377  * to re-create the lost OST object. Otherwise, if it is not replay case, the
1378  * OSP will reserve pre-created object for the subsequent create operation;
1379  * if the MDT side cached pre-created objects are less than some threshold,
1380  * then it will wakeup the pre-create thread.
1381  *
1382  * \param[in] env       pointer to the thread context
1383  * \param[in] dt        pointer to the OSP layer dt_object
1384  * \param[in] attr      the attribute for the object to be created
1385  * \param[in] hint      pointer to the hint for creating the object, such as
1386  *                      the parent object
1387  * \param[in] dof       pointer to the dt_object_format for help the creation
1388  * \param[in] th        pointer to the transaction handler
1389  *
1390  * \retval              0 for success
1391  * \retval              negative error number on failure
1392  */
1393 static int osp_declare_create(const struct lu_env *env, struct dt_object *dt,
1394                               struct lu_attr *attr,
1395                               struct dt_allocation_hint *hint,
1396                               struct dt_object_format *dof, struct thandle *th)
1397 {
1398         struct osp_thread_info  *osi = osp_env_info(env);
1399         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1400         struct osp_object       *o = dt2osp_obj(dt);
1401         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1402         struct thandle          *local_th;
1403         int                      rc = 0;
1404
1405         ENTRY;
1406
1407         if (is_only_remote_trans(th) && !fid_is_zero(fid)) {
1408                 LASSERT(fid_is_sane(fid));
1409
1410                 rc = osp_md_declare_create(env, dt, attr, hint, dof, th);
1411
1412                 RETURN(rc);
1413         }
1414
1415         /* should happen to non-0 OSP only so that at least one object
1416          * has been already declared in the scenario and LOD should
1417          * cleanup that */
1418         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OSC_CREATE_FAIL) && d->opd_index == 1)
1419                 RETURN(-ENOSPC);
1420
1421         LASSERT(d->opd_last_used_oid_file);
1422
1423         /*
1424          * There can be gaps in precreated ids and record to unlink llog
1425          * XXX: we do not handle gaps yet, implemented before solution
1426          *      was found to be racy, so we disabled that. there is no
1427          *      point in making useless but expensive llog declaration.
1428          */
1429         /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
1430
1431         local_th = osp_get_storage_thandle(env, th, d);
1432         if (IS_ERR(local_th))
1433                 RETURN(PTR_ERR(local_th));
1434
1435         if (unlikely(!fid_is_zero(fid))) {
1436                 /* replay case: caller knows fid */
1437                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, NULL,
1438                                    d->opd_index);
1439                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1440                                              &osi->osi_lb, osi->osi_off,
1441                                              local_th);
1442                 RETURN(rc);
1443         }
1444
1445         /*
1446          * in declaration we need to reserve object so that we don't block
1447          * awaiting precreation RPC to complete
1448          */
1449         rc = osp_precreate_reserve(env, d);
1450         /*
1451          * we also need to declare update to local "last used id" file for
1452          * recovery if object isn't used for a reason, we need to release
1453          * reservation, this can be made in osd_object_release()
1454          */
1455         if (rc == 0) {
1456                 /* mark id is reserved: in create we don't want to talk
1457                  * to OST */
1458                 LASSERT(o->opo_reserved == 0);
1459                 o->opo_reserved = 1;
1460
1461                 /* common for all OSPs file hystorically */
1462                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, NULL,
1463                                    d->opd_index);
1464                 rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
1465                                              &osi->osi_lb, osi->osi_off,
1466                                              local_th);
1467         } else {
1468                 /* not needed in the cache anymore */
1469                 set_bit(LU_OBJECT_HEARD_BANSHEE,
1470                             &dt->do_lu.lo_header->loh_flags);
1471         }
1472         RETURN(rc);
1473 }
1474
1475 /**
1476  * Implement OSP layer dt_object_operations::do_create() interface.
1477  *
1478  * Create the OST object.
1479  *
1480  * If the transaction is a remote transaction and the FID for the OST-object
1481  * has been assigned already, then handle it as handling MDT object via the
1482  * osp_md_create(). For other cases, the OSP will assign FID to the
1483  * object to be created, and update last_used Object ID (OID) file.
1484  *
1485  * \param[in] env       pointer to the thread context
1486  * \param[in] dt        pointer to the OSP layer dt_object
1487  * \param[in] attr      the attribute for the object to be created
1488  * \param[in] hint      pointer to the hint for creating the object, such as
1489  *                      the parent object
1490  * \param[in] dof       pointer to the dt_object_format for help the creation
1491  * \param[in] th        pointer to the transaction handler
1492  *
1493  * \retval              0 for success
1494  * \retval              negative error number on failure
1495  */
1496 static int osp_create(const struct lu_env *env, struct dt_object *dt,
1497                       struct lu_attr *attr, struct dt_allocation_hint *hint,
1498                       struct dt_object_format *dof, struct thandle *th)
1499 {
1500         struct osp_thread_info  *osi = osp_env_info(env);
1501         struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
1502         struct osp_object       *o = dt2osp_obj(dt);
1503         int                     rc = 0;
1504         struct lu_fid           *fid = &osi->osi_fid;
1505         struct thandle          *local_th;
1506         ENTRY;
1507
1508         if (is_only_remote_trans(th) &&
1509             !fid_is_zero(lu_object_fid(&dt->do_lu))) {
1510                 LASSERT(fid_is_sane(lu_object_fid(&dt->do_lu)));
1511
1512                 rc = osp_md_create(env, dt, attr, hint, dof, th);
1513                 if (rc == 0)
1514                         o->opo_non_exist = 0;
1515
1516                 RETURN(rc);
1517         }
1518
1519         o->opo_non_exist = 0;
1520         if (o->opo_reserved) {
1521                 /* regular case, fid is assigned holding transaction open */
1522                  osp_object_assign_fid(env, d, o);
1523         }
1524
1525         memcpy(fid, lu_object_fid(&dt->do_lu), sizeof(*fid));
1526
1527         LASSERTF(fid_is_sane(fid), "fid for osp_object %p is insane"DFID"!\n",
1528                  o, PFID(fid));
1529
1530         if (!o->opo_reserved) {
1531                 /* special case, id was assigned outside of transaction
1532                  * see comments in osp_declare_attr_set */
1533                 LASSERT(d->opd_pre != NULL);
1534                 spin_lock(&d->opd_pre_lock);
1535                 osp_update_last_fid(d, fid);
1536                 spin_unlock(&d->opd_pre_lock);
1537         }
1538
1539         CDEBUG(D_INODE, "fid for osp_object %p is "DFID"\n", o, PFID(fid));
1540
1541         /* If the precreate ends, it means it will be ready to rollover to
1542          * the new sequence soon, all the creation should be synchronized,
1543          * otherwise during replay, the replay fid will be inconsistent with
1544          * last_used/create fid */
1545         if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
1546                 th->th_sync = 1;
1547
1548         local_th = osp_get_storage_thandle(env, th, d);
1549         if (IS_ERR(local_th))
1550                 RETURN(PTR_ERR(local_th));
1551         /*
1552          * it's OK if the import is inactive by this moment - id was created
1553          * by OST earlier, we just need to maintain it consistently on the disk
1554          * once import is reconnected, OSP will claim this and other objects
1555          * used and OST either keep them, if they exist or recreate
1556          */
1557
1558         /* we might have lost precreated objects */
1559         if (unlikely(d->opd_gap_count) > 0) {
1560                 LASSERT(d->opd_pre != NULL);
1561                 spin_lock(&d->opd_pre_lock);
1562                 if (d->opd_gap_count > 0) {
1563                         int count = d->opd_gap_count;
1564
1565                         rc = ostid_set_id(&osi->osi_oi,
1566                                           fid_oid(&d->opd_gap_start_fid));
1567                         if (rc) {
1568                                 spin_unlock(&d->opd_pre_lock);
1569                                 RETURN(rc);
1570                         }
1571                         d->opd_gap_count = 0;
1572                         spin_unlock(&d->opd_pre_lock);
1573
1574                         CDEBUG(D_HA, "Writing gap "DFID"+%d in llog\n",
1575                                PFID(&d->opd_gap_start_fid), count);
1576                         /* real gap handling is disabled intil ORI-692 will be
1577                          * fixed, now we only report gaps */
1578                 } else {
1579                         spin_unlock(&d->opd_pre_lock);
1580                 }
1581         }
1582
1583         /* Only need update last_used oid file, seq file will only be update
1584          * during seq rollover */
1585         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off,
1586                            &d->opd_last_id, d->opd_index);
1587
1588         rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
1589                              &osi->osi_off, local_th);
1590
1591         CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
1592                d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
1593
1594         RETURN(rc);
1595 }
1596
1597 /**
1598  * Implement OSP layer dt_object_operations::do_declare_destroy() interface.
1599  *
1600  * Declare that the caller will destroy the specified OST object.
1601  *
1602  * The OST object destroy will be handled via llog asynchronously. This
1603  * function will declare the credits for generating MDS_UNLINK64_REC llog.
1604  *
1605  * \param[in] env       pointer to the thread context
1606  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1607  * \param[in] th        pointer to the transaction handler
1608  *
1609  * \retval              0 for success
1610  * \retval              negative error number on failure
1611  */
1612 int osp_declare_destroy(const struct lu_env *env, struct dt_object *dt,
1613                         struct thandle *th)
1614 {
1615         struct osp_object       *o = dt2osp_obj(dt);
1616         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1617         int                      rc = 0;
1618
1619         ENTRY;
1620
1621         LASSERT(!osp->opd_connect_mdt);
1622
1623         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
1624                 rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
1625
1626         RETURN(rc);
1627 }
1628
1629 /**
1630  * Implement OSP layer dt_object_operations::do_destroy() interface.
1631  *
1632  * Destroy the specified OST object.
1633  *
1634  * The OSP generates a MDS_UNLINK64_REC record in the llog. There
1635  * will be some dedicated thread to handle the llog asynchronously.
1636  *
1637  * It also marks the object as non-cached.
1638  *
1639  * \param[in] env       pointer to the thread context
1640  * \param[in] dt        pointer to the OSP layer dt_object to be destroyed
1641  * \param[in] th        pointer to the transaction handler
1642  *
1643  * \retval              0 for success
1644  * \retval              negative error number on failure
1645  */
1646 static int osp_destroy(const struct lu_env *env, struct dt_object *dt,
1647                        struct thandle *th)
1648 {
1649         struct osp_object       *o = dt2osp_obj(dt);
1650         struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
1651         int                      rc = 0;
1652
1653         ENTRY;
1654
1655         o->opo_non_exist = 1;
1656
1657         LASSERT(!osp->opd_connect_mdt);
1658
1659         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ)) {
1660                 /* once transaction is committed put proper command on
1661                  * the queue going to our OST. */
1662                 rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
1663                 if (rc < 0)
1664                         RETURN(rc);
1665         }
1666
1667         /* not needed in cache any more */
1668         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
1669
1670         RETURN(rc);
1671 }
1672
1673 static int osp_orphan_index_lookup(const struct lu_env *env,
1674                                    struct dt_object *dt,
1675                                    struct dt_rec *rec,
1676                                    const struct dt_key *key)
1677 {
1678         return -EOPNOTSUPP;
1679 }
1680
1681 static int osp_orphan_index_declare_insert(const struct lu_env *env,
1682                                            struct dt_object *dt,
1683                                            const struct dt_rec *rec,
1684                                            const struct dt_key *key,
1685                                            struct thandle *handle)
1686 {
1687         return -EOPNOTSUPP;
1688 }
1689
1690 static int osp_orphan_index_insert(const struct lu_env *env,
1691                                    struct dt_object *dt,
1692                                    const struct dt_rec *rec,
1693                                    const struct dt_key *key,
1694                                    struct thandle *handle)
1695 {
1696         return -EOPNOTSUPP;
1697 }
1698
1699 static int osp_orphan_index_declare_delete(const struct lu_env *env,
1700                                            struct dt_object *dt,
1701                                            const struct dt_key *key,
1702                                            struct thandle *handle)
1703 {
1704         return -EOPNOTSUPP;
1705 }
1706
1707 static int osp_orphan_index_delete(const struct lu_env *env,
1708                                    struct dt_object *dt,
1709                                    const struct dt_key *key,
1710                                    struct thandle *handle)
1711 {
1712         return -EOPNOTSUPP;
1713 }
1714
1715 /**
1716  * Initialize the OSP layer index iteration.
1717  *
1718  * \param[in] env       pointer to the thread context
1719  * \param[in] dt        pointer to the index object to be iterated
1720  * \param[in] attr      unused
1721  *
1722  * \retval              pointer to the iteration structure
1723  * \retval              negative error number on failure
1724  */
1725 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
1726                           __u32 attr)
1727 {
1728         struct osp_it *it;
1729
1730         OBD_ALLOC_PTR(it);
1731         if (it == NULL)
1732                 return ERR_PTR(-ENOMEM);
1733
1734         it->ooi_pos_ent = -1;
1735         it->ooi_obj = dt;
1736         it->ooi_attr = attr;
1737
1738         return (struct dt_it *)it;
1739 }
1740
1741 /**
1742  * Finalize the OSP layer index iteration.
1743  *
1744  * \param[in] env       pointer to the thread context
1745  * \param[in] di        pointer to the iteration structure
1746  */
1747 void osp_it_fini(const struct lu_env *env, struct dt_it *di)
1748 {
1749         struct osp_it   *it = (struct osp_it *)di;
1750         struct page     **pages = it->ooi_pages;
1751         int             npages = it->ooi_total_npages;
1752         int             i;
1753
1754         if (pages != NULL) {
1755                 for (i = 0; i < npages; i++) {
1756                         if (pages[i] != NULL) {
1757                                 if (pages[i] == it->ooi_cur_page) {
1758                                         kunmap(pages[i]);
1759                                         it->ooi_cur_page = NULL;
1760                                 }
1761                                 __free_page(pages[i]);
1762                         }
1763                 }
1764                 OBD_FREE(pages, npages * sizeof(*pages));
1765         }
1766         OBD_FREE_PTR(it);
1767 }
1768
1769 /**
1770  * Get more records for the iteration from peer.
1771  *
1772  * The new records will be filled in an array of pages. The OSP side
1773  * allows 1MB bulk data to be transferred.
1774  *
1775  * \param[in] env       pointer to the thread context
1776  * \param[in] it        pointer to the iteration structure
1777  *
1778  * \retval              0 for success
1779  * \retval              negative error number on failure
1780  */
1781 static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
1782 {
1783         struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
1784         struct osp_device        *osp   = lu2osp_dev(dev);
1785         struct page             **pages;
1786         struct ptlrpc_request    *req   = NULL;
1787         struct ptlrpc_bulk_desc  *desc;
1788         struct idx_info          *ii;
1789         int                       npages;
1790         int                       rc;
1791         int                       i;
1792         ENTRY;
1793
1794         /* 1MB bulk */
1795         npages = min_t(unsigned int, OFD_MAX_BRW_SIZE, 1 << 20);
1796         npages /= PAGE_SIZE;
1797
1798         OBD_ALLOC(pages, npages * sizeof(*pages));
1799         if (pages == NULL)
1800                 RETURN(-ENOMEM);
1801
1802         it->ooi_pages = pages;
1803         it->ooi_total_npages = npages;
1804         for (i = 0; i < npages; i++) {
1805                 pages[i] = alloc_page(GFP_NOFS);
1806                 if (pages[i] == NULL)
1807                         RETURN(-ENOMEM);
1808         }
1809
1810         req = ptlrpc_request_alloc(osp->opd_obd->u.cli.cl_import,
1811                                    &RQF_OBD_IDX_READ);
1812         if (req == NULL)
1813                 RETURN(-ENOMEM);
1814
1815         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, OBD_IDX_READ);
1816         if (rc != 0) {
1817                 ptlrpc_request_free(req);
1818                 RETURN(rc);
1819         }
1820
1821         osp_set_req_replay(osp, req);
1822         req->rq_request_portal = OUT_PORTAL;
1823         ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
1824         memset(ii, 0, sizeof(*ii));
1825         if (fid_is_last_id(lu_object_fid(&it->ooi_obj->do_lu))) {
1826                 /* LFSCK will iterate orphan object[FID_SEQ_LAYOUT_BTREE,
1827                  * ost_index, 0] with LAST_ID FID, so it needs to replace
1828                  * the FID with orphan FID here */
1829                 ii->ii_fid.f_seq = FID_SEQ_LAYOUT_RBTREE;
1830                 ii->ii_fid.f_oid = osp->opd_index;
1831                 ii->ii_fid.f_ver = 0;
1832                 ii->ii_flags = II_FL_NOHASH;
1833                 ii->ii_attrs = osp_dev2node(osp);
1834         } else {
1835                 ii->ii_fid = *lu_object_fid(&it->ooi_obj->do_lu);
1836                 ii->ii_flags = II_FL_NOHASH | II_FL_NOKEY | II_FL_VARKEY |
1837                                II_FL_VARREC;
1838                 ii->ii_attrs = it->ooi_attr;
1839         }
1840         ii->ii_magic = IDX_INFO_MAGIC;
1841         ii->ii_count = npages * LU_PAGE_COUNT;
1842         ii->ii_hash_start = it->ooi_next;
1843
1844         ptlrpc_at_set_req_timeout(req);
1845
1846         desc = ptlrpc_prep_bulk_imp(req, npages, 1,
1847                                     PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
1848                                     MDS_BULK_PORTAL,
1849                                     &ptlrpc_bulk_kiov_pin_ops);
1850         if (desc == NULL)
1851                 GOTO(out, rc = -ENOMEM);
1852
1853         for (i = 0; i < npages; i++)
1854                 desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
1855                                                  PAGE_SIZE);
1856
1857         ptlrpc_request_set_replen(req);
1858         rc = ptlrpc_queue_wait(req);
1859         if (rc != 0)
1860                 GOTO(out, rc);
1861
1862         rc = sptlrpc_cli_unwrap_bulk_read(req, req->rq_bulk,
1863                                           req->rq_bulk->bd_nob_transferred);
1864         if (rc < 0)
1865                 GOTO(out, rc);
1866         rc = 0;
1867
1868         ii = req_capsule_server_get(&req->rq_pill, &RMF_IDX_INFO);
1869         if (ii->ii_magic != IDX_INFO_MAGIC)
1870                  GOTO(out, rc = -EPROTO);
1871
1872         npages = (ii->ii_count + LU_PAGE_COUNT - 1) >>
1873                  (PAGE_SHIFT - LU_PAGE_SHIFT);
1874         if (npages > it->ooi_total_npages) {
1875                 CERROR("%s: returned more pages than expected, %u > %u\n",
1876                        osp->opd_obd->obd_name, npages, it->ooi_total_npages);
1877                 GOTO(out, rc = -EINVAL);
1878         }
1879
1880         it->ooi_rec_size = ii->ii_recsize;
1881         it->ooi_valid_npages = npages;
1882         if (ptlrpc_rep_need_swab(req))
1883                 it->ooi_swab = 1;
1884
1885         it->ooi_next = ii->ii_hash_end;
1886
1887 out:
1888         ptlrpc_req_finished(req);
1889
1890         return rc;
1891 }
1892
1893 /**
1894  * Move the iteration cursor to the next lu_page.
1895  *
1896  * One system page (PAGE_SIZE) may contain multiple lu_page (4KB),
1897  * that depends on the LU_PAGE_COUNT. If it is not the last lu_page
1898  * in current system page, then move the iteration cursor to the next
1899  * lu_page in current system page. Otherwise, if there are more system
1900  * pages in the cache, then move the iteration cursor to the next system
1901  * page. If all the cached records (pages) have been iterated, then fetch
1902  * more records via osp_it_fetch().
1903  *
1904  * \param[in] env       pointer to the thread context
1905  * \param[in] di        pointer to the iteration structure
1906  *
1907  * \retval              positive for end of the directory
1908  * \retval              0 for success
1909  * \retval              negative error number on failure
1910  */
1911 int osp_it_next_page(const struct lu_env *env, struct dt_it *di)
1912 {
1913         struct osp_it           *it = (struct osp_it *)di;
1914         struct lu_idxpage       *idxpage;
1915         struct page             **pages;
1916         int                     rc;
1917         int                     i;
1918         ENTRY;
1919
1920 again2:
1921         idxpage = it->ooi_cur_idxpage;
1922         if (idxpage != NULL) {
1923                 if (idxpage->lip_nr == 0)
1924                         RETURN(1);
1925
1926                 if (it->ooi_pos_ent < idxpage->lip_nr) {
1927                         CDEBUG(D_INFO, "ooi_pos %d nr %d\n",
1928                                (int)it->ooi_pos_ent, (int)idxpage->lip_nr);
1929                         RETURN(0);
1930                 }
1931                 it->ooi_cur_idxpage = NULL;
1932                 it->ooi_pos_lu_page++;
1933
1934 again1:
1935                 if (it->ooi_pos_lu_page < LU_PAGE_COUNT) {
1936                         it->ooi_cur_idxpage = (void *)it->ooi_cur_page +
1937                                          LU_PAGE_SIZE * it->ooi_pos_lu_page;
1938                         if (it->ooi_swab)
1939                                 lustre_swab_lip_header(it->ooi_cur_idxpage);
1940                         if (it->ooi_cur_idxpage->lip_magic != LIP_MAGIC) {
1941                                 struct osp_device *osp =
1942                                         lu2osp_dev(it->ooi_obj->do_lu.lo_dev);
1943
1944                                 CERROR("%s: invalid magic (%x != %x) for page "
1945                                        "%d/%d while read layout orphan index\n",
1946                                        osp->opd_obd->obd_name,
1947                                        it->ooi_cur_idxpage->lip_magic,
1948                                        LIP_MAGIC, it->ooi_pos_page,
1949                                        it->ooi_pos_lu_page);
1950                                 /* Skip this lu_page next time. */
1951                                 it->ooi_pos_ent = idxpage->lip_nr - 1;
1952                                 RETURN(-EINVAL);
1953                         }
1954                         it->ooi_pos_ent = -1;
1955                         goto again2;
1956                 }
1957
1958                 kunmap(it->ooi_cur_page);
1959                 it->ooi_cur_page = NULL;
1960                 it->ooi_pos_page++;
1961
1962 again0:
1963                 pages = it->ooi_pages;
1964                 if (it->ooi_pos_page < it->ooi_valid_npages) {
1965                         it->ooi_cur_page = kmap(pages[it->ooi_pos_page]);
1966                         it->ooi_pos_lu_page = 0;
1967                         goto again1;
1968                 }
1969
1970                 for (i = 0; i < it->ooi_total_npages; i++) {
1971                         if (pages[i] != NULL)
1972                                 __free_page(pages[i]);
1973                 }
1974                 OBD_FREE(pages, it->ooi_total_npages * sizeof(*pages));
1975
1976                 it->ooi_pos_page = 0;
1977                 it->ooi_total_npages = 0;
1978                 it->ooi_valid_npages = 0;
1979                 it->ooi_swab = 0;
1980                 it->ooi_ent = NULL;
1981                 it->ooi_cur_page = NULL;
1982                 it->ooi_cur_idxpage = NULL;
1983                 it->ooi_pages = NULL;
1984         }
1985
1986         if (it->ooi_next == II_END_OFF)
1987                 RETURN(1);
1988
1989         rc = osp_it_fetch(env, it);
1990         if (rc == 0)
1991                 goto again0;
1992
1993         RETURN(rc);
1994 }
1995
1996 /**
1997  * Move the iteration cursor to the next record.
1998  *
1999  * If there are more records in the lu_page, then move the iteration
2000  * cursor to the next record directly. Otherwise, move the iteration
2001  * cursor to the record in the next lu_page via osp_it_next_page()
2002  *
2003  * \param[in] env       pointer to the thread context
2004  * \param[in] di        pointer to the iteration structure
2005  *
2006  * \retval              positive for end of the directory
2007  * \retval              0 for success
2008  * \retval              negative error number on failure
2009  */
2010 static int osp_orphan_it_next(const struct lu_env *env, struct dt_it *di)
2011 {
2012         struct osp_it           *it = (struct osp_it *)di;
2013         struct lu_idxpage       *idxpage;
2014         int                     rc;
2015         ENTRY;
2016
2017 again:
2018         idxpage = it->ooi_cur_idxpage;
2019         if (idxpage != NULL) {
2020                 if (idxpage->lip_nr == 0)
2021                         RETURN(1);
2022
2023                 it->ooi_pos_ent++;
2024                 if (it->ooi_pos_ent < idxpage->lip_nr) {
2025                         if (it->ooi_rec_size ==
2026                                         sizeof(struct lu_orphan_rec_v3)) {
2027                                 it->ooi_ent =
2028                                 (struct lu_orphan_ent_v3 *)idxpage->lip_entries+
2029                                                         it->ooi_pos_ent;
2030                                 if (it->ooi_swab)
2031                                         lustre_swab_orphan_ent_v3(it->ooi_ent);
2032                         } else if (it->ooi_rec_size ==
2033                                         sizeof(struct lu_orphan_rec_v2)) {
2034                                 it->ooi_ent =
2035                                 (struct lu_orphan_ent_v2 *)idxpage->lip_entries+
2036                                                         it->ooi_pos_ent;
2037                                 if (it->ooi_swab)
2038                                         lustre_swab_orphan_ent_v2(it->ooi_ent);
2039                         } else {
2040                                 it->ooi_ent =
2041                                 (struct lu_orphan_ent *)idxpage->lip_entries +
2042                                                         it->ooi_pos_ent;
2043                                 if (it->ooi_swab)
2044                                         lustre_swab_orphan_ent(it->ooi_ent);
2045                         }
2046                         RETURN(0);
2047                 }
2048         }
2049
2050         rc = osp_it_next_page(env, di);
2051         if (rc == 0)
2052                 goto again;
2053
2054         RETURN(rc);
2055 }
2056
2057 int osp_it_get(const struct lu_env *env, struct dt_it *di,
2058                const struct dt_key *key)
2059 {
2060         return 1;
2061 }
2062
2063 void osp_it_put(const struct lu_env *env, struct dt_it *di)
2064 {
2065 }
2066
2067 static struct dt_key *osp_orphan_it_key(const struct lu_env *env,
2068                                         const struct dt_it *di)
2069 {
2070         struct osp_it   *it  = (struct osp_it *)di;
2071         struct lu_orphan_ent    *ent = (struct lu_orphan_ent *)it->ooi_ent;
2072
2073         if (likely(ent != NULL))
2074                 return (struct dt_key *)(&ent->loe_key);
2075
2076         return NULL;
2077 }
2078
2079 static int osp_orphan_it_key_size(const struct lu_env *env,
2080                                   const struct dt_it *di)
2081 {
2082         return sizeof(struct lu_fid);
2083 }
2084
2085 static int osp_orphan_it_rec(const struct lu_env *env, const struct dt_it *di,
2086                              struct dt_rec *rec, __u32 attr)
2087 {
2088         struct osp_it *it = (struct osp_it *)di;
2089
2090         if (likely(it->ooi_ent)) {
2091                 if (it->ooi_rec_size == sizeof(struct lu_orphan_rec_v3)) {
2092                         struct lu_orphan_ent_v3 *ent =
2093                                 (struct lu_orphan_ent_v3 *)it->ooi_ent;
2094
2095                         *(struct lu_orphan_rec_v3 *)rec = ent->loe_rec;
2096                 } else if (it->ooi_rec_size ==
2097                                 sizeof(struct lu_orphan_rec_v2)) {
2098                         struct lu_orphan_ent_v2 *ent =
2099                                 (struct lu_orphan_ent_v2 *)it->ooi_ent;
2100
2101                         *(struct lu_orphan_rec_v2 *)rec = ent->loe_rec;
2102                 } else {
2103                         struct lu_orphan_ent *ent =
2104                                 (struct lu_orphan_ent *)it->ooi_ent;
2105
2106                         *(struct lu_orphan_rec *)rec = ent->loe_rec;
2107                 }
2108                 return 0;
2109         }
2110
2111         return -EINVAL;
2112 }
2113
2114 __u64 osp_it_store(const struct lu_env *env, const struct dt_it *di)
2115 {
2116         struct osp_it   *it = (struct osp_it *)di;
2117
2118         return it->ooi_next;
2119 }
2120
2121 /**
2122  * Locate the iteration cursor to the specified position (cookie).
2123  *
2124  * \param[in] env       pointer to the thread context
2125  * \param[in] di        pointer to the iteration structure
2126  * \param[in] hash      the specified position
2127  *
2128  * \retval              positive number for locating to the exactly position
2129  *                      or the next
2130  * \retval              0 for arriving at the end of the iteration
2131  * \retval              negative error number on failure
2132  */
2133 int osp_orphan_it_load(const struct lu_env *env, const struct dt_it *di,
2134                        __u64 hash)
2135 {
2136         struct osp_it   *it     = (struct osp_it *)di;
2137         int              rc;
2138
2139         it->ooi_next = hash;
2140         rc = osp_orphan_it_next(env, (struct dt_it *)di);
2141         if (rc == 1)
2142                 return 0;
2143
2144         if (rc == 0)
2145                 return 1;
2146
2147         return rc;
2148 }
2149
2150 int osp_it_key_rec(const struct lu_env *env, const struct dt_it *di,
2151                    void *key_rec)
2152 {
2153         return 0;
2154 }
2155
2156 static const struct dt_index_operations osp_orphan_index_ops = {
2157         .dio_lookup             = osp_orphan_index_lookup,
2158         .dio_declare_insert     = osp_orphan_index_declare_insert,
2159         .dio_insert             = osp_orphan_index_insert,
2160         .dio_declare_delete     = osp_orphan_index_declare_delete,
2161         .dio_delete             = osp_orphan_index_delete,
2162         .dio_it = {
2163                 .init           = osp_it_init,
2164                 .fini           = osp_it_fini,
2165                 .next           = osp_orphan_it_next,
2166                 .get            = osp_it_get,
2167                 .put            = osp_it_put,
2168                 .key            = osp_orphan_it_key,
2169                 .key_size       = osp_orphan_it_key_size,
2170                 .rec            = osp_orphan_it_rec,
2171                 .store          = osp_it_store,
2172                 .load           = osp_orphan_it_load,
2173                 .key_rec        = osp_it_key_rec,
2174         }
2175 };
2176
2177 /**
2178  * Implement OSP layer dt_object_operations::do_index_try() interface.
2179  *
2180  * Negotiate the index type.
2181  *
2182  * If the target index is an IDIF object, then use osp_orphan_index_ops.
2183  * Otherwise, assign osp_md_index_ops to the dt_object::do_index_ops.
2184  * (\see lustre/include/lustre_fid.h for IDIF.)
2185  *
2186  * \param[in] env       pointer to the thread context
2187  * \param[in] dt        pointer to the OSP layer dt_object
2188  * \param[in] feat      unused
2189  *
2190  * \retval              0 for success
2191  */
2192 static int osp_index_try(const struct lu_env *env,
2193                          struct dt_object *dt,
2194                          const struct dt_index_features *feat)
2195 {
2196         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2197
2198         if (fid_is_last_id(fid) && fid_is_idif(fid))
2199                 dt->do_index_ops = &osp_orphan_index_ops;
2200         else
2201                 dt->do_index_ops = &osp_md_index_ops;
2202         return 0;
2203 }
2204
2205 static struct dt_object_operations osp_obj_ops = {
2206         .do_declare_attr_get    = osp_declare_attr_get,
2207         .do_attr_get            = osp_attr_get,
2208         .do_declare_attr_set    = osp_declare_attr_set,
2209         .do_attr_set            = osp_attr_set,
2210         .do_declare_xattr_get   = osp_declare_xattr_get,
2211         .do_xattr_get           = osp_xattr_get,
2212         .do_declare_xattr_set   = osp_declare_xattr_set,
2213         .do_xattr_set           = osp_xattr_set,
2214         .do_declare_create      = osp_declare_create,
2215         .do_create              = osp_create,
2216         .do_declare_destroy     = osp_declare_destroy,
2217         .do_destroy             = osp_destroy,
2218         .do_index_try           = osp_index_try,
2219 };
2220
2221 /**
2222  * Implement OSP layer lu_object_operations::loo_object_init() interface.
2223  *
2224  * Initialize the object.
2225  *
2226  * If it is a remote MDT object, then call do_attr_get() to fetch
2227  * the attribute from the peer.
2228  *
2229  * \param[in] env       pointer to the thread context
2230  * \param[in] o         pointer to the OSP layer lu_object
2231  * \param[in] conf      unused
2232  *
2233  * \retval              0 for success
2234  * \retval              negative error number on failure
2235  */
2236 static int osp_object_init(const struct lu_env *env, struct lu_object *o,
2237                            const struct lu_object_conf *conf)
2238 {
2239         struct osp_object       *po = lu2osp_obj(o);
2240         int                     rc = 0;
2241         ENTRY;
2242
2243         spin_lock_init(&po->opo_lock);
2244         o->lo_header->loh_attr |= LOHA_REMOTE;
2245         INIT_LIST_HEAD(&po->opo_xattr_list);
2246         INIT_LIST_HEAD(&po->opo_invalidate_cb_list);
2247         init_rwsem(&po->opo_invalidate_sem);
2248
2249         if (is_ost_obj(o)) {
2250                 po->opo_obj.do_ops = &osp_obj_ops;
2251         } else {
2252                 struct lu_attr *la = &osp_env_info(env)->osi_attr;
2253
2254                 po->opo_obj.do_ops = &osp_md_obj_ops;
2255                 po->opo_obj.do_body_ops = &osp_md_body_ops;
2256
2257                 if (conf != NULL && conf->loc_flags & LOC_F_NEW) {
2258                         po->opo_non_exist = 1;
2259                 } else {
2260                         rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
2261                                                              la);
2262                         if (rc == 0)
2263                                 o->lo_header->loh_attr |=
2264                                         LOHA_EXISTS | (la->la_mode & S_IFMT);
2265                         if (rc == -ENOENT) {
2266                                 po->opo_non_exist = 1;
2267                                 rc = 0;
2268                         }
2269                 }
2270                 init_rwsem(&po->opo_sem);
2271         }
2272         RETURN(rc);
2273 }
2274
2275 /**
2276  * Implement OSP layer lu_object_operations::loo_object_free() interface.
2277  *
2278  * Finalize the object.
2279  *
2280  * If the OSP object has attributes cache, then destroy the cache.
2281  * Free the object finally.
2282  *
2283  * \param[in] env       pointer to the thread context
2284  * \param[in] o         pointer to the OSP layer lu_object
2285  */
2286 static void osp_object_free(const struct lu_env *env, struct lu_object *o)
2287 {
2288         struct osp_object       *obj = lu2osp_obj(o);
2289         struct lu_object_header *h = o->lo_header;
2290         struct osp_xattr_entry *oxe;
2291         struct osp_xattr_entry *tmp;
2292         int                     count;
2293
2294         dt_object_fini(&obj->opo_obj);
2295         lu_object_header_fini(h);
2296         list_for_each_entry_safe(oxe, tmp, &obj->opo_xattr_list, oxe_list) {
2297                 list_del(&oxe->oxe_list);
2298                 count = atomic_read(&oxe->oxe_ref);
2299                 LASSERTF(count == 1,
2300                          "Still has %d users on the xattr entry %.*s\n",
2301                          count-1, (int)oxe->oxe_namelen, oxe->oxe_buf);
2302
2303                 OBD_FREE(oxe, oxe->oxe_buflen);
2304         }
2305         OBD_SLAB_FREE_PTR(obj, osp_object_kmem);
2306 }
2307
2308 /**
2309  * Implement OSP layer lu_object_operations::loo_object_release() interface.
2310  *
2311  * Cleanup (not free) the object.
2312  *
2313  * If it is a reserved object but failed to be created, or it is an OST
2314  * object, then mark the object as non-cached.
2315  *
2316  * \param[in] env       pointer to the thread context
2317  * \param[in] o         pointer to the OSP layer lu_object
2318  */
2319 static void osp_object_release(const struct lu_env *env, struct lu_object *o)
2320 {
2321         struct osp_object       *po = lu2osp_obj(o);
2322         struct osp_device       *d  = lu2osp_dev(o->lo_dev);
2323
2324         ENTRY;
2325
2326         /*
2327          * release reservation if object was declared but not created
2328          * this may require lu_object_put() in LOD
2329          */
2330         if (unlikely(po->opo_reserved)) {
2331                 LASSERT(d->opd_pre != NULL);
2332                 LASSERT(d->opd_pre_reserved > 0);
2333                 spin_lock(&d->opd_pre_lock);
2334                 d->opd_pre_reserved--;
2335                 spin_unlock(&d->opd_pre_lock);
2336
2337                 /*
2338                  * Check that osp_precreate_cleanup_orphans is not blocked
2339                  * due to opd_pre_reserved > 0.
2340                  */
2341                 if (unlikely(d->opd_pre_reserved == 0 &&
2342                              (d->opd_pre_recovering || d->opd_pre_status)))
2343                         wake_up(&d->opd_pre_waitq);
2344
2345                 /* not needed in cache any more */
2346                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2347         }
2348
2349         if (is_ost_obj(o))
2350                 /* XXX: Currently, NOT cache OST-object on MDT because:
2351                  *      1. it is not often accessed on MDT.
2352                  *      2. avoid up layer (such as LFSCK) to load too many
2353                  *         once-used OST-objects. */
2354                 set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
2355
2356         EXIT;
2357 }
2358
2359 static int osp_object_print(const struct lu_env *env, void *cookie,
2360                             lu_printer_t p, const struct lu_object *l)
2361 {
2362         const struct osp_object *o = lu2osp_obj((struct lu_object *)l);
2363
2364         return (*p)(env, cookie, LUSTRE_OSP_NAME"-object@%p", o);
2365 }
2366
2367 static int osp_object_invariant(const struct lu_object *o)
2368 {
2369         LBUG();
2370 }
2371
2372 struct lu_object_operations osp_lu_obj_ops = {
2373         .loo_object_init        = osp_object_init,
2374         .loo_object_free        = osp_object_free,
2375         .loo_object_release     = osp_object_release,
2376         .loo_object_print       = osp_object_print,
2377         .loo_object_invariant   = osp_object_invariant
2378 };