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