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