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