Whamcloud - gitweb
b=2264
[fs/lustre-release.git] / lustre / mds / mds_fs.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  mds/mds_fs.c
5  *  Lustre Metadata Server (MDS) filesystem interface code
6  *
7  *  Copyright (C) 2002, 2003 Cluster File Systems, Inc.
8  *   Author: Andreas Dilger <adilger@clusterfs.com>
9  *
10  *   This file is part of Lustre, http://www.lustre.org.
11  *
12  *   Lustre is free software; you can redistribute it and/or
13  *   modify it under the terms of version 2 of the GNU General Public
14  *   License as published by the Free Software Foundation.
15  *
16  *   Lustre is distributed in the hope that it will be useful,
17  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
18  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
19  *   GNU General Public License for more details.
20  *
21  *   You should have received a copy of the GNU General Public License
22  *   along with Lustre; if not, write to the Free Software
23  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
24  */
25
26 #ifndef EXPORT_SYMTAB
27 # define EXPORT_SYMTAB
28 #endif
29 #define DEBUG_SUBSYSTEM S_MDS
30
31 #include <linux/module.h>
32 #include <linux/kmod.h>
33 #include <linux/version.h>
34 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
35 #include <linux/mount.h>
36 #endif
37 #include <linux/lustre_mds.h>
38 #include <linux/obd_class.h>
39 #include <linux/obd_support.h>
40 #include <linux/lustre_lib.h>
41 #include <linux/lustre_fsfilt.h>
42 #include <portals/list.h>
43
44 #include "mds_internal.h"
45
46 /* This limit is arbitrary, but for now we fit it in 1 page (32k clients) */
47 #define MDS_MAX_CLIENTS (PAGE_SIZE * 8)
48 #define MDS_MAX_CLIENT_WORDS (MDS_MAX_CLIENTS / sizeof(unsigned long))
49
50 #define LAST_RCVD "last_rcvd"
51 #define LOV_OBJID "lov_objid"
52
53 /* Add client data to the MDS.  We use a bitmap to locate a free space
54  * in the last_rcvd file if cl_off is -1 (i.e. a new client).
55  * Otherwise, we have just read the data from the last_rcvd file and
56  * we know its offset.
57  */
58 int mds_client_add(struct obd_device *obd, struct mds_obd *mds,
59                    struct mds_export_data *med, int cl_idx)
60 {
61         unsigned long *bitmap = mds->mds_client_bitmap;
62         int new_client = (cl_idx == -1);
63         ENTRY;
64
65         LASSERT(bitmap != NULL);
66
67         /* XXX if mcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
68         if (!strcmp(med->med_mcd->mcd_uuid, obd->obd_uuid.uuid))
69                 RETURN(0);
70
71         /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
72          * there's no need for extra complication here
73          */
74         if (new_client) {
75                 cl_idx = find_first_zero_bit(bitmap, MDS_MAX_CLIENTS);
76         repeat:
77                 if (cl_idx >= MDS_MAX_CLIENTS) {
78                         CERROR("no room for clients - fix MDS_MAX_CLIENTS\n");
79                         return -ENOMEM;
80                 }
81                 if (test_and_set_bit(cl_idx, bitmap)) {
82                         cl_idx = find_next_zero_bit(bitmap, MDS_MAX_CLIENTS,
83                                                     cl_idx);
84                         goto repeat;
85                 }
86         } else {
87                 if (test_and_set_bit(cl_idx, bitmap)) {
88                         CERROR("MDS client %d: bit already set in bitmap!!\n",
89                                cl_idx);
90                         LBUG();
91                 }
92         }
93
94         CDEBUG(D_INFO, "client at idx %d with UUID '%s' added\n",
95                cl_idx, med->med_mcd->mcd_uuid);
96
97         med->med_idx = cl_idx;
98         med->med_off = MDS_LR_CLIENT_START + (cl_idx * MDS_LR_CLIENT_SIZE);
99
100         if (new_client) {
101                 struct obd_run_ctxt saved;
102                 loff_t off = med->med_off;
103                 struct file *file = mds->mds_rcvd_filp;
104                 int rc;
105
106                 push_ctxt(&saved, &obd->obd_ctxt, NULL);
107                 rc = fsfilt_write_record(obd, file, med->med_mcd,
108                                          sizeof(*med->med_mcd), &off, 1);
109                 pop_ctxt(&saved, &obd->obd_ctxt, NULL);
110
111                 if (rc)
112                         return rc;
113                 CDEBUG(D_INFO, "wrote client mcd at idx %u off %llu (len %u)\n",
114                        med->med_idx, med->med_off,
115                        (unsigned int)sizeof(*med->med_mcd));
116         }
117         return 0;
118 }
119
120 int mds_client_free(struct obd_export *exp, int clear_client)
121 {
122         struct mds_export_data *med = &exp->exp_mds_data;
123         struct mds_obd *mds = &exp->exp_obd->u.mds;
124         struct obd_device *obd = exp->exp_obd;
125         struct mds_client_data zero_mcd;
126         struct obd_run_ctxt saved;
127         int rc;
128         unsigned long *bitmap = mds->mds_client_bitmap;
129
130         LASSERT(bitmap);
131         if (!med->med_mcd)
132                 RETURN(0);
133
134         /* XXX if mcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
135         if (!strcmp(med->med_mcd->mcd_uuid, obd->obd_uuid.uuid))
136                 GOTO(free_and_out, 0);
137
138         CDEBUG(D_INFO, "freeing client at idx %u (%lld)with UUID '%s'\n",
139                med->med_idx, med->med_off, med->med_mcd->mcd_uuid);
140
141         /* Clear the bit _after_ zeroing out the client so we don't
142            race with mds_client_add and zero out new clients.*/
143         if (!test_bit(med->med_idx, bitmap)) {
144                 CERROR("MDS client %u: bit already clear in bitmap!!\n",
145                        med->med_idx);
146                 LBUG();
147         }
148
149         if (clear_client) {
150                 memset(&zero_mcd, 0, sizeof zero_mcd);
151                 push_ctxt(&saved, &obd->obd_ctxt, NULL);
152                 rc = fsfilt_write_record(obd, mds->mds_rcvd_filp, &zero_mcd,
153                                          sizeof(zero_mcd), &med->med_off, 1);
154                 pop_ctxt(&saved, &obd->obd_ctxt, NULL);
155
156                 CDEBUG(rc == 0 ? D_INFO : D_ERROR,
157                        "zeroing out client %s idx %u in %s rc %d\n",
158                        med->med_mcd->mcd_uuid, med->med_idx, LAST_RCVD, rc);
159         }
160
161         if (!test_and_clear_bit(med->med_idx, bitmap)) {
162                 CERROR("MDS client %u: bit already clear in bitmap!!\n",
163                        med->med_idx);
164                 LBUG();
165         }
166
167  free_and_out:
168         OBD_FREE(med->med_mcd, sizeof(*med->med_mcd));
169
170         return 0;
171 }
172
173 static int mds_server_free_data(struct mds_obd *mds)
174 {
175         OBD_FREE(mds->mds_client_bitmap,
176                  MDS_MAX_CLIENT_WORDS * sizeof(unsigned long));
177         OBD_FREE(mds->mds_server_data, sizeof(*mds->mds_server_data));
178         mds->mds_server_data = NULL;
179
180         return 0;
181 }
182
183 static int mds_read_last_rcvd(struct obd_device *obd, struct file *file)
184 {
185         struct mds_obd *mds = &obd->u.mds;
186         struct mds_server_data *msd;
187         struct mds_client_data *mcd = NULL;
188         loff_t off = 0;
189         unsigned long last_rcvd_size = file->f_dentry->d_inode->i_size;
190         __u64 mount_count;
191         int cl_idx, rc = 0;
192         ENTRY;
193
194         /* ensure padding in the struct is the correct size */
195         LASSERT (offsetof(struct mds_server_data, msd_padding) +
196                  sizeof(msd->msd_padding) == MDS_LR_SERVER_SIZE);
197         LASSERT (offsetof(struct mds_client_data, mcd_padding) +
198                  sizeof(mcd->mcd_padding) == MDS_LR_CLIENT_SIZE);
199
200         OBD_ALLOC_WAIT(msd, sizeof(*msd));
201         if (!msd)
202                 RETURN(-ENOMEM);
203
204         OBD_ALLOC_WAIT(mds->mds_client_bitmap,
205                   MDS_MAX_CLIENT_WORDS * sizeof(unsigned long));
206         if (!mds->mds_client_bitmap) {
207                 OBD_FREE(msd, sizeof(*msd));
208                 RETURN(-ENOMEM);
209         }
210
211         mds->mds_server_data = msd;
212
213         if (last_rcvd_size == 0) {
214                 CWARN("%s: initializing new %s\n", obd->obd_name, LAST_RCVD);
215
216                 memcpy(msd->msd_uuid, obd->obd_uuid.uuid,sizeof(msd->msd_uuid));
217                 msd->msd_last_transno = 0;
218                 mount_count = msd->msd_mount_count = 0; 
219                 msd->msd_server_size = cpu_to_le32(MDS_LR_SERVER_SIZE);
220                 msd->msd_client_start = cpu_to_le32(MDS_LR_CLIENT_START);
221                 msd->msd_client_size = cpu_to_le16(MDS_LR_CLIENT_SIZE);
222                 msd->msd_feature_rocompat = cpu_to_le32(MDS_ROCOMPAT_LOVOBJID);
223         } else {
224                 rc = fsfilt_read_record(obd, file, msd, sizeof(*msd), &off);
225                 if (rc) {
226                         CERROR("error reading MDS %s: rc = %d\n", LAST_RCVD, rc);
227                         GOTO(err_msd, rc);
228                 }
229                 if (strcmp(msd->msd_uuid, obd->obd_uuid.uuid) != 0) {
230                         CERROR("OBD UUID %s does not match last_rcvd UUID %s\n",
231                                obd->obd_uuid.uuid, msd->msd_uuid);
232                         GOTO(err_msd, rc = -EINVAL);
233                 }
234                 mount_count = le64_to_cpu(msd->msd_mount_count);
235         }
236         if (msd->msd_feature_incompat & ~cpu_to_le32(MDS_INCOMPAT_SUPP)) {
237                 CERROR("unsupported incompat feature %x\n",
238                        le32_to_cpu(msd->msd_feature_incompat) &
239                        ~MDS_INCOMPAT_SUPP);
240                 GOTO(err_msd, rc = -EINVAL);
241         }
242         /* XXX updating existing b_devel fs only, can be removed in future */
243         msd->msd_feature_rocompat = cpu_to_le32(MDS_ROCOMPAT_LOVOBJID);
244         if (msd->msd_feature_rocompat & ~cpu_to_le32(MDS_ROCOMPAT_SUPP)) {
245                 CERROR("unsupported read-only feature %x\n",
246                        le32_to_cpu(msd->msd_feature_rocompat) &
247                        ~MDS_ROCOMPAT_SUPP);
248                 /* Do something like remount filesystem read-only */
249                 GOTO(err_msd, rc = -EINVAL);
250         }
251
252         mds->mds_last_transno = le64_to_cpu(msd->msd_last_transno);
253
254         CDEBUG(D_INODE, "%s: server last_transno: "LPU64"\n",
255                obd->obd_name, mds->mds_last_transno);
256         CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n",
257                obd->obd_name, mount_count + 1);
258         CDEBUG(D_INODE, "%s: server data size: %u\n",
259                obd->obd_name, le32_to_cpu(msd->msd_server_size));
260         CDEBUG(D_INODE, "%s: per-client data start: %u\n",
261                obd->obd_name, le32_to_cpu(msd->msd_client_start));
262         CDEBUG(D_INODE, "%s: per-client data size: %u\n",
263                obd->obd_name, le32_to_cpu(msd->msd_client_size));
264         CDEBUG(D_INODE, "%s: last_rcvd size: %lu\n",
265                obd->obd_name, last_rcvd_size);
266         CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name,
267                last_rcvd_size <= MDS_LR_CLIENT_START ? 0 :
268                (last_rcvd_size - MDS_LR_CLIENT_START) / MDS_LR_CLIENT_SIZE);
269
270         /* When we do a clean MDS shutdown, we save the last_transno into
271          * the header.  If we find clients with higher last_transno values
272          * then those clients may need recovery done. */
273         for (cl_idx = 0, off = le32_to_cpu(msd->msd_client_start);
274              off < last_rcvd_size; cl_idx++) {
275                 __u64 last_transno;
276                 struct obd_export *exp;
277                 struct mds_export_data *med;
278
279                 if (!mcd) {
280                         OBD_ALLOC_WAIT(mcd, sizeof(*mcd));
281                         if (!mcd)
282                                 GOTO(err_client, rc = -ENOMEM);
283                 }
284
285                 /* Don't assume off is incremented properly by
286                  * fsfilt_read_record(), in case sizeof(*mcd)
287                  * isn't the same as msd->msd_client_size.  */
288                 off = le32_to_cpu(msd->msd_client_start) +
289                         cl_idx * le16_to_cpu(msd->msd_client_size);
290                 rc = fsfilt_read_record(obd, file, mcd, sizeof(*mcd), &off);
291                 if (rc) {
292                         CERROR("error reading MDS %s idx %d, off %llu: rc %d\n",
293                                LAST_RCVD, cl_idx, off, rc);
294                         break; /* read error shouldn't cause startup to fail */
295                 }
296
297                 if (mcd->mcd_uuid[0] == '\0') {
298                         CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
299                                cl_idx);
300                         continue;
301                 }
302
303                 last_transno = le64_to_cpu(mcd->mcd_last_transno);
304
305                 /* These exports are cleaned up by mds_disconnect(), so they
306                  * need to be set up like real exports as mds_connect() does.
307                  */
308                 CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64
309                        " srv lr: "LPU64"\n", mcd->mcd_uuid, cl_idx,
310                        last_transno, le64_to_cpu(msd->msd_last_transno));
311
312                 exp = class_new_export(obd);
313                 if (exp == NULL)
314                         GOTO(err_client, rc = -ENOMEM);
315
316                 memcpy(&exp->exp_client_uuid.uuid, mcd->mcd_uuid,
317                        sizeof exp->exp_client_uuid.uuid);
318                 med = &exp->exp_mds_data;
319                 med->med_mcd = mcd;
320                 mds_client_add(obd, mds, med, cl_idx);
321                 /* create helper if export init gets more complex */
322                 INIT_LIST_HEAD(&med->med_open_head);
323                 spin_lock_init(&med->med_open_lock);
324
325                 mcd = NULL;
326                 obd->obd_recoverable_clients++;
327                 obd->obd_max_recoverable_clients++;
328                 class_export_put(exp);
329
330                 CDEBUG(D_OTHER, "client at idx %d has last_transno = "LPU64"\n",
331                        cl_idx, last_transno);
332
333                 if (last_transno > mds->mds_last_transno)
334                        mds->mds_last_transno = last_transno;
335         }
336
337         obd->obd_last_committed = mds->mds_last_transno;
338         if (obd->obd_recoverable_clients) {
339                 CWARN("RECOVERY: service %s, %d recoverable clients, "
340                       "last_transno "LPU64"\n", obd->obd_name,
341                       obd->obd_recoverable_clients, mds->mds_last_transno);
342                 obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
343                 obd->obd_recovering = 1;
344         }
345
346         if (mcd)
347                 OBD_FREE(mcd, sizeof(*mcd));
348         
349         mds->mds_mount_count = mount_count + 1;
350         msd->msd_mount_count = cpu_to_le64(mds->mds_mount_count);
351
352         /* save it, so mount count and last_transno is current */
353         rc = mds_update_server_data(obd, 1);
354
355         RETURN(rc);
356
357 err_client:
358         class_disconnect_exports(obd, 0);
359 err_msd:
360         mds_server_free_data(mds);
361         RETURN(rc);
362 }
363
364 int mds_fs_setup(struct obd_device *obd, struct vfsmount *mnt)
365 {
366         struct mds_obd *mds = &obd->u.mds;
367         struct obd_run_ctxt saved;
368         struct dentry *dentry;
369         struct file *file;
370         int rc;
371         ENTRY;
372
373
374         /* Get rid of unneeded supplementary groups */
375         current->ngroups = 0;
376         memset(current->groups, 0, sizeof(current->groups));
377
378         mds->mds_vfsmnt = mnt;
379         mds->mds_sb = mnt->mnt_root->d_inode->i_sb;
380
381         fsfilt_setup(obd, mds->mds_sb);
382
383         OBD_SET_CTXT_MAGIC(&obd->obd_ctxt);
384         obd->obd_ctxt.pwdmnt = mnt;
385         obd->obd_ctxt.pwd = mnt->mnt_root;
386         obd->obd_ctxt.fs = get_ds();
387         obd->obd_ctxt.cb_ops = mds_lvfs_ops;
388
389         /* setup the directory tree */
390         push_ctxt(&saved, &obd->obd_ctxt, NULL);
391         dentry = simple_mkdir(current->fs->pwd, "ROOT", 0755, 0);
392         if (IS_ERR(dentry)) {
393                 rc = PTR_ERR(dentry);
394                 CERROR("cannot create ROOT directory: rc = %d\n", rc);
395                 GOTO(err_pop, rc);
396         }
397
398         mds->mds_rootfid.id = dentry->d_inode->i_ino;
399         mds->mds_rootfid.generation = dentry->d_inode->i_generation;
400         mds->mds_rootfid.f_type = S_IFDIR;
401
402         dput(dentry);
403
404         dentry = lookup_one_len("__iopen__", current->fs->pwd,
405                                 strlen("__iopen__"));
406         if (IS_ERR(dentry) || !dentry->d_inode) {
407                 rc = (IS_ERR(dentry)) ? PTR_ERR(dentry): -ENOENT;
408                 CERROR("cannot open iopen FH directory: rc = %d\n", rc);
409                 GOTO(err_pop, rc);
410         }
411         mds->mds_fid_de = dentry;
412
413         dentry = simple_mkdir(current->fs->pwd, "PENDING", 0777, 1);
414         if (IS_ERR(dentry)) {
415                 rc = PTR_ERR(dentry);
416                 CERROR("cannot create PENDING directory: rc = %d\n", rc);
417                 GOTO(err_fid, rc);
418         }
419         mds->mds_pending_dir = dentry;
420
421         dentry = simple_mkdir(current->fs->pwd, "LOGS", 0777, 1);
422         if (IS_ERR(dentry)) {
423                 rc = PTR_ERR(dentry);
424                 CERROR("cannot create LOGS directory: rc = %d\n", rc);
425                 GOTO(err_pending, rc);
426         }
427         mds->mds_logs_dir = dentry;
428
429         dentry = simple_mkdir(current->fs->pwd, "OBJECTS", 0777, 1);
430         if (IS_ERR(dentry)) {
431                 rc = PTR_ERR(dentry);
432                 CERROR("cannot create OBJECTS directory: rc = %d\n", rc);
433                 GOTO(err_logs, rc);
434         }
435         mds->mds_objects_dir = dentry;
436
437         /* open and test the last rcvd file */
438         file = filp_open(LAST_RCVD, O_RDWR | O_CREAT, 0644);
439         if (IS_ERR(file)) {
440                 rc = PTR_ERR(file);
441                 CERROR("cannot open/create %s file: rc = %d\n", LAST_RCVD, rc);
442                 GOTO(err_objects, rc = PTR_ERR(file));
443         }
444         mds->mds_rcvd_filp = file;
445         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
446                 CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD,
447                        file->f_dentry->d_inode->i_mode);
448                 GOTO(err_last_rcvd, rc = -ENOENT);
449         }
450
451         rc = mds_read_last_rcvd(obd, file);
452         if (rc) {
453                 CERROR("cannot read %s: rc = %d\n", LAST_RCVD, rc);
454                 GOTO(err_last_rcvd, rc);
455         }
456
457         /* open and test the lov objd file */
458         file = filp_open(LOV_OBJID, O_RDWR | O_CREAT, 0644);
459         if (IS_ERR(file)) {
460                 rc = PTR_ERR(file);
461                 CERROR("cannot open/create %s file: rc = %d\n", LOV_OBJID, rc);
462                 GOTO(err_client, rc = PTR_ERR(file));
463         }
464         mds->mds_lov_objid_filp = file;
465         if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
466                 CERROR("%s is not a regular file!: mode = %o\n", LOV_OBJID,
467                        file->f_dentry->d_inode->i_mode);
468                 GOTO(err_lov_objid, rc = -ENOENT);
469         }
470 err_pop:
471         pop_ctxt(&saved, &obd->obd_ctxt, NULL);
472
473         return rc;
474
475 err_lov_objid:
476         if (mds->mds_lov_objid_filp && filp_close(mds->mds_lov_objid_filp, 0))
477                 CERROR("can't close %s after error\n", LOV_OBJID);
478 err_client:
479         class_disconnect_exports(obd, 0);
480 err_last_rcvd:
481         if (mds->mds_rcvd_filp && filp_close(mds->mds_rcvd_filp, 0))
482                 CERROR("can't close %s after error\n", LAST_RCVD);
483 err_objects:
484         dput(mds->mds_objects_dir);
485 err_logs:
486         dput(mds->mds_logs_dir);
487 err_pending:
488         dput(mds->mds_pending_dir);
489 err_fid:
490         dput(mds->mds_fid_de);
491         goto err_pop;
492 }
493
494
495 int mds_fs_cleanup(struct obd_device *obd, int flags)
496 {
497         struct mds_obd *mds = &obd->u.mds;
498         struct obd_run_ctxt saved;
499         int rc = 0;
500
501         if (flags & OBD_OPT_FAILOVER)
502                 CERROR("%s: shutting down for failover; client state will"
503                        " be preserved.\n", obd->obd_name);
504
505         class_disconnect_exports(obd, flags); /* cleans up client info too */
506         mds_server_free_data(mds);
507
508         push_ctxt(&saved, &obd->obd_ctxt, NULL);
509         if (mds->mds_rcvd_filp) {
510                 rc = filp_close(mds->mds_rcvd_filp, 0);
511                 mds->mds_rcvd_filp = NULL;
512                 if (rc)
513                         CERROR("%s file won't close, rc=%d\n", LAST_RCVD, rc);
514         }
515         if (mds->mds_lov_objid_filp) {
516                 rc = filp_close(mds->mds_lov_objid_filp, 0);
517                 mds->mds_lov_objid_filp = NULL;
518                 if (rc)
519                         CERROR("%s file won't close, rc=%d\n", LOV_OBJID, rc);
520         }
521         if (mds->mds_objects_dir != NULL) {
522                 l_dput(mds->mds_objects_dir);
523                 mds->mds_objects_dir = NULL;
524         }
525         if (mds->mds_logs_dir) {
526                 l_dput(mds->mds_logs_dir);
527                 mds->mds_logs_dir = NULL;
528         }
529         if (mds->mds_pending_dir) {
530                 l_dput(mds->mds_pending_dir);
531                 mds->mds_pending_dir = NULL;
532         }
533         pop_ctxt(&saved, &obd->obd_ctxt, NULL);
534         shrink_dcache_parent(mds->mds_fid_de);
535         dput(mds->mds_fid_de);
536
537         return rc;
538 }
539
540 /* Creates an object with the same name as its fid.  Because this is not at all
541  * performance sensitive, it is accomplished by creating a file, checking the
542  * fid, and renaming it. */
543 int mds_obd_create(struct obd_export *exp, struct obdo *oa,
544                       struct lov_stripe_md **ea, struct obd_trans_info *oti)
545 {
546         struct mds_obd *mds = &exp->exp_obd->u.mds;
547         struct inode *parent_inode = mds->mds_objects_dir->d_inode;
548         unsigned int tmpname = ll_insecure_random_int();
549         struct file *filp;
550         struct dentry *new_child;
551         struct obd_run_ctxt saved;
552         char fidname[LL_FID_NAMELEN];
553         void *handle;
554         int rc = 0, err, namelen;
555         ENTRY;
556
557         push_ctxt(&saved, &exp->exp_obd->obd_ctxt, NULL);
558         
559         sprintf(fidname, "OBJECTS/%u", tmpname);
560         filp = filp_open(fidname, O_CREAT | O_EXCL, 0644);
561         if (IS_ERR(filp)) {
562                 rc = PTR_ERR(filp);
563                 if (rc == -EEXIST) {
564                         CERROR("impossible object name collision %u\n",
565                                tmpname);
566                         LBUG();
567                 }
568                 CERROR("error creating tmp object %u: rc %d\n", tmpname, rc);
569                 GOTO(out_pop, rc);
570         }
571
572         LASSERT(mds->mds_objects_dir == filp->f_dentry->d_parent);
573
574         oa->o_id = filp->f_dentry->d_inode->i_ino;
575         oa->o_generation = filp->f_dentry->d_inode->i_generation;
576         namelen = ll_fid2str(fidname, oa->o_id, oa->o_generation);
577
578         down(&parent_inode->i_sem);
579         new_child = lookup_one_len(fidname, mds->mds_objects_dir, namelen);
580
581         if (IS_ERR(new_child)) {
582                 CERROR("getting neg dentry for obj rename: %d\n", rc);
583                 GOTO(out_close, rc = PTR_ERR(new_child));
584         }
585         if (new_child->d_inode != NULL) {
586                 CERROR("impossible non-negative obj dentry " LPU64":%u!\n",
587                        oa->o_id, oa->o_generation);
588                 LBUG();
589         }
590
591         handle = fsfilt_start(exp->exp_obd, mds->mds_objects_dir->d_inode,
592                               FSFILT_OP_RENAME, NULL);
593         if (IS_ERR(handle)) 
594                 GOTO(out_dput, rc = PTR_ERR(handle));
595         
596         lock_kernel();
597         rc = vfs_rename(mds->mds_objects_dir->d_inode, filp->f_dentry,
598                         mds->mds_objects_dir->d_inode, new_child);
599         unlock_kernel();
600         if (rc)
601                 CERROR("error renaming new object "LPU64":%u: rc %d\n",
602                        oa->o_id, oa->o_generation, rc);
603
604         err = fsfilt_commit(exp->exp_obd, mds->mds_objects_dir->d_inode,
605                             handle, 0);
606         if (!err)
607                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGENER;
608         else if (!rc)
609                 rc = err;
610 out_dput:
611         dput(new_child);
612 out_close:
613         up(&parent_inode->i_sem);
614         err = filp_close(filp, 0);
615         if (err) {
616                 CERROR("closing tmpfile %u: rc %d\n", tmpname, rc);
617                 if (!rc)
618                         rc = err;
619         }
620 out_pop:
621         pop_ctxt(&saved, &exp->exp_obd->obd_ctxt, NULL);
622         RETURN(rc);
623 }
624
625 int mds_obd_destroy(struct obd_export *exp, struct obdo *oa,
626                     struct lov_stripe_md *ea, struct obd_trans_info *oti)
627 {
628         struct mds_obd *mds = &exp->exp_obd->u.mds;
629         struct inode *parent_inode = mds->mds_objects_dir->d_inode;
630         struct obd_device *obd = exp->exp_obd;
631         struct obd_run_ctxt saved;
632         char fidname[LL_FID_NAMELEN];
633         struct dentry *de;
634         void *handle;
635         int err, namelen, rc = 0;
636         ENTRY;
637
638         push_ctxt(&saved, &obd->obd_ctxt, NULL);
639
640         namelen = ll_fid2str(fidname, oa->o_id, oa->o_generation);
641
642         down(&parent_inode->i_sem);
643         de = lookup_one_len(fidname, mds->mds_objects_dir, namelen);
644         if (de == NULL || de->d_inode == NULL) {
645                 CERROR("destroying non-existent object "LPU64"\n", oa->o_id);
646                 GOTO(out_dput, rc = IS_ERR(de) ? PTR_ERR(de) : -ENOENT);
647         }
648
649         handle = fsfilt_start(obd, mds->mds_objects_dir->d_inode,
650                               FSFILT_OP_UNLINK_LOG, oti);
651         if (IS_ERR(handle)) {
652                 GOTO(out_dput, rc = PTR_ERR(handle));
653         }
654         
655         rc = vfs_unlink(mds->mds_objects_dir->d_inode, de);
656         if (rc) 
657                 CERROR("error destroying object "LPU64":%u: rc %d\n",
658                        oa->o_id, oa->o_generation, rc);
659         
660         err = fsfilt_commit(obd, mds->mds_objects_dir->d_inode, handle, 0);
661         if (err && !rc)
662                 rc = err;
663 out_dput:
664         if (de != NULL)
665                 l_dput(de);
666         up(&parent_inode->i_sem);
667         pop_ctxt(&saved, &obd->obd_ctxt, NULL);
668         RETURN(rc);
669 }