Whamcloud - gitweb
b=21563 Metadata performance has degraded for some operations between 1.6.5 and 1.8.1
[fs/lustre-release.git] / lustre / osc / osc_create.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/osc/osc_create.c
37  * For testing and management it is treated as an obd_device,
38  * although * it does not export a full OBD method table (the
39  * requests are coming * in over the wire, so object target modules
40  * do not have a full * method table.)
41  *
42  * Author: Peter Braam <braam@clusterfs.com>
43  */
44
45 #ifndef EXPORT_SYMTAB
46 # define EXPORT_SYMTAB
47 #endif
48 #define DEBUG_SUBSYSTEM S_OSC
49
50 #ifdef __KERNEL__
51 # include <libcfs/libcfs.h>
52 #else /* __KERNEL__ */
53 # include <liblustre.h>
54 #endif
55
56 #ifdef  __CYGWIN__
57 # include <ctype.h>
58 #endif
59
60 #include <lustre_dlm.h>
61 #include <obd_class.h>
62 #include "osc_internal.h"
63
64 /* XXX need AT adjust ? */
65 #define osc_create_timeout      (obd_timeout / 2)
66
67 struct osc_create_async_args {
68         struct osc_creator      *rq_oscc;
69         struct lov_stripe_md    *rq_lsm;
70         struct obd_info         *rq_oinfo;
71 };
72
73 static int oscc_internal_create(struct osc_creator *oscc);
74 static int handle_async_create(struct ptlrpc_request *req, int rc);
75
76 static int osc_interpret_create(const struct lu_env *env,
77                                 struct ptlrpc_request *req, void *data, int rc)
78 {
79         struct osc_creator *oscc;
80         struct ost_body *body = NULL;
81         struct ptlrpc_request *fake_req, *pos;
82         ENTRY;
83
84         if (req->rq_repmsg) {
85                 body = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
86                 if (body == NULL && rc == 0)
87                         rc = -EPROTO;
88         }
89
90         oscc = req->rq_async_args.pointer_arg[0];
91         LASSERT(oscc && (oscc->oscc_obd != LP_POISON));
92
93         cfs_spin_lock(&oscc->oscc_lock);
94         oscc->oscc_flags &= ~OSCC_FLAG_CREATING;
95         switch (rc) {
96         case 0: {
97                 if (body) {
98                         int diff =ostid_id(&body->oa.o_oi)- oscc->oscc_last_id;
99
100                         /* oscc_internal_create() stores the original value of
101                          * grow_count in rq_async_args.space[0].
102                          * We can't compare against oscc_grow_count directly,
103                          * because it may have been increased while the RPC
104                          * is in flight, so we would always find ourselves
105                          * having created fewer objects and decreasing the
106                          * precreate request size.  b=18577 */
107                         if (diff < (int) req->rq_async_args.space[0]) {
108                                 /* the OST has not managed to create all the
109                                  * objects we asked for */
110                                 oscc->oscc_grow_count = max(diff,
111                                                             OST_MIN_PRECREATE);
112                                 /* don't bump grow_count next time */
113                                 oscc->oscc_flags |= OSCC_FLAG_LOW;
114                         } else {
115                                 /* the OST is able to keep up with the work,
116                                  * we could consider increasing grow_count
117                                  * next time if needed */
118                                 oscc->oscc_flags &= ~OSCC_FLAG_LOW;
119                         }
120                         oscc->oscc_last_id = ostid_id(&body->oa.o_oi);
121                 }
122                 cfs_spin_unlock(&oscc->oscc_lock);
123                 break;
124         }
125         case -EROFS:
126                 oscc->oscc_flags |= OSCC_FLAG_RDONLY;
127         case -ENOSPC:
128         case -EFBIG: 
129                 if (rc != -EROFS) {
130                         oscc->oscc_flags |= OSCC_FLAG_NOSPC;
131                         if (body && rc == -ENOSPC) {
132                                 oscc->oscc_last_id = body->oa.o_id;
133                                 oscc->oscc_grow_count = OST_MIN_PRECREATE;
134                         }
135                 }
136                 cfs_spin_unlock(&oscc->oscc_lock);
137                 DEBUG_REQ(D_INODE, req, "OST out of space, flagging");
138                 break;
139         case -EIO: {
140                 /* filter always set body->oa.o_id as the last_id
141                  * of filter (see filter_handle_precreate for detail)*/
142                 if (body && body->oa.o_id > oscc->oscc_last_id)
143                         oscc->oscc_last_id = body->oa.o_id;
144                 cfs_spin_unlock(&oscc->oscc_lock);
145                 break;
146         }
147         case -EINTR:
148         case -EWOULDBLOCK: {
149                 /* aka EAGAIN we should not delay create if import failed -
150                  * this avoid client stick in create and avoid race with
151                  * delorphan */
152                 /* EINTR say - old create request is killed due mds<>ost
153                  * eviction - OSCC_FLAG_RECOVERING can already set due
154                  * IMP_DISCONN event */
155                 oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
156                 /* oscc->oscc_grow_count = OST_MIN_PRECREATE; */
157                 cfs_spin_unlock(&oscc->oscc_lock);
158                 break;
159         }
160         default: {
161                 oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
162                 oscc->oscc_grow_count = OST_MIN_PRECREATE;
163                 cfs_spin_unlock(&oscc->oscc_lock);
164                 DEBUG_REQ(D_ERROR, req,
165                           "Unknown rc %d from async create: failing oscc", rc);
166                 ptlrpc_fail_import(req->rq_import,
167                                    lustre_msg_get_conn_cnt(req->rq_reqmsg));
168         }
169         }
170
171         CDEBUG(D_HA, "preallocated through id "LPU64" (next to use "LPU64")\n",
172                oscc->oscc_last_id, oscc->oscc_next_id);
173
174         cfs_spin_lock(&oscc->oscc_lock);
175         cfs_list_for_each_entry_safe(fake_req, pos,
176                                      &oscc->oscc_wait_create_list, rq_list) {
177                 if (handle_async_create(fake_req, rc)  == -EAGAIN) {
178                         oscc_internal_create(oscc);
179                         /* sending request should be never fail because
180                          * osc use preallocated requests pool */
181                         GOTO(exit_wakeup, rc);
182                 }
183         }
184         cfs_spin_unlock(&oscc->oscc_lock);
185
186 exit_wakeup:
187         cfs_waitq_signal(&oscc->oscc_waitq);
188         RETURN(rc);
189 }
190
191 static int oscc_internal_create(struct osc_creator *oscc)
192 {
193         struct ptlrpc_request *request;
194         struct ost_body *body;
195         ENTRY;
196
197         LASSERT_SPIN_LOCKED(&oscc->oscc_lock);
198
199         if ((oscc->oscc_flags & OSCC_FLAG_RECOVERING) ||
200             (oscc->oscc_flags & OSCC_FLAG_DEGRADED)) {
201                 cfs_spin_unlock(&oscc->oscc_lock);
202                 RETURN(0);
203         }
204
205         /* we need check it before OSCC_FLAG_CREATING - because need
206          * see lower number of precreate objects */
207         if (oscc->oscc_grow_count < oscc->oscc_max_grow_count &&
208             ((oscc->oscc_flags & OSCC_FLAG_LOW) == 0) &&
209             (__s64)(oscc->oscc_last_id - oscc->oscc_next_id) <=
210                    (oscc->oscc_grow_count / 4 + 1)) {
211                 oscc->oscc_flags |= OSCC_FLAG_LOW;
212                 oscc->oscc_grow_count *= 2;
213         }
214
215         if (oscc->oscc_flags & OSCC_FLAG_CREATING) {
216                 cfs_spin_unlock(&oscc->oscc_lock);
217                 RETURN(0);
218         }
219
220         if (oscc->oscc_grow_count > oscc->oscc_max_grow_count / 2)
221                 oscc->oscc_grow_count = oscc->oscc_max_grow_count / 2;
222
223         oscc->oscc_flags |= OSCC_FLAG_CREATING;
224         cfs_spin_unlock(&oscc->oscc_lock);
225
226         request = ptlrpc_request_alloc_pack(oscc->oscc_obd->u.cli.cl_import,
227                                             &RQF_OST_CREATE,
228                                             LUSTRE_OST_VERSION, OST_CREATE);
229         if (request == NULL) {
230                 cfs_spin_lock(&oscc->oscc_lock);
231                 oscc->oscc_flags &= ~OSCC_FLAG_CREATING;
232                 cfs_spin_unlock(&oscc->oscc_lock);
233                 RETURN(-ENOMEM);
234         }
235
236         request->rq_request_portal = OST_CREATE_PORTAL;
237         ptlrpc_at_set_req_timeout(request);
238         body = req_capsule_client_get(&request->rq_pill, &RMF_OST_BODY);
239
240         cfs_spin_lock(&oscc->oscc_lock);
241
242         if (likely(fid_seq_is_mdt(oscc->oscc_oa.o_seq))) {
243                 body->oa.o_oi.oi_seq = oscc->oscc_oa.o_seq;
244                 body->oa.o_oi.oi_id  = oscc->oscc_last_id +
245                                        oscc->oscc_grow_count;
246         } else {
247                 /*Just warning here currently, since not sure how fid-on-ost
248                  *will be implemented here */
249                 CWARN("o_seq: "LPU64" is not indicate any MDTs.\n",
250                        oscc->oscc_oa.o_seq);
251         }
252
253         body->oa.o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
254         request->rq_async_args.space[0] = oscc->oscc_grow_count;
255         cfs_spin_unlock(&oscc->oscc_lock);
256         CDEBUG(D_RPCTRACE, "prealloc through id "LPU64" (last seen "LPU64")\n",
257                body->oa.o_id, oscc->oscc_last_id);
258
259         /* we should not resend create request - anyway we will have delorphan
260          * and kill these objects */
261         request->rq_no_delay = request->rq_no_resend = 1;
262         ptlrpc_request_set_replen(request);
263
264         request->rq_async_args.pointer_arg[0] = oscc;
265         request->rq_interpret_reply = osc_interpret_create;
266         ptlrpcd_add_req(request, PSCOPE_OTHER);
267
268         RETURN(0);
269 }
270
271 static int oscc_has_objects_nolock(struct osc_creator *oscc, int count)
272 {
273         return ((__s64)(oscc->oscc_last_id - oscc->oscc_next_id) >= count);
274 }
275
276
277 static int oscc_has_objects(struct osc_creator *oscc, int count)
278 {
279         int have_objs;
280
281         cfs_spin_lock(&oscc->oscc_lock);
282         have_objs = oscc_has_objects_nolock(oscc, count);
283         cfs_spin_unlock(&oscc->oscc_lock);
284
285         return have_objs;
286 }
287
288 static int oscc_wait_for_objects(struct osc_creator *oscc, int count)
289 {
290         int have_objs;
291         int ost_unusable;
292
293         ost_unusable = oscc->oscc_obd->u.cli.cl_import->imp_invalid;
294
295         cfs_spin_lock(&oscc->oscc_lock);
296         ost_unusable |= (OSCC_FLAG_NOSPC | OSCC_FLAG_RDONLY |
297                          OSCC_FLAG_EXITING) & oscc->oscc_flags;
298         have_objs = oscc_has_objects_nolock(oscc, count);
299
300         if (!ost_unusable && !have_objs)
301                 /* they release lock himself */
302                 have_objs = oscc_internal_create(oscc);
303         else
304                 cfs_spin_unlock(&oscc->oscc_lock);
305
306         return have_objs || ost_unusable;
307 }
308
309 static int oscc_precreate(struct osc_creator *oscc)
310 {
311         struct l_wait_info lwi;
312         int rc = 0;
313         ENTRY;
314
315         if (oscc_has_objects(oscc, oscc->oscc_grow_count / 2))
316                 RETURN(0);
317
318         /* we should be not block forever - because client's create rpc can
319          * stick in mds for long time and forbid client reconnect */
320         lwi = LWI_TIMEOUT(cfs_timeout_cap(cfs_time_seconds(osc_create_timeout)),
321                           NULL, NULL);
322
323         rc = l_wait_event(oscc->oscc_waitq, oscc_wait_for_objects(oscc, 1), &lwi);
324         RETURN(rc);
325 }
326
327 static int oscc_in_sync(struct osc_creator *oscc)
328 {
329         int sync;
330
331         cfs_spin_lock(&oscc->oscc_lock);
332         sync = oscc->oscc_flags & OSCC_FLAG_SYNC_IN_PROGRESS;
333         cfs_spin_unlock(&oscc->oscc_lock);
334
335         return sync;
336 }
337
338 /* decide if the OST has remaining object, return value :
339         0 : the OST has remaining object, and don't need to do precreate.
340         1 : the OST has no remaining object, and will send a RPC for precreate.
341         2 : the OST has no remaining object, and will not get any for
342             a potentially very long time
343      1000 : unusable
344  */
345 int osc_precreate(struct obd_export *exp)
346 {
347         struct osc_creator *oscc = &exp->exp_obd->u.cli.cl_oscc;
348         struct obd_import *imp = exp->exp_imp_reverse;
349         ENTRY;
350
351         LASSERT(oscc != NULL);
352         if (imp != NULL && imp->imp_deactive)
353                 RETURN(1000);
354
355         /* Handle critical states first */
356         cfs_spin_lock(&oscc->oscc_lock);
357         if (oscc->oscc_flags & OSCC_FLAG_NOSPC ||
358             oscc->oscc_flags & OSCC_FLAG_RDONLY ||
359             oscc->oscc_flags & OSCC_FLAG_EXITING) {
360                 cfs_spin_unlock(&oscc->oscc_lock);
361                 RETURN(1000);
362         }
363
364         /* Do not check for a degraded OST here - bug21563/bug18539 */
365         if (oscc->oscc_flags & OSCC_FLAG_RECOVERING) {
366                 cfs_spin_unlock(&oscc->oscc_lock);
367                 RETURN(2);
368         }
369
370         if (oscc_has_objects_nolock(oscc, oscc->oscc_grow_count / 2)) {
371                 cfs_spin_unlock(&oscc->oscc_lock);
372                 RETURN(0);
373         }
374
375         /* Do not check for OSCC_FLAG_CREATING flag here, let
376          * osc_precreate() call oscc_internal_create() and
377          * adjust oscc_grow_count bug21563 */
378         if (oscc->oscc_flags & OSCC_FLAG_SYNC_IN_PROGRESS) {
379                 cfs_spin_unlock(&oscc->oscc_lock);
380                 RETURN(1);
381         }
382
383         if (oscc_internal_create(oscc))
384                 RETURN(1000);
385         RETURN(1);
386 }
387
388 static int handle_async_create(struct ptlrpc_request *req, int rc)
389 {
390         struct osc_create_async_args *args = ptlrpc_req_async_args(req);
391         struct osc_creator    *oscc = args->rq_oscc;
392         struct lov_stripe_md  *lsm  = args->rq_lsm;
393         struct obd_info       *oinfo = args->rq_oinfo;
394         struct obdo           *oa = oinfo->oi_oa;
395
396         LASSERT_SPIN_LOCKED(&oscc->oscc_lock);
397
398         if(rc)
399                 GOTO(out_wake, rc);
400
401         /* Handle the critical type errors first.
402          * Should we also test cl_import state as well ? */
403         if (oscc->oscc_flags & OSCC_FLAG_EXITING)
404                 GOTO(out_wake, rc = -EIO);
405
406         if (oscc->oscc_flags & OSCC_FLAG_NOSPC)
407                 GOTO(out_wake, rc = -ENOSPC);
408
409         if (oscc->oscc_flags & OSCC_FLAG_RDONLY)
410                 GOTO(out_wake, rc = -EROFS);
411
412         /* should be try wait until recovery finished */
413         if((oscc->oscc_flags & OSCC_FLAG_RECOVERING) ||
414            (oscc->oscc_flags & OSCC_FLAG_DEGRADED))
415                 RETURN(-EAGAIN);
416
417         if (oscc_has_objects_nolock(oscc, 1)) {
418                 memcpy(oa, &oscc->oscc_oa, sizeof(*oa));
419                 oa->o_id = oscc->oscc_next_id;
420                 lsm->lsm_object_id = oscc->oscc_next_id;
421                 oscc->oscc_next_id++;
422
423                 CDEBUG(D_RPCTRACE, " set oscc_next_id = "LPU64"\n",
424                        oscc->oscc_next_id);
425                 GOTO(out_wake, rc = 0);
426         }
427
428         /* we don't have objects now - continue wait */
429         RETURN(-EAGAIN);
430
431 out_wake:
432
433         rc = oinfo->oi_cb_up(oinfo, rc);
434         ptlrpc_fakereq_finished(req);
435
436         RETURN(rc);
437 }
438
439 static int async_create_interpret(const struct lu_env *env,
440                                   struct ptlrpc_request *req, void *data,
441                                   int rc)
442 {
443         struct osc_create_async_args *args = ptlrpc_req_async_args(req);
444         struct osc_creator    *oscc = args->rq_oscc;
445         int ret;
446
447         cfs_spin_lock(&oscc->oscc_lock);
448         ret = handle_async_create(req, rc);
449         cfs_spin_unlock(&oscc->oscc_lock);
450
451         return ret;
452 }
453
454 int osc_create_async(struct obd_export *exp, struct obd_info *oinfo,
455                      struct lov_stripe_md **ea, struct obd_trans_info *oti)
456 {
457         int rc;
458         struct ptlrpc_request *fake_req;
459         struct osc_create_async_args *args;
460         struct osc_creator *oscc = &exp->exp_obd->u.cli.cl_oscc;
461         struct obdo *oa = oinfo->oi_oa;
462         ENTRY;
463
464         if ((oa->o_valid & OBD_MD_FLGROUP) && !fid_seq_is_mdt(oa->o_seq)) {
465                 rc = osc_real_create(exp, oinfo->oi_oa, ea, oti);
466                 rc = oinfo->oi_cb_up(oinfo, rc);
467                 RETURN(rc);
468         }
469
470         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
471             oa->o_flags == OBD_FL_RECREATE_OBJS) {
472                 rc = osc_real_create(exp, oinfo->oi_oa, ea, oti);
473                 rc = oinfo->oi_cb_up(oinfo, rc);
474                 RETURN(rc);
475         }
476
477         LASSERT((*ea) != NULL);
478
479         fake_req = ptlrpc_prep_fakereq(oscc->oscc_obd->u.cli.cl_import,
480                                        osc_create_timeout,
481                                        async_create_interpret);
482         if (fake_req == NULL) {
483                 rc = oinfo->oi_cb_up(oinfo, -ENOMEM);
484                 RETURN(-ENOMEM);
485         }
486
487         args = ptlrpc_req_async_args(fake_req);
488         CLASSERT(sizeof(*args) <= sizeof(fake_req->rq_async_args));
489
490         args->rq_oscc  = oscc;
491         args->rq_lsm   = *ea;
492         args->rq_oinfo = oinfo;
493
494         cfs_spin_lock(&oscc->oscc_lock);
495         /* try fast path */
496         rc = handle_async_create(fake_req, 0);
497         if (rc == -EAGAIN) {
498                 int is_add;
499                 /* we not have objects - try wait */
500                 is_add = ptlrpcd_add_req(fake_req, PSCOPE_OTHER);
501                 if (!is_add)
502                         cfs_list_add(&fake_req->rq_list,
503                                      &oscc->oscc_wait_create_list);
504                 else
505                         rc = is_add;
506         }
507         cfs_spin_unlock(&oscc->oscc_lock);
508
509         if (rc != -EAGAIN)
510                 /* need free request if was error hit or
511                  * objects already allocated */
512                 ptlrpc_req_finished(fake_req);
513         else
514                 /* EAGAIN mean - request is delayed */
515                 rc = 0;
516
517         RETURN(rc);
518 }
519
520 int osc_create(struct obd_export *exp, struct obdo *oa,
521                struct lov_stripe_md **ea, struct obd_trans_info *oti)
522 {
523         struct osc_creator *oscc = &exp->exp_obd->u.cli.cl_oscc;
524         struct obd_import  *imp  = exp->exp_obd->u.cli.cl_import;
525         struct lov_stripe_md *lsm;
526         int del_orphan = 0, rc = 0;
527         ENTRY;
528
529         LASSERT(oa);
530         LASSERT(ea);
531         LASSERT(oa->o_valid & OBD_MD_FLGROUP);
532
533         if ((oa->o_valid & OBD_MD_FLFLAGS) &&
534             oa->o_flags == OBD_FL_RECREATE_OBJS) {
535                 RETURN(osc_real_create(exp, oa, ea, oti));
536         }
537
538         if (!fid_seq_is_mdt(oa->o_seq))
539                 RETURN(osc_real_create(exp, oa, ea, oti));
540
541         /* this is the special case where create removes orphans */
542         if (oa->o_valid & OBD_MD_FLFLAGS &&
543             oa->o_flags == OBD_FL_DELORPHAN) {
544                 cfs_spin_lock(&oscc->oscc_lock);
545                 if (oscc->oscc_flags & OSCC_FLAG_SYNC_IN_PROGRESS) {
546                         cfs_spin_unlock(&oscc->oscc_lock);
547                         RETURN(-EBUSY);
548                 }
549                 if (!(oscc->oscc_flags & OSCC_FLAG_RECOVERING)) {
550                         cfs_spin_unlock(&oscc->oscc_lock);
551                         RETURN(0);
552                 }
553
554                 oscc->oscc_flags |= OSCC_FLAG_SYNC_IN_PROGRESS;
555                 /* seting flag LOW we prevent extra grow precreate size
556                  * and enforce use last assigned size */
557                 oscc->oscc_flags |= OSCC_FLAG_LOW;
558                 cfs_spin_unlock(&oscc->oscc_lock);
559                 CDEBUG(D_HA, "%s: oscc recovery started - delete to "LPU64"\n",
560                        oscc->oscc_obd->obd_name, oscc->oscc_next_id - 1);
561
562                 del_orphan = 1;
563
564                 /* delete from next_id on up */
565                 oa->o_valid |= OBD_MD_FLID;
566                 oa->o_id = oscc->oscc_next_id - 1;
567
568                 rc = osc_real_create(exp, oa, ea, NULL);
569
570                 cfs_spin_lock(&oscc->oscc_lock);
571                 oscc->oscc_flags &= ~OSCC_FLAG_SYNC_IN_PROGRESS;
572                 if (rc == 0 || rc == -ENOSPC) {
573                         struct obd_connect_data *ocd;
574
575                         if (rc == -ENOSPC)
576                                 oscc->oscc_flags |= OSCC_FLAG_NOSPC;
577                         oscc->oscc_flags &= ~OSCC_FLAG_RECOVERING;
578
579                         oscc->oscc_last_id = oa->o_id;
580                         ocd = &imp->imp_connect_data;
581                         if (ocd->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN) {
582                                 CDEBUG(D_HA, "%s: Skip orphan set, reset last "
583                                        "objid\n", oscc->oscc_obd->obd_name);
584                                 oscc->oscc_next_id = oa->o_id + 1;
585                         }
586
587                         /* sanity check for next objid. see bug 17025 */
588                         LASSERT(oscc->oscc_next_id == oa->o_id + 1);
589
590                         CDEBUG(D_HA, "%s: oscc recovery finished, last_id: "
591                                LPU64", rc: %d\n", oscc->oscc_obd->obd_name,
592                                oscc->oscc_last_id, rc);
593                 } else {
594                         CDEBUG(D_ERROR, "%s: oscc recovery failed: %d\n",
595                                oscc->oscc_obd->obd_name, rc);
596                 }
597
598                 cfs_waitq_signal(&oscc->oscc_waitq);
599                 cfs_spin_unlock(&oscc->oscc_lock);
600
601                 if (rc < 0)
602                         RETURN(rc);
603         }
604
605         lsm = *ea;
606         if (lsm == NULL) {
607                 rc = obd_alloc_memmd(exp, &lsm);
608                 if (rc < 0)
609                         RETURN(rc);
610         }
611
612         while (1) {
613                 if (oscc_in_sync(oscc))
614                         CDEBUG(D_HA,"%s: oscc recovery in progress, waiting\n",
615                                oscc->oscc_obd->obd_name);
616
617                 rc = oscc_precreate(oscc);
618                 if (rc)
619                         CDEBUG(D_HA,"%s: error create %d\n",
620                                oscc->oscc_obd->obd_name, rc);
621
622                 cfs_spin_lock(&oscc->oscc_lock);
623
624                 /* wakeup but recovery did not finished */
625                 if ((oscc->oscc_obd->u.cli.cl_import->imp_invalid) ||
626                     (oscc->oscc_flags & OSCC_FLAG_RECOVERING)) {
627                         rc = -EIO;
628                         cfs_spin_unlock(&oscc->oscc_lock);
629                         break;
630                 }
631
632                 if (oscc->oscc_flags & OSCC_FLAG_NOSPC) {
633                         rc = -ENOSPC;
634                         cfs_spin_unlock(&oscc->oscc_lock);
635                         break;
636                 }
637
638                 if (oscc->oscc_flags & OSCC_FLAG_RDONLY) {
639                         rc = -EROFS;
640                         cfs_spin_unlock(&oscc->oscc_lock);
641                         break;
642                 }
643
644                 // Should we report -EIO error ?
645                 if (oscc->oscc_flags & OSCC_FLAG_EXITING) {
646                         cfs_spin_unlock(&oscc->oscc_lock);
647                         break;
648                 }
649
650                 if (oscc_has_objects_nolock(oscc, 1)) {
651                         memcpy(oa, &oscc->oscc_oa, sizeof(*oa));
652                         oa->o_id = oscc->oscc_next_id;
653                         lsm->lsm_object_id = oscc->oscc_next_id;
654                         *ea = lsm;
655                         oscc->oscc_next_id++;
656                         cfs_spin_unlock(&oscc->oscc_lock);
657
658                         CDEBUG(D_RPCTRACE, "%s: set oscc_next_id = "LPU64"\n",
659                                exp->exp_obd->obd_name, oscc->oscc_next_id);
660                         break;
661                 }
662
663                 cfs_spin_unlock(&oscc->oscc_lock);
664         }
665
666         if (rc == 0) {
667                 CDEBUG(D_INFO, "%s: returning objid "LPU64"\n",
668                        obd2cli_tgt(oscc->oscc_obd), lsm->lsm_object_id);
669         } else {
670                 if (*ea == NULL)
671                         obd_free_memmd(exp, &lsm);
672                 if (del_orphan != 0 && rc != -EIO)
673                         /* Ignore non-IO precreate error for clear orphan */
674                         rc = 0;
675         }
676         RETURN(rc);
677 }
678
679 void oscc_init(struct obd_device *obd)
680 {
681         struct osc_creator *oscc;
682
683         if (obd == NULL)
684                 return;
685
686         oscc = &obd->u.cli.cl_oscc;
687
688         memset(oscc, 0, sizeof(*oscc));
689
690         cfs_waitq_init(&oscc->oscc_waitq);
691         cfs_spin_lock_init(&oscc->oscc_lock);
692         oscc->oscc_obd = obd;
693         oscc->oscc_grow_count = OST_MIN_PRECREATE;
694         oscc->oscc_max_grow_count = OST_MAX_PRECREATE;
695
696         oscc->oscc_next_id = 2;
697         oscc->oscc_last_id = 1;
698         oscc->oscc_flags |= OSCC_FLAG_RECOVERING;
699
700         CFS_INIT_LIST_HEAD(&oscc->oscc_wait_create_list);
701
702         /* XXX the export handle should give the oscc the last object */
703         /* oed->oed_oscc.oscc_last_id = exph->....; */
704 }
705
706 void oscc_fini(struct obd_device *obd)
707 {
708         struct osc_creator *oscc = &obd->u.cli.cl_oscc;
709         ENTRY;
710
711
712         cfs_spin_lock(&oscc->oscc_lock);
713         oscc->oscc_flags &= ~OSCC_FLAG_RECOVERING;
714         oscc->oscc_flags |= OSCC_FLAG_EXITING;
715         cfs_spin_unlock(&oscc->oscc_lock);
716 }