Whamcloud - gitweb
LU-5099 api: transfer object type via dt_insert API
[fs/lustre-release.git] / lustre / obdclass / local_storage.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,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2012, 2013, Intel Corporation.
24  */
25 /*
26  * lustre/obdclass/local_storage.c
27  *
28  * Local storage for file/objects with fid generation. Works on top of OSD.
29  *
30  * Author: Mikhail Pershin <mike.pershin@intel.com>
31  */
32
33 #define DEBUG_SUBSYSTEM S_CLASS
34
35 #include "local_storage.h"
36
37 /* all initialized local storages on this node are linked on this */
38 static CFS_LIST_HEAD(ls_list_head);
39 static DEFINE_MUTEX(ls_list_mutex);
40
41 static int ls_object_init(const struct lu_env *env, struct lu_object *o,
42                           const struct lu_object_conf *unused)
43 {
44         struct ls_device        *ls;
45         struct lu_object        *below;
46         struct lu_device        *under;
47
48         ENTRY;
49
50         ls = container_of0(o->lo_dev, struct ls_device, ls_top_dev.dd_lu_dev);
51         under = &ls->ls_osd->dd_lu_dev;
52         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
53         if (below == NULL)
54                 RETURN(-ENOMEM);
55
56         lu_object_add(o, below);
57
58         RETURN(0);
59 }
60
61 static void ls_object_free(const struct lu_env *env, struct lu_object *o)
62 {
63         struct ls_object        *obj = lu2ls_obj(o);
64         struct lu_object_header *h = o->lo_header;
65
66         dt_object_fini(&obj->ls_obj);
67         lu_object_header_fini(h);
68         OBD_FREE_PTR(obj);
69 }
70
71 struct lu_object_operations ls_lu_obj_ops = {
72         .loo_object_init  = ls_object_init,
73         .loo_object_free  = ls_object_free,
74 };
75
76 struct lu_object *ls_object_alloc(const struct lu_env *env,
77                                   const struct lu_object_header *_h,
78                                   struct lu_device *d)
79 {
80         struct lu_object_header *h;
81         struct ls_object        *o;
82         struct lu_object        *l;
83
84         LASSERT(_h == NULL);
85
86         OBD_ALLOC_PTR(o);
87         if (o != NULL) {
88                 l = &o->ls_obj.do_lu;
89                 h = &o->ls_header;
90
91                 lu_object_header_init(h);
92                 dt_object_init(&o->ls_obj, h, d);
93                 lu_object_add_top(h, l);
94
95                 l->lo_ops = &ls_lu_obj_ops;
96
97                 return l;
98         } else {
99                 return NULL;
100         }
101 }
102
103 static struct lu_device_operations ls_lu_dev_ops = {
104         .ldo_object_alloc =     ls_object_alloc
105 };
106
107 static struct ls_device *__ls_find_dev(struct dt_device *dev)
108 {
109         struct ls_device *ls, *ret = NULL;
110
111         cfs_list_for_each_entry(ls, &ls_list_head, ls_linkage) {
112                 if (ls->ls_osd == dev) {
113                         atomic_inc(&ls->ls_refcount);
114                         ret = ls;
115                         break;
116                 }
117         }
118         return ret;
119 }
120
121 struct ls_device *ls_find_dev(struct dt_device *dev)
122 {
123         struct ls_device *ls;
124
125         mutex_lock(&ls_list_mutex);
126         ls = __ls_find_dev(dev);
127         mutex_unlock(&ls_list_mutex);
128
129         return ls;
130 }
131
132 static struct lu_device_type_operations ls_device_type_ops = {
133         .ldto_start = NULL,
134         .ldto_stop  = NULL,
135 };
136
137 static struct lu_device_type ls_lu_type = {
138         .ldt_name = "local_storage",
139         .ldt_ops  = &ls_device_type_ops,
140 };
141
142 struct ls_device *ls_device_get(struct dt_device *dev)
143 {
144         struct ls_device *ls;
145
146         ENTRY;
147
148         mutex_lock(&ls_list_mutex);
149         ls = __ls_find_dev(dev);
150         if (ls)
151                 GOTO(out_ls, ls);
152
153         /* not found, then create */
154         OBD_ALLOC_PTR(ls);
155         if (ls == NULL)
156                 GOTO(out_ls, ls = ERR_PTR(-ENOMEM));
157
158         atomic_set(&ls->ls_refcount, 1);
159         CFS_INIT_LIST_HEAD(&ls->ls_los_list);
160         mutex_init(&ls->ls_los_mutex);
161
162         ls->ls_osd = dev;
163
164         LASSERT(dev->dd_lu_dev.ld_site);
165         lu_device_init(&ls->ls_top_dev.dd_lu_dev, &ls_lu_type);
166         ls->ls_top_dev.dd_lu_dev.ld_ops = &ls_lu_dev_ops;
167         ls->ls_top_dev.dd_lu_dev.ld_site = dev->dd_lu_dev.ld_site;
168
169         /* finally add ls to the list */
170         cfs_list_add(&ls->ls_linkage, &ls_list_head);
171 out_ls:
172         mutex_unlock(&ls_list_mutex);
173         RETURN(ls);
174 }
175
176 void ls_device_put(const struct lu_env *env, struct ls_device *ls)
177 {
178         LASSERT(env);
179         if (!atomic_dec_and_test(&ls->ls_refcount))
180                 return;
181
182         mutex_lock(&ls_list_mutex);
183         if (atomic_read(&ls->ls_refcount) == 0) {
184                 LASSERT(cfs_list_empty(&ls->ls_los_list));
185                 cfs_list_del(&ls->ls_linkage);
186                 lu_site_purge(env, ls->ls_top_dev.dd_lu_dev.ld_site, ~0);
187                 lu_device_fini(&ls->ls_top_dev.dd_lu_dev);
188                 OBD_FREE_PTR(ls);
189         }
190         mutex_unlock(&ls_list_mutex);
191 }
192
193 /**
194  * local file fid generation
195  */
196 int local_object_fid_generate(const struct lu_env *env,
197                               struct local_oid_storage *los,
198                               struct lu_fid *fid)
199 {
200         LASSERT(los->los_dev);
201         LASSERT(los->los_obj);
202
203         /* take next OID */
204
205         /* to make it unique after reboot we store
206          * the latest generated fid atomically with
207          * object creation see local_object_create() */
208
209         mutex_lock(&los->los_id_lock);
210         fid->f_seq = los->los_seq;
211         fid->f_oid = ++los->los_last_oid;
212         fid->f_ver = 0;
213         mutex_unlock(&los->los_id_lock);
214
215         return 0;
216 }
217
218 int local_object_declare_create(const struct lu_env *env,
219                                 struct local_oid_storage *los,
220                                 struct dt_object *o, struct lu_attr *attr,
221                                 struct dt_object_format *dof,
222                                 struct thandle *th)
223 {
224         struct dt_thread_info   *dti = dt_info(env);
225         int                      rc;
226
227         ENTRY;
228
229         /* update fid generation file */
230         if (los != NULL) {
231                 LASSERT(dt_object_exists(los->los_obj));
232                 dti->dti_lb.lb_buf = NULL;
233                 dti->dti_lb.lb_len = sizeof(struct los_ondisk);
234                 rc = dt_declare_record_write(env, los->los_obj,
235                                              &dti->dti_lb, 0, th);
236                 if (rc)
237                         RETURN(rc);
238         }
239
240         rc = dt_declare_create(env, o, attr, NULL, dof, th);
241         if (rc)
242                 RETURN(rc);
243
244         dti->dti_lb.lb_buf = NULL;
245         dti->dti_lb.lb_len = sizeof(dti->dti_lma);
246         rc = dt_declare_xattr_set(env, o, &dti->dti_lb, XATTR_NAME_LMA, 0, th);
247
248         RETURN(rc);
249 }
250
251 int local_object_create(const struct lu_env *env,
252                         struct local_oid_storage *los,
253                         struct dt_object *o, struct lu_attr *attr,
254                         struct dt_object_format *dof, struct thandle *th)
255 {
256         struct dt_thread_info   *dti = dt_info(env);
257         obd_id                   lastid;
258         int                      rc;
259
260         ENTRY;
261
262         rc = dt_create(env, o, attr, NULL, dof, th);
263         if (rc)
264                 RETURN(rc);
265
266         if (los == NULL)
267                 RETURN(rc);
268
269         LASSERT(los->los_obj);
270         LASSERT(dt_object_exists(los->los_obj));
271
272         /* many threads can be updated this, serialize
273          * them here to avoid the race where one thread
274          * takes the value first, but writes it last */
275         mutex_lock(&los->los_id_lock);
276
277         /* update local oid number on disk so that
278          * we know the last one used after reboot */
279         lastid = cpu_to_le64(los->los_last_oid);
280
281         dti->dti_off = 0;
282         dti->dti_lb.lb_buf = &lastid;
283         dti->dti_lb.lb_len = sizeof(lastid);
284         rc = dt_record_write(env, los->los_obj, &dti->dti_lb, &dti->dti_off,
285                              th);
286         mutex_unlock(&los->los_id_lock);
287
288         RETURN(rc);
289 }
290
291 /*
292  * Create local named object (file, directory or index) in parent directory.
293  */
294 struct dt_object *__local_file_create(const struct lu_env *env,
295                                       const struct lu_fid *fid,
296                                       struct local_oid_storage *los,
297                                       struct ls_device *ls,
298                                       struct dt_object *parent,
299                                       const char *name, struct lu_attr *attr,
300                                       struct dt_object_format *dof)
301 {
302         struct dt_thread_info   *dti    = dt_info(env);
303         struct lu_object_conf   *conf   = &dti->dti_conf;
304         struct dt_insert_rec    *rec    = &dti->dti_dt_rec;
305         struct dt_object        *dto;
306         struct thandle          *th;
307         int                      rc;
308
309         /* We know that the target object does not exist, to be created,
310          * then give some hints - LOC_F_NEW to help low layer to handle
311          * that efficiently and properly. */
312         memset(conf, 0, sizeof(*conf));
313         conf->loc_flags = LOC_F_NEW;
314         dto = ls_locate(env, ls, fid, conf);
315         if (unlikely(IS_ERR(dto)))
316                 RETURN(dto);
317
318         LASSERT(dto != NULL);
319         if (dt_object_exists(dto))
320                 GOTO(out, rc = -EEXIST);
321
322         th = dt_trans_create(env, ls->ls_osd);
323         if (IS_ERR(th))
324                 GOTO(out, rc = PTR_ERR(th));
325
326         rc = local_object_declare_create(env, los, dto, attr, dof, th);
327         if (rc)
328                 GOTO(trans_stop, rc);
329
330         if (dti->dti_dof.dof_type == DFT_DIR) {
331                 dt_declare_ref_add(env, dto, th);
332                 dt_declare_ref_add(env, parent, th);
333         }
334
335         rec->rec_fid = fid;
336         rec->rec_type = dto->do_lu.lo_header->loh_attr;
337         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
338                                (const struct dt_key *)name, th);
339         if (rc)
340                 GOTO(trans_stop, rc);
341
342         rc = dt_trans_start_local(env, ls->ls_osd, th);
343         if (rc)
344                 GOTO(trans_stop, rc);
345
346         dt_write_lock(env, dto, 0);
347         if (dt_object_exists(dto))
348                 GOTO(unlock, rc = 0);
349
350         CDEBUG(D_OTHER, "create new object "DFID"\n",
351                PFID(lu_object_fid(&dto->do_lu)));
352         rc = local_object_create(env, los, dto, attr, dof, th);
353         if (rc)
354                 GOTO(unlock, rc);
355         LASSERT(dt_object_exists(dto));
356
357         if (dti->dti_dof.dof_type == DFT_DIR) {
358                 if (!dt_try_as_dir(env, dto))
359                         GOTO(destroy, rc = -ENOTDIR);
360
361                 rec->rec_type = S_IFDIR;
362                 rec->rec_fid = fid;
363                 /* Add "." and ".." for newly created dir */
364                 rc = dt_insert(env, dto, (const struct dt_rec *)rec,
365                                (const struct dt_key *)".", th, BYPASS_CAPA, 1);
366                 if (rc != 0)
367                         GOTO(destroy, rc);
368
369                 dt_ref_add(env, dto, th);
370                 rec->rec_fid = lu_object_fid(&parent->do_lu);
371                 rc = dt_insert(env, dto, (const struct dt_rec *)rec,
372                                (const struct dt_key *)"..", th, BYPASS_CAPA, 1);
373                 if (rc != 0)
374                         GOTO(destroy, rc);
375         }
376
377         rec->rec_fid = fid;
378         rec->rec_type = dto->do_lu.lo_header->loh_attr;
379         dt_write_lock(env, parent, 0);
380         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
381                        (const struct dt_key *)name, th, BYPASS_CAPA, 1);
382         if (dti->dti_dof.dof_type == DFT_DIR)
383                 dt_ref_add(env, parent, th);
384         dt_write_unlock(env, parent);
385         if (rc)
386                 GOTO(destroy, rc);
387 destroy:
388         if (rc)
389                 dt_destroy(env, dto, th);
390 unlock:
391         dt_write_unlock(env, dto);
392 trans_stop:
393         dt_trans_stop(env, ls->ls_osd, th);
394 out:
395         if (rc) {
396                 lu_object_put_nocache(env, &dto->do_lu);
397                 dto = ERR_PTR(rc);
398         }
399         RETURN(dto);
400 }
401
402 /*
403  * Look up and create (if it does not exist) a local named file or directory in
404  * parent directory.
405  */
406 struct dt_object *local_file_find_or_create(const struct lu_env *env,
407                                             struct local_oid_storage *los,
408                                             struct dt_object *parent,
409                                             const char *name, __u32 mode)
410 {
411         struct dt_thread_info   *dti = dt_info(env);
412         struct dt_object        *dto;
413         int                      rc;
414
415         LASSERT(parent);
416
417         rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
418         if (rc == 0)
419                 /* name is found, get the object */
420                 dto = ls_locate(env, dt2ls_dev(los->los_dev),
421                                 &dti->dti_fid, NULL);
422         else if (rc != -ENOENT)
423                 dto = ERR_PTR(rc);
424         else {
425                 rc = local_object_fid_generate(env, los, &dti->dti_fid);
426                 if (rc < 0) {
427                         dto = ERR_PTR(rc);
428                 } else {
429                         /* create the object */
430                         dti->dti_attr.la_valid  = LA_MODE;
431                         dti->dti_attr.la_mode   = mode;
432                         dti->dti_dof.dof_type   = dt_mode_to_dft(mode & S_IFMT);
433                         dto = __local_file_create(env, &dti->dti_fid, los,
434                                                   dt2ls_dev(los->los_dev),
435                                                   parent, name, &dti->dti_attr,
436                                                   &dti->dti_dof);
437                 }
438         }
439         return dto;
440 }
441 EXPORT_SYMBOL(local_file_find_or_create);
442
443 struct dt_object *local_file_find_or_create_with_fid(const struct lu_env *env,
444                                                      struct dt_device *dt,
445                                                      const struct lu_fid *fid,
446                                                      struct dt_object *parent,
447                                                      const char *name,
448                                                      __u32 mode)
449 {
450         struct dt_thread_info   *dti = dt_info(env);
451         struct dt_object        *dto;
452         int                      rc;
453
454         LASSERT(parent);
455
456         rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
457         if (rc == 0) {
458                 dto = dt_locate(env, dt, &dti->dti_fid);
459         } else if (rc != -ENOENT) {
460                 dto = ERR_PTR(rc);
461         } else {
462                 struct ls_device *ls;
463
464                 ls = ls_device_get(dt);
465                 if (IS_ERR(ls)) {
466                         dto = ERR_PTR(PTR_ERR(ls));
467                 } else {
468                         /* create the object */
469                         dti->dti_attr.la_valid  = LA_MODE;
470                         dti->dti_attr.la_mode   = mode;
471                         dti->dti_dof.dof_type   = dt_mode_to_dft(mode & S_IFMT);
472                         dto = __local_file_create(env, fid, NULL, ls, parent,
473                                                   name, &dti->dti_attr,
474                                                   &dti->dti_dof);
475                         /* ls_device_put() will finalize the ls device, we
476                          * have to open the object in other device stack */
477                         if (!IS_ERR(dto)) {
478                                 dti->dti_fid = dto->do_lu.lo_header->loh_fid;
479                                 lu_object_put_nocache(env, &dto->do_lu);
480                                 dto = dt_locate(env, dt, &dti->dti_fid);
481                         }
482                         ls_device_put(env, ls);
483                 }
484         }
485         return dto;
486 }
487 EXPORT_SYMBOL(local_file_find_or_create_with_fid);
488
489 /*
490  * Look up and create (if it does not exist) a local named index file in parent
491  * directory.
492  */
493 struct dt_object *local_index_find_or_create(const struct lu_env *env,
494                                              struct local_oid_storage *los,
495                                              struct dt_object *parent,
496                                              const char *name, __u32 mode,
497                                              const struct dt_index_features *ft)
498 {
499         struct dt_thread_info   *dti = dt_info(env);
500         struct dt_object        *dto;
501         int                      rc;
502
503         LASSERT(parent);
504
505         rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
506         if (rc == 0) {
507                 /* name is found, get the object */
508                 dto = ls_locate(env, dt2ls_dev(los->los_dev),
509                                 &dti->dti_fid, NULL);
510         } else if (rc != -ENOENT) {
511                 dto = ERR_PTR(rc);
512         } else {
513                 rc = local_object_fid_generate(env, los, &dti->dti_fid);
514                 if (rc < 0) {
515                         dto = ERR_PTR(rc);
516                 } else {
517                         /* create the object */
518                         dti->dti_attr.la_valid          = LA_MODE;
519                         dti->dti_attr.la_mode           = mode;
520                         dti->dti_dof.dof_type           = DFT_INDEX;
521                         dti->dti_dof.u.dof_idx.di_feat  = ft;
522                         dto = __local_file_create(env, &dti->dti_fid, los,
523                                                   dt2ls_dev(los->los_dev),
524                                                   parent, name, &dti->dti_attr,
525                                                   &dti->dti_dof);
526                 }
527         }
528         return dto;
529
530 }
531 EXPORT_SYMBOL(local_index_find_or_create);
532
533 struct dt_object *
534 local_index_find_or_create_with_fid(const struct lu_env *env,
535                                     struct dt_device *dt,
536                                     const struct lu_fid *fid,
537                                     struct dt_object *parent,
538                                     const char *name, __u32 mode,
539                                     const struct dt_index_features *ft)
540 {
541         struct dt_thread_info   *dti = dt_info(env);
542         struct dt_object        *dto;
543         int                      rc;
544
545         LASSERT(parent);
546
547         rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
548         if (rc == 0) {
549                 /* name is found, get the object */
550                 if (!lu_fid_eq(fid, &dti->dti_fid))
551                         dto = ERR_PTR(-EINVAL);
552                 else
553                         dto = dt_locate(env, dt, fid);
554         } else if (rc != -ENOENT) {
555                 dto = ERR_PTR(rc);
556         } else {
557                 struct ls_device *ls;
558
559                 ls = ls_device_get(dt);
560                 if (IS_ERR(ls)) {
561                         dto = ERR_PTR(PTR_ERR(ls));
562                 } else {
563                         /* create the object */
564                         dti->dti_attr.la_valid          = LA_MODE;
565                         dti->dti_attr.la_mode           = mode;
566                         dti->dti_dof.dof_type           = DFT_INDEX;
567                         dti->dti_dof.u.dof_idx.di_feat  = ft;
568                         dto = __local_file_create(env, fid, NULL, ls, parent,
569                                                   name, &dti->dti_attr,
570                                                   &dti->dti_dof);
571                         /* ls_device_put() will finalize the ls device, we
572                          * have to open the object in other device stack */
573                         if (!IS_ERR(dto)) {
574                                 dti->dti_fid = dto->do_lu.lo_header->loh_fid;
575                                 lu_object_put_nocache(env, &dto->do_lu);
576                                 dto = dt_locate(env, dt, &dti->dti_fid);
577                         }
578                         ls_device_put(env, ls);
579                 }
580         }
581         return dto;
582 }
583 EXPORT_SYMBOL(local_index_find_or_create_with_fid);
584
585 static int local_object_declare_unlink(const struct lu_env *env,
586                                        struct dt_device *dt,
587                                        struct dt_object *p,
588                                        struct dt_object *c, const char *name,
589                                        struct thandle *th)
590 {
591         int rc;
592
593         rc = dt_declare_delete(env, p, (const struct dt_key *)name, th);
594         if (rc < 0)
595                 return rc;
596
597         rc = dt_declare_ref_del(env, c, th);
598         if (rc < 0)
599                 return rc;
600
601         return dt_declare_destroy(env, c, th);
602 }
603
604 int local_object_unlink(const struct lu_env *env, struct dt_device *dt,
605                         struct dt_object *parent, const char *name)
606 {
607         struct dt_thread_info   *dti = dt_info(env);
608         struct dt_object        *dto;
609         struct thandle          *th;
610         int                      rc;
611
612         ENTRY;
613
614         rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
615         if (rc == -ENOENT)
616                 RETURN(0);
617         else if (rc < 0)
618                 RETURN(rc);
619
620         dto = dt_locate(env, dt, &dti->dti_fid);
621         if (unlikely(IS_ERR(dto)))
622                 RETURN(PTR_ERR(dto));
623
624         th = dt_trans_create(env, dt);
625         if (IS_ERR(th))
626                 GOTO(out, rc = PTR_ERR(th));
627
628         rc = local_object_declare_unlink(env, dt, parent, dto, name, th);
629         if (rc < 0)
630                 GOTO(stop, rc);
631
632         rc = dt_trans_start_local(env, dt, th);
633         if (rc < 0)
634                 GOTO(stop, rc);
635
636         dt_write_lock(env, dto, 0);
637         rc = dt_delete(env, parent, (struct dt_key *)name, th, BYPASS_CAPA);
638         if (rc < 0)
639                 GOTO(unlock, rc);
640
641         rc = dt_ref_del(env, dto, th);
642         if (rc < 0) {
643                 struct dt_insert_rec *rec = &dti->dti_dt_rec;
644
645                 rec->rec_fid = &dti->dti_fid;
646                 rec->rec_type = dto->do_lu.lo_header->loh_attr;
647                 rc = dt_insert(env, parent, (const struct dt_rec *)rec,
648                                (const struct dt_key *)name, th, BYPASS_CAPA, 1);
649                 GOTO(unlock, rc);
650         }
651
652         rc = dt_destroy(env, dto, th);
653 unlock:
654         dt_write_unlock(env, dto);
655 stop:
656         dt_trans_stop(env, dt, th);
657 out:
658         lu_object_put_nocache(env, &dto->do_lu);
659         return rc;
660 }
661 EXPORT_SYMBOL(local_object_unlink);
662
663 struct local_oid_storage *dt_los_find(struct ls_device *ls, __u64 seq)
664 {
665         struct local_oid_storage *los, *ret = NULL;
666
667         cfs_list_for_each_entry(los, &ls->ls_los_list, los_list) {
668                 if (los->los_seq == seq) {
669                         atomic_inc(&los->los_refcount);
670                         ret = los;
671                         break;
672                 }
673         }
674         return ret;
675 }
676
677 void dt_los_put(struct local_oid_storage *los)
678 {
679         if (atomic_dec_and_test(&los->los_refcount))
680                 /* should never happen, only local_oid_storage_fini should
681                  * drop refcount to zero */
682                 LBUG();
683         return;
684 }
685
686 /* after Lustre 2.3 release there may be old file to store last generated FID
687  * If such file exists then we have to read its content
688  */
689 int lastid_compat_check(const struct lu_env *env, struct dt_device *dev,
690                         __u64 lastid_seq, __u32 *first_oid, struct ls_device *ls)
691 {
692         struct dt_thread_info   *dti = dt_info(env);
693         struct dt_object        *root = NULL;
694         struct los_ondisk        losd;
695         struct dt_object        *o = NULL;
696         int                      rc = 0;
697
698         rc = dt_root_get(env, dev, &dti->dti_fid);
699         if (rc)
700                 return rc;
701
702         root = ls_locate(env, ls, &dti->dti_fid, NULL);
703         if (IS_ERR(root))
704                 return PTR_ERR(root);
705
706         /* find old last_id file */
707         snprintf(dti->dti_buf, sizeof(dti->dti_buf), "seq-"LPX64"-lastid",
708                  lastid_seq);
709         rc = dt_lookup_dir(env, root, dti->dti_buf, &dti->dti_fid);
710         lu_object_put_nocache(env, &root->do_lu);
711         if (rc == -ENOENT) {
712                 /* old llog lastid accessed by FID only */
713                 if (lastid_seq != FID_SEQ_LLOG)
714                         return 0;
715                 dti->dti_fid.f_seq = FID_SEQ_LLOG;
716                 dti->dti_fid.f_oid = 1;
717                 dti->dti_fid.f_ver = 0;
718                 o = ls_locate(env, ls, &dti->dti_fid, NULL);
719                 if (IS_ERR(o))
720                         return PTR_ERR(o);
721
722                 if (!dt_object_exists(o)) {
723                         lu_object_put_nocache(env, &o->do_lu);
724                         return 0;
725                 }
726                 CDEBUG(D_INFO, "Found old llog lastid file\n");
727         } else if (rc < 0) {
728                 return rc;
729         } else {
730                 CDEBUG(D_INFO, "Found old lastid file for sequence "LPX64"\n",
731                        lastid_seq);
732                 o = ls_locate(env, ls, &dti->dti_fid, NULL);
733                 if (IS_ERR(o))
734                         return PTR_ERR(o);
735         }
736         /* let's read seq-NNNNNN-lastid file value */
737         LASSERT(dt_object_exists(o));
738         dti->dti_off = 0;
739         dti->dti_lb.lb_buf = &losd;
740         dti->dti_lb.lb_len = sizeof(losd);
741         dt_read_lock(env, o, 0);
742         rc = dt_record_read(env, o, &dti->dti_lb, &dti->dti_off);
743         dt_read_unlock(env, o);
744         if (rc == 0 && le32_to_cpu(losd.lso_magic) != LOS_MAGIC) {
745                 CERROR("%s: wrong content of seq-"LPX64"-lastid file, magic %x\n",
746                        o->do_lu.lo_dev->ld_obd->obd_name, lastid_seq,
747                        le32_to_cpu(losd.lso_magic));
748                 rc = -EINVAL;
749         } else if (rc < 0) {
750                 CERROR("%s: failed to read seq-"LPX64"-lastid: rc = %d\n",
751                        o->do_lu.lo_dev->ld_obd->obd_name, lastid_seq, rc);
752         }
753         lu_object_put_nocache(env, &o->do_lu);
754         if (rc == 0)
755                 *first_oid = le32_to_cpu(losd.lso_next_oid);
756         return rc;
757 }
758
759 /**
760  * Initialize local OID storage for required sequence.
761  * That may be needed for services that uses local files and requires
762  * dynamic OID allocation for them.
763  *
764  * Per each sequence we have an object with 'first_fid' identificator
765  * containing the counter for OIDs of locally created files with that
766  * sequence.
767  *
768  * It is used now by llog subsystem and MGS for NID tables
769  *
770  * Function gets first_fid to create counter object.
771  * All dynamic fids will be generated with the same sequence and incremented
772  * OIDs
773  *
774  * Returned local_oid_storage is in-memory representaion of OID storage
775  */
776 int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
777                            const struct lu_fid *first_fid,
778                            struct local_oid_storage **los)
779 {
780         struct dt_thread_info   *dti = dt_info(env);
781         struct ls_device        *ls;
782         obd_id                   lastid;
783         struct dt_object        *o = NULL;
784         struct thandle          *th;
785         __u32                    first_oid = fid_oid(first_fid);
786         int                      rc = 0;
787
788         ENTRY;
789
790         ls = ls_device_get(dev);
791         if (IS_ERR(ls))
792                 RETURN(PTR_ERR(ls));
793
794         mutex_lock(&ls->ls_los_mutex);
795         *los = dt_los_find(ls, fid_seq(first_fid));
796         if (*los != NULL)
797                 GOTO(out, rc = 0);
798
799         /* not found, then create */
800         OBD_ALLOC_PTR(*los);
801         if (*los == NULL)
802                 GOTO(out, rc = -ENOMEM);
803
804         atomic_set(&(*los)->los_refcount, 1);
805         mutex_init(&(*los)->los_id_lock);
806         (*los)->los_dev = &ls->ls_top_dev;
807         atomic_inc(&ls->ls_refcount);
808         cfs_list_add(&(*los)->los_list, &ls->ls_los_list);
809
810         /* Use {seq, 0, 0} to create the LAST_ID file for every
811          * sequence.  OIDs start at LUSTRE_FID_INIT_OID.
812          */
813         dti->dti_fid.f_seq = fid_seq(first_fid);
814         dti->dti_fid.f_oid = LUSTRE_FID_LASTID_OID;
815         dti->dti_fid.f_ver = 0;
816         o = ls_locate(env, ls, &dti->dti_fid, NULL);
817         if (IS_ERR(o))
818                 GOTO(out_los, rc = PTR_ERR(o));
819
820         if (!dt_object_exists(o)) {
821                 rc = lastid_compat_check(env, dev, fid_seq(first_fid),
822                                          &first_oid, ls);
823                 if (rc < 0)
824                         GOTO(out_los, rc);
825
826                 th = dt_trans_create(env, dev);
827                 if (IS_ERR(th))
828                         GOTO(out_los, rc = PTR_ERR(th));
829
830                 dti->dti_attr.la_valid = LA_MODE | LA_TYPE;
831                 dti->dti_attr.la_mode = S_IFREG | S_IRUGO | S_IWUSR;
832                 dti->dti_dof.dof_type = dt_mode_to_dft(S_IFREG);
833
834                 rc = dt_declare_create(env, o, &dti->dti_attr, NULL,
835                                        &dti->dti_dof, th);
836                 if (rc)
837                         GOTO(out_trans, rc);
838
839                 lastid = cpu_to_le64(first_oid);
840
841                 dti->dti_off = 0;
842                 dti->dti_lb.lb_buf = &lastid;
843                 dti->dti_lb.lb_len = sizeof(lastid);
844                 rc = dt_declare_record_write(env, o, &dti->dti_lb, dti->dti_off,
845                                              th);
846                 if (rc)
847                         GOTO(out_trans, rc);
848
849                 rc = dt_trans_start_local(env, dev, th);
850                 if (rc)
851                         GOTO(out_trans, rc);
852
853                 dt_write_lock(env, o, 0);
854                 if (dt_object_exists(o))
855                         GOTO(out_lock, rc = 0);
856
857                 rc = dt_create(env, o, &dti->dti_attr, NULL, &dti->dti_dof,
858                                th);
859                 if (rc)
860                         GOTO(out_lock, rc);
861
862                 rc = dt_record_write(env, o, &dti->dti_lb, &dti->dti_off, th);
863                 if (rc)
864                         GOTO(out_lock, rc);
865 out_lock:
866                 dt_write_unlock(env, o);
867 out_trans:
868                 dt_trans_stop(env, dev, th);
869         } else {
870                 dti->dti_off = 0;
871                 dti->dti_lb.lb_buf = &lastid;
872                 dti->dti_lb.lb_len = sizeof(lastid);
873                 dt_read_lock(env, o, 0);
874                 rc = dt_record_read(env, o, &dti->dti_lb, &dti->dti_off);
875                 dt_read_unlock(env, o);
876                 if (rc == 0 && le64_to_cpu(lastid) > OBIF_MAX_OID) {
877                         CERROR("%s: bad oid "LPU64" is read from LAST_ID\n",
878                                o->do_lu.lo_dev->ld_obd->obd_name,
879                                le64_to_cpu(lastid));
880                         rc = -EINVAL;
881                 }
882         }
883 out_los:
884         if (rc != 0) {
885                 cfs_list_del(&(*los)->los_list);
886                 atomic_dec(&ls->ls_refcount);
887                 OBD_FREE_PTR(*los);
888                 *los = NULL;
889                 if (o != NULL && !IS_ERR(o))
890                         lu_object_put_nocache(env, &o->do_lu);
891         } else {
892                 (*los)->los_seq = fid_seq(first_fid);
893                 (*los)->los_last_oid = le64_to_cpu(lastid);
894                 (*los)->los_obj = o;
895                 /* Read value should not be less than initial one
896                  * but possible after upgrade from older fs.
897                  * In this case just switch to the first_oid in memory and
898                  * it will be updated on disk with first object generated */
899                 if ((*los)->los_last_oid < first_oid)
900                         (*los)->los_last_oid = first_oid;
901         }
902 out:
903         mutex_unlock(&ls->ls_los_mutex);
904         ls_device_put(env, ls);
905         return rc;
906 }
907 EXPORT_SYMBOL(local_oid_storage_init);
908
909 void local_oid_storage_fini(const struct lu_env *env,
910                             struct local_oid_storage *los)
911 {
912         struct ls_device *ls;
913
914         LASSERT(env);
915         LASSERT(los->los_dev);
916         ls = dt2ls_dev(los->los_dev);
917
918         /* Take the mutex before decreasing the reference to avoid race
919          * conditions as described in LU-4721. */
920         mutex_lock(&ls->ls_los_mutex);
921         if (!atomic_dec_and_test(&los->los_refcount)) {
922                 mutex_unlock(&ls->ls_los_mutex);
923                 return;
924         }
925
926         if (los->los_obj)
927                 lu_object_put_nocache(env, &los->los_obj->do_lu);
928         cfs_list_del(&los->los_list);
929         OBD_FREE_PTR(los);
930         mutex_unlock(&ls->ls_los_mutex);
931         ls_device_put(env, ls);
932 }
933 EXPORT_SYMBOL(local_oid_storage_fini);