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