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