Whamcloud - gitweb
Cleanup Compiler Warnings(Patch by Robert Read)
[fs/lustre-release.git] / lustre / mdt / mdt_recovery.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  linux/mdt/mdt_recovery.c
5  *  Lustre Metadata Target (mdt) recovery-related methods
6  *
7  *  Copyright (C) 2002-2006 Cluster File Systems, Inc.
8  *   Author: Huang Hua <huanghua@clusterfs.com>
9  *   Author: Pershin Mike <tappro@clusterfs.com>
10  *
11  *   This file is part of the Lustre file system, http://www.lustre.org
12  *   Lustre is a trademark of Cluster File Systems, Inc.
13  *
14  *   You may have signed or agreed to another license before downloading
15  *   this software.  If so, you are bound by the terms and conditions
16  *   of that agreement, and the following does not apply to you.  See the
17  *   LICENSE file included with this distribution for more information.
18  *
19  *   If you did not agree to a different license, then this copy of Lustre
20  *   is open source software; you can redistribute it and/or modify it
21  *   under the terms of version 2 of the GNU General Public License as
22  *   published by the Free Software Foundation.
23  *
24  *   In either case, Lustre is distributed in the hope that it will be
25  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
26  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
27  *   license text for more details.
28  */
29 #ifndef EXPORT_SYMTAB
30 # define EXPORT_SYMTAB
31 #endif
32 #define DEBUG_SUBSYSTEM S_MDS
33
34 #include "mdt_internal.h"
35
36 static int mdt_server_data_update(const struct lu_env *env,
37                                   struct mdt_device *mdt);
38
39 struct lu_buf *mdt_buf(const struct lu_env *env, void *area, ssize_t len)
40 {
41         struct lu_buf *buf;
42         struct mdt_thread_info *mti;
43
44         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
45         buf = &mti->mti_buf;
46         buf->lb_buf = area;
47         buf->lb_len = len;
48         return buf;
49 }
50
51 const struct lu_buf *mdt_buf_const(const struct lu_env *env,
52                                    const void *area, ssize_t len)
53 {
54         struct lu_buf *buf;
55         struct mdt_thread_info *mti;
56
57         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
58         buf = &mti->mti_buf;
59
60         buf->lb_buf = (void *)area;
61         buf->lb_len = len;
62         return buf;
63 }
64
65 int mdt_record_read(const struct lu_env *env,
66                     struct dt_object *dt, struct lu_buf *buf, loff_t *pos)
67 {
68         int rc;
69
70         LASSERTF(dt != NULL, "dt is NULL when we want to read record\n");
71
72         rc = dt->do_body_ops->dbo_read(env, dt, buf, pos, BYPASS_CAPA);
73
74         if (rc == buf->lb_len)
75                 rc = 0;
76         else if (rc >= 0)
77                 rc = -EFAULT;
78         return rc;
79 }
80
81 int mdt_record_write(const struct lu_env *env,
82                      struct dt_object *dt, const struct lu_buf *buf,
83                      loff_t *pos, struct thandle *th)
84 {
85         int rc;
86
87         LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
88         LASSERT(th != NULL);
89         rc = dt->do_body_ops->dbo_write(env, dt, buf, pos, th, BYPASS_CAPA);
90         if (rc == buf->lb_len)
91                 rc = 0;
92         else if (rc >= 0)
93                 rc = -EFAULT;
94         return rc;
95 }
96 /* only one record write */
97
98 enum {
99         MDT_TXN_LAST_RCVD_WRITE_CREDITS = 3
100 };
101
102 struct thandle* mdt_trans_start(const struct lu_env *env,
103                                 struct mdt_device *mdt, int credits)
104 {
105         struct mdt_thread_info *mti;
106         struct txn_param *p;
107
108         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
109         p = &mti->mti_txn_param;
110         txn_param_init(p, credits);
111
112         /* export can require sync operations */
113         if (mti->mti_exp != NULL)
114                 p->tp_sync = mti->mti_exp->exp_need_sync;
115
116         return mdt->mdt_bottom->dd_ops->dt_trans_start(env, mdt->mdt_bottom, p);
117 }
118
119 void mdt_trans_stop(const struct lu_env *env,
120                     struct mdt_device *mdt, struct thandle *th)
121 {
122         mdt->mdt_bottom->dd_ops->dt_trans_stop(env, th);
123 }
124
125 /* last_rcvd handling */
126 static inline void lsd_le_to_cpu(struct lr_server_data *buf,
127                                  struct lr_server_data *lsd)
128 {
129         memcpy(lsd->lsd_uuid, buf->lsd_uuid, sizeof (lsd->lsd_uuid));
130         lsd->lsd_last_transno     = le64_to_cpu(buf->lsd_last_transno);
131         lsd->lsd_mount_count      = le64_to_cpu(buf->lsd_mount_count);
132         lsd->lsd_feature_compat   = le32_to_cpu(buf->lsd_feature_compat);
133         lsd->lsd_feature_rocompat = le32_to_cpu(buf->lsd_feature_rocompat);
134         lsd->lsd_feature_incompat = le32_to_cpu(buf->lsd_feature_incompat);
135         lsd->lsd_server_size      = le32_to_cpu(buf->lsd_server_size);
136         lsd->lsd_client_start     = le32_to_cpu(buf->lsd_client_start);
137         lsd->lsd_client_size      = le16_to_cpu(buf->lsd_client_size);
138 }
139
140 static inline void lsd_cpu_to_le(struct lr_server_data *lsd,
141                                  struct lr_server_data *buf)
142 {
143         memcpy(buf->lsd_uuid, lsd->lsd_uuid, sizeof (lsd->lsd_uuid));
144         buf->lsd_last_transno     = cpu_to_le64(lsd->lsd_last_transno);
145         buf->lsd_mount_count      = cpu_to_le64(lsd->lsd_mount_count);
146         buf->lsd_feature_compat   = cpu_to_le32(lsd->lsd_feature_compat);
147         buf->lsd_feature_rocompat = cpu_to_le32(lsd->lsd_feature_rocompat);
148         buf->lsd_feature_incompat = cpu_to_le32(lsd->lsd_feature_incompat);
149         buf->lsd_server_size      = cpu_to_le32(lsd->lsd_server_size);
150         buf->lsd_client_start     = cpu_to_le32(lsd->lsd_client_start);
151         buf->lsd_client_size      = cpu_to_le16(lsd->lsd_client_size);
152 }
153
154 static inline void lcd_le_to_cpu(struct lsd_client_data *buf,
155                                  struct lsd_client_data *lcd)
156 {
157         memcpy(lcd->lcd_uuid, buf->lcd_uuid, sizeof (lcd->lcd_uuid));
158         lcd->lcd_last_transno       = le64_to_cpu(buf->lcd_last_transno);
159         lcd->lcd_last_xid           = le64_to_cpu(buf->lcd_last_xid);
160         lcd->lcd_last_result        = le32_to_cpu(buf->lcd_last_result);
161         lcd->lcd_last_data          = le32_to_cpu(buf->lcd_last_data);
162         lcd->lcd_last_close_transno = le64_to_cpu(buf->lcd_last_close_transno);
163         lcd->lcd_last_close_xid     = le64_to_cpu(buf->lcd_last_close_xid);
164         lcd->lcd_last_close_result  = le32_to_cpu(buf->lcd_last_close_result);
165 }
166
167 static inline void lcd_cpu_to_le(struct lsd_client_data *lcd,
168                                  struct lsd_client_data *buf)
169 {
170         memcpy(buf->lcd_uuid, lcd->lcd_uuid, sizeof (lcd->lcd_uuid));
171         buf->lcd_last_transno       = cpu_to_le64(lcd->lcd_last_transno);
172         buf->lcd_last_xid           = cpu_to_le64(lcd->lcd_last_xid);
173         buf->lcd_last_result        = cpu_to_le32(lcd->lcd_last_result);
174         buf->lcd_last_data          = cpu_to_le32(lcd->lcd_last_data);
175         buf->lcd_last_close_transno = cpu_to_le64(lcd->lcd_last_close_transno);
176         buf->lcd_last_close_xid     = cpu_to_le64(lcd->lcd_last_close_xid);
177         buf->lcd_last_close_result  = cpu_to_le32(lcd->lcd_last_close_result);
178 }
179
180 static inline int mdt_last_rcvd_header_read(const struct lu_env *env,
181                                             struct mdt_device *mdt)
182 {
183         struct mdt_thread_info *mti;
184         int rc;
185
186         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
187
188         mti->mti_off = 0;
189         rc = mdt_record_read(env, mdt->mdt_last_rcvd,
190                              mdt_buf(env, &mti->mti_lsd, sizeof(mti->mti_lsd)),
191                              &mti->mti_off);
192         if (rc == 0)
193                 lsd_le_to_cpu(&mti->mti_lsd, &mdt->mdt_lsd);
194
195         CDEBUG(D_INFO, "read last_rcvd header rc = %d:\n"
196                        "uuid = %s\n"
197                        "last_transno = "LPU64"\n",
198                         rc, mdt->mdt_lsd.lsd_uuid,
199                         mdt->mdt_lsd.lsd_last_transno);
200         return rc;
201 }
202
203 static inline int mdt_last_rcvd_header_write(const struct lu_env *env,
204                                              struct mdt_device *mdt)
205 {
206         struct mdt_thread_info *mti;
207         struct thandle *th;
208         int rc;
209         ENTRY;
210
211         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
212
213         th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
214         if (IS_ERR(th))
215                 RETURN(PTR_ERR(th));
216
217         mti->mti_off = 0;        
218         lsd_cpu_to_le(&mdt->mdt_lsd, &mti->mti_lsd);
219
220         rc = mdt_record_write(env, mdt->mdt_last_rcvd,
221                               mdt_buf_const(env, &mti->mti_lsd, sizeof(mti->mti_lsd)),
222                               &mti->mti_off, th);
223
224         mdt_trans_stop(env, mdt, th);
225
226         CDEBUG(D_INFO, "write last_rcvd header rc = %d:\n"
227                "uuid = %s\nlast_transno = "LPU64"\n",
228                rc, mdt->mdt_lsd.lsd_uuid, mdt->mdt_lsd.lsd_last_transno);
229         
230         RETURN(rc);
231 }
232
233 static int mdt_last_rcvd_read(const struct lu_env *env,
234                               struct mdt_device *mdt,
235                               struct lsd_client_data *lcd, loff_t *off)
236 {
237         struct mdt_thread_info *mti;
238         struct lsd_client_data *tmp;
239         int rc;
240
241         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
242         tmp = &mti->mti_lcd;
243         rc = mdt_record_read(env, mdt->mdt_last_rcvd,
244                              mdt_buf(env, tmp, sizeof(*tmp)), off);
245         if (rc == 0)
246                 lcd_le_to_cpu(tmp, lcd);
247
248         CDEBUG(D_INFO, "read lcd @%d rc = %d:\n"
249                        "uuid = %s\n"
250                        "last_transno = "LPU64"\n"
251                        "last_xid = "LPU64"\n"
252                        "last_result = %u\n"
253                        "last_data = %u\n"
254                        "last_close_transno = "LPU64"\n"
255                        "last_close_xid = "LPU64"\n"
256                        "last_close_result = %u\n",
257                         (int)(*off - sizeof(*tmp)),
258                         rc,
259                         lcd->lcd_uuid,
260                         lcd->lcd_last_transno,
261                         lcd->lcd_last_xid,
262                         lcd->lcd_last_result,
263                         lcd->lcd_last_data,
264                         lcd->lcd_last_close_transno,
265                         lcd->lcd_last_close_xid,
266                         lcd->lcd_last_close_result);
267         return rc;
268 }
269
270 static int mdt_last_rcvd_write(const struct lu_env *env,
271                                struct mdt_device *mdt,
272                                struct lsd_client_data *lcd,
273                                loff_t *off, struct thandle *th)
274 {
275         struct mdt_thread_info *mti;
276         struct lsd_client_data *tmp;
277         int rc;
278
279         LASSERT(th != NULL);
280         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
281         tmp = &mti->mti_lcd;
282
283         lcd_cpu_to_le(lcd, tmp);
284
285         rc = mdt_record_write(env, mdt->mdt_last_rcvd,
286                               mdt_buf_const(env, tmp, sizeof(*tmp)), off, th);
287
288         CDEBUG(D_INFO, "write lcd @%d rc = %d:\n"
289                        "uuid = %s\n"
290                        "last_transno = "LPU64"\n"
291                        "last_xid = "LPU64"\n"
292                        "last_result = %u\n"
293                        "last_data = %u\n"
294                        "last_close_transno = "LPU64"\n"
295                        "last_close_xid = "LPU64"\n"
296                        "last_close_result = %u\n",
297                         (int)(*off - sizeof(*tmp)),
298                         rc,
299                         lcd->lcd_uuid,
300                         lcd->lcd_last_transno,
301                         lcd->lcd_last_xid,
302                         lcd->lcd_last_result,
303                         lcd->lcd_last_data,
304                         lcd->lcd_last_close_transno,
305                         lcd->lcd_last_close_xid,
306                         lcd->lcd_last_close_result);
307         return rc;
308 }
309
310
311 static int mdt_clients_data_init(const struct lu_env *env,
312                                  struct mdt_device *mdt,
313                                  unsigned long last_size)
314 {
315         struct lr_server_data  *lsd = &mdt->mdt_lsd;
316         struct lsd_client_data *lcd = NULL;
317         struct obd_device      *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
318         loff_t off;
319         int cl_idx;
320         int rc = 0;
321         ENTRY;
322
323         /* When we do a clean MDS shutdown, we save the last_transno into
324          * the header.  If we find clients with higher last_transno values
325          * then those clients may need recovery done. */
326         LASSERT(atomic_read(&obd->obd_req_replay_clients) == 0);
327         for (cl_idx = 0, off = lsd->lsd_client_start;
328              off < last_size; cl_idx++) {
329                 __u64 last_transno;
330                 struct obd_export *exp;
331
332                 if (!lcd) {
333                         OBD_ALLOC_PTR(lcd);
334                         if (!lcd)
335                                 RETURN(-ENOMEM);
336                 }
337
338                 off = lsd->lsd_client_start +
339                         cl_idx * lsd->lsd_client_size;
340
341                 rc = mdt_last_rcvd_read(env, mdt, lcd, &off);
342                 if (rc) {
343                         CERROR("error reading MDS %s idx %d, off %llu: rc %d\n",
344                                LAST_RCVD, cl_idx, off, rc);
345                         rc = 0;
346                         break; /* read error shouldn't cause startup to fail */
347                 }
348
349                 if (lcd->lcd_uuid[0] == '\0') {
350                         CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
351                                cl_idx);
352                         continue;
353                 }
354
355                 last_transno = lcd_last_transno(lcd);
356
357                 /* These exports are cleaned up by mdt_obd_disconnect(), so
358                  * they need to be set up like real exports as
359                  * mdt_obd_connect() does.
360                  */
361                 CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64
362                        " srv lr: "LPU64" lx: "LPU64"\n", lcd->lcd_uuid, cl_idx,
363                        last_transno, lsd->lsd_last_transno,
364                        lcd_last_xid(lcd));
365
366                 exp = class_new_export(obd, (struct obd_uuid *)lcd->lcd_uuid);
367                 if (IS_ERR(exp)) {
368                         if (PTR_ERR(exp) == -EALREADY) {
369                                 /* export already exists, zero out this one */
370                                 lcd->lcd_uuid[0] = '\0'; 
371                         } else
372                                 GOTO(err_client, rc = PTR_ERR(exp));
373                 } else {
374                         struct mdt_thread_info *mti;                        
375                         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
376                         LASSERT(mti != NULL);
377                         mti->mti_exp = exp;
378                         exp->exp_mdt_data.med_lcd = lcd;
379                         rc = mdt_client_add(env, mdt, cl_idx);
380                         /* can't fail existing */
381                         LASSERTF(rc == 0, "rc = %d\n", rc); 
382                         lcd = NULL;
383                         spin_lock(&exp->exp_lock);
384                         exp->exp_connecting = 0;
385                         exp->exp_in_recovery = 0;
386                         spin_unlock(&exp->exp_lock);
387                         obd->obd_max_recoverable_clients++;
388                         class_export_put(exp);
389                 }
390
391                 CDEBUG(D_OTHER, "client at idx %d has last_transno="LPU64"\n",
392                        cl_idx, last_transno);
393                 /* protect __u64 value update */
394                 spin_lock(&mdt->mdt_transno_lock);
395                 mdt->mdt_last_transno = max(last_transno,
396                                             mdt->mdt_last_transno);
397                 spin_unlock(&mdt->mdt_transno_lock);
398         }
399
400 err_client:
401         if (lcd)
402                 OBD_FREE_PTR(lcd);
403         RETURN(rc);
404 }
405
406 static int mdt_server_data_init(const struct lu_env *env,
407                                 struct mdt_device *mdt)
408 {
409         struct lr_server_data  *lsd = &mdt->mdt_lsd;
410         struct lsd_client_data *lcd = NULL;
411         struct obd_device      *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
412         struct mdt_thread_info *mti;
413         struct dt_object       *obj;
414         struct lu_attr         *la;
415         unsigned long last_rcvd_size;
416         __u64 mount_count;
417         int rc;
418         ENTRY;
419
420         /* ensure padding in the struct is the correct size */
421         CLASSERT(offsetof(struct lr_server_data, lsd_padding) +
422                 sizeof(lsd->lsd_padding) == LR_SERVER_SIZE);
423         CLASSERT(offsetof(struct lsd_client_data, lcd_padding) +
424                 sizeof(lcd->lcd_padding) == LR_CLIENT_SIZE);
425
426         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
427         LASSERT(mti != NULL);
428         la = &mti->mti_attr.ma_attr;
429
430         obj = mdt->mdt_last_rcvd;
431         rc = obj->do_ops->do_attr_get(env, mdt->mdt_last_rcvd, la, BYPASS_CAPA);
432         if (rc)
433                 RETURN(rc);
434
435         last_rcvd_size = (unsigned long)la->la_size;
436
437         if (last_rcvd_size == 0) {
438                 LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name);
439
440                 memcpy(lsd->lsd_uuid, obd->obd_uuid.uuid,
441                        sizeof(lsd->lsd_uuid));
442                 lsd->lsd_last_transno = 0;
443                 lsd->lsd_mount_count = 0;
444                 lsd->lsd_server_size = LR_SERVER_SIZE;
445                 lsd->lsd_client_start = LR_CLIENT_START;
446                 lsd->lsd_client_size = LR_CLIENT_SIZE;
447                 lsd->lsd_feature_rocompat = OBD_ROCOMPAT_LOVOBJID;
448                 lsd->lsd_feature_incompat = OBD_INCOMPAT_MDT |
449                                                        OBD_INCOMPAT_COMMON_LR;
450         } else {
451                 LCONSOLE_WARN("%s: used disk, loading\n", obd->obd_name);
452                 rc = mdt_last_rcvd_header_read(env, mdt);
453                 if (rc) {
454                         CERROR("error reading MDS %s: rc %d\n", LAST_RCVD, rc);
455                         GOTO(out, rc);
456                 }
457                 if (strcmp(lsd->lsd_uuid, obd->obd_uuid.uuid) != 0) {
458                         LCONSOLE_ERROR_MSG(0x157, "Trying to start OBD %s using"
459                                            "the wrong disk %s. Were the /dev/ "
460                                            "assignments rearranged?\n",
461                                            obd->obd_uuid.uuid, lsd->lsd_uuid);
462                         GOTO(out, rc = -EINVAL);
463                 }
464         }
465         mount_count = lsd->lsd_mount_count;
466
467         lsd->lsd_feature_compat = OBD_COMPAT_MDT;
468
469         spin_lock(&mdt->mdt_transno_lock);
470         mdt->mdt_last_transno = lsd->lsd_last_transno;
471         spin_unlock(&mdt->mdt_transno_lock);
472
473         CDEBUG(D_INODE, "========BEGIN DUMPING LAST_RCVD========\n");
474         CDEBUG(D_INODE, "%s: server last_transno: "LPU64"\n",
475                obd->obd_name, mdt->mdt_last_transno);
476         CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n",
477                obd->obd_name, mount_count + 1);
478         CDEBUG(D_INODE, "%s: server data size: %u\n",
479                obd->obd_name, lsd->lsd_server_size);
480         CDEBUG(D_INODE, "%s: per-client data start: %u\n",
481                obd->obd_name, lsd->lsd_client_start);
482         CDEBUG(D_INODE, "%s: per-client data size: %u\n",
483                obd->obd_name, lsd->lsd_client_size);
484         CDEBUG(D_INODE, "%s: last_rcvd size: %lu\n",
485                obd->obd_name, last_rcvd_size);
486         CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name,
487                last_rcvd_size <= lsd->lsd_client_start ? 0 :
488                (last_rcvd_size - lsd->lsd_client_start) /
489                 lsd->lsd_client_size);
490         CDEBUG(D_INODE, "========END DUMPING LAST_RCVD========\n");
491
492         if (!lsd->lsd_server_size || !lsd->lsd_client_start ||
493             !lsd->lsd_client_size) {
494                 CERROR("Bad last_rcvd contents!\n");
495                 GOTO(out, rc = -EINVAL);
496         }
497
498         rc = mdt_clients_data_init(env, mdt, last_rcvd_size);
499         if (rc)
500                 GOTO(err_client, rc);
501
502         spin_lock(&mdt->mdt_transno_lock);
503         /* obd_last_committed is used for compatibility
504          * with other lustre recovery code */
505         obd->obd_last_committed = mdt->mdt_last_transno;
506         spin_unlock(&mdt->mdt_transno_lock);
507
508         mdt->mdt_mount_count++;
509         lsd->lsd_mount_count = mdt->mdt_mount_count;
510
511         /* save it, so mount count and last_transno is current */
512         rc = mdt_server_data_update(env, mdt);
513         if (rc)
514                 GOTO(err_client, rc);
515
516         RETURN(0);
517
518 err_client:
519         target_recovery_fini(obd);
520 out:
521         return rc;
522 }
523
524 static int mdt_server_data_update(const struct lu_env *env,
525                                   struct mdt_device *mdt)
526 {
527         int rc = 0;
528         ENTRY;
529
530         CDEBUG(D_SUPER, "MDS mount_count is "LPU64", last_transno is "LPU64"\n",
531                mdt->mdt_mount_count, mdt->mdt_last_transno);
532
533         spin_lock(&mdt->mdt_transno_lock);
534         mdt->mdt_lsd.lsd_last_transno = mdt->mdt_last_transno;
535         spin_unlock(&mdt->mdt_transno_lock);
536
537         /*
538          * This may be called from difficult reply handler and
539          * mdt->mdt_last_rcvd may be NULL that time.
540          */
541         if (mdt->mdt_last_rcvd != NULL)
542                 rc = mdt_last_rcvd_header_write(env, mdt);
543         RETURN(rc);
544 }
545
546 void mdt_cb_new_client(const struct mdt_device *mdt, __u64 transno,
547                                   void *data, int err) 
548 {
549         struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
550
551         target_client_add_cb(obd, transno, data, err);
552 }
553
554 int mdt_client_new(const struct lu_env *env, struct mdt_device *mdt)
555 {
556         unsigned long *bitmap = mdt->mdt_client_bitmap;
557         struct mdt_thread_info *mti;
558         struct mdt_export_data *med;
559         struct lsd_client_data *lcd;
560         struct lr_server_data  *lsd = &mdt->mdt_lsd;
561         struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
562         struct thandle *th;
563         loff_t off;
564         int rc;
565         int cl_idx;
566         ENTRY;
567
568         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
569         LASSERT(mti != NULL);
570
571         med = &mti->mti_exp->exp_mdt_data;
572         lcd = med->med_lcd;
573
574         LASSERT(bitmap != NULL);
575         if (!strcmp(med->med_lcd->lcd_uuid, obd->obd_uuid.uuid))
576                 RETURN(0);
577
578         /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
579          * there's no need for extra complication here
580          */
581         spin_lock(&mdt->mdt_client_bitmap_lock);
582         cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
583         if (cl_idx >= LR_MAX_CLIENTS ||
584             OBD_FAIL_CHECK(OBD_FAIL_MDS_CLIENT_ADD)) {
585                 CERROR("no room for %u clients - fix LR_MAX_CLIENTS\n",
586                        cl_idx);
587                 spin_unlock(&mdt->mdt_client_bitmap_lock);
588                 RETURN(-EOVERFLOW);
589         }
590         set_bit(cl_idx, bitmap);
591         spin_unlock(&mdt->mdt_client_bitmap_lock);
592
593         CDEBUG(D_INFO, "client at idx %d with UUID '%s' added\n",
594                cl_idx, med->med_lcd->lcd_uuid);
595
596         med->med_lr_idx = cl_idx;
597         med->med_lr_off = lsd->lsd_client_start +
598                           (cl_idx * lsd->lsd_client_size);
599         init_mutex(&med->med_lcd_lock);
600
601         LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
602         /* write new client data */
603         off = med->med_lr_off;
604         th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
605         if (IS_ERR(th))
606                 RETURN(PTR_ERR(th));
607
608         /* until this operations will be committed the sync is needed for this
609          * export */
610         mdt_trans_add_cb(th, mdt_cb_new_client, mti->mti_exp);
611         spin_lock(&mti->mti_exp->exp_lock);
612         mti->mti_exp->exp_need_sync = 1;
613         spin_unlock(&mti->mti_exp->exp_lock);
614
615         rc = mdt_last_rcvd_write(env, mdt, lcd, &off, th);
616         CDEBUG(D_INFO, "wrote client lcd at idx %u off %llu (len "LPSZ")\n",
617                cl_idx, med->med_lr_off, sizeof(*lcd));
618         mdt_trans_stop(env, mdt, th);
619
620         RETURN(rc);
621 }
622
623 /* Add client data to the MDS.  We use a bitmap to locate a free space
624  * in the last_rcvd file if cl_off is -1 (i.e. a new client).
625  * Otherwise, we just have to read the data from the last_rcvd file and
626  * we know its offset.
627  *
628  * It should not be possible to fail adding an existing client - otherwise
629  * mdt_init_server_data() callsite needs to be fixed.
630  */
631 int mdt_client_add(const struct lu_env *env,
632                    struct mdt_device *mdt, int cl_idx)
633 {
634         struct mdt_thread_info *mti;
635         struct mdt_export_data *med;
636         unsigned long *bitmap = mdt->mdt_client_bitmap;
637         struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
638         struct lr_server_data *lsd = &mdt->mdt_lsd;
639         int rc = 0;
640         ENTRY;
641
642         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
643         LASSERT(mti != NULL);
644
645         med = &mti->mti_exp->exp_mdt_data;
646
647         LASSERT(bitmap != NULL);
648         LASSERTF(cl_idx >= 0, "%d\n", cl_idx);
649
650         if (!strcmp(med->med_lcd->lcd_uuid, obd->obd_uuid.uuid))
651                 RETURN(0);
652
653         spin_lock(&mdt->mdt_client_bitmap_lock);
654         if (test_and_set_bit(cl_idx, bitmap)) {
655                 CERROR("MDS client %d: bit already set in bitmap!!\n",
656                        cl_idx);
657                 LBUG();
658         }
659         spin_unlock(&mdt->mdt_client_bitmap_lock);
660
661         CDEBUG(D_INFO, "client at idx %d with UUID '%s' added\n",
662                cl_idx, med->med_lcd->lcd_uuid);
663
664         med->med_lr_idx = cl_idx;
665         med->med_lr_off = lsd->lsd_client_start +
666                           (cl_idx * lsd->lsd_client_size);
667         init_mutex(&med->med_lcd_lock);
668
669         LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
670
671         RETURN(rc);
672 }
673
674 int mdt_client_del(const struct lu_env *env, struct mdt_device *mdt)
675 {
676         struct mdt_thread_info *mti;
677         struct mdt_export_data *med;
678         struct lsd_client_data *lcd;
679         struct obd_device      *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
680         struct thandle *th;
681         loff_t off;
682         int rc = 0;
683         ENTRY;
684
685         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
686         LASSERT(mti != NULL);
687
688         med = &mti->mti_exp->exp_mdt_data;
689         lcd = med->med_lcd;
690         if (!lcd)
691                 RETURN(0);
692
693         /* XXX: If lcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
694         if (!strcmp(med->med_lcd->lcd_uuid, obd->obd_uuid.uuid))
695                 GOTO(free, 0);
696
697         CDEBUG(D_INFO, "freeing client at idx %u, offset %lld\n",
698                med->med_lr_idx, med->med_lr_off);
699
700         off = med->med_lr_off;
701
702         /*
703          * Don't clear med_lr_idx here as it is likely also unset.  At worst we
704          * leak a client slot that will be cleaned on the next recovery.
705          */
706         if (off <= 0) {
707                 CERROR("client idx %d has offset %lld\n",
708                         med->med_lr_idx, off);
709                 GOTO(free, rc = -EINVAL);
710         }
711
712         /*
713          * Clear the bit _after_ zeroing out the client so we don't race with
714          * mdt_client_add and zero out new clients.
715          */
716         if (!test_bit(med->med_lr_idx, mdt->mdt_client_bitmap)) {
717                 CERROR("MDT client %u: bit already clear in bitmap!!\n",
718                        med->med_lr_idx);
719                 LBUG();
720         }
721
722         /*
723          * This may be called from difficult reply handler path and
724          * mdt->mdt_last_rcvd may be NULL that time.
725          */
726         if (mdt->mdt_last_rcvd != NULL) {
727                 th = mdt_trans_start(env, mdt, MDT_TXN_LAST_RCVD_WRITE_CREDITS);
728                 if (IS_ERR(th))
729                         GOTO(free, rc = PTR_ERR(th));
730
731                 mutex_down(&med->med_lcd_lock);
732                 memset(lcd, 0, sizeof *lcd);
733
734                 rc = mdt_last_rcvd_write(env, mdt, lcd, &off, th);
735                 mutex_up(&med->med_lcd_lock);
736                 mdt_trans_stop(env, mdt, th);
737         }
738
739         CDEBUG(rc == 0 ? D_INFO : D_ERROR, "Zeroing out client idx %u in "
740                "%s rc %d\n",  med->med_lr_idx, LAST_RCVD, rc);
741
742         spin_lock(&mdt->mdt_client_bitmap_lock);
743         clear_bit(med->med_lr_idx, mdt->mdt_client_bitmap);
744         spin_unlock(&mdt->mdt_client_bitmap_lock);
745
746         /*
747          * Make sure the server's last_transno is up to date. Do this after the
748          * client is freed so we know all the client's transactions have been
749          * committed.
750          */
751         mdt_server_data_update(env, mdt);
752         EXIT;
753 free:
754         OBD_FREE_PTR(lcd);
755         med->med_lcd = NULL;
756         return 0;
757 }
758
759 /*
760  * last_rcvd & last_committed update callbacks
761  */
762 static int mdt_last_rcvd_update(struct mdt_thread_info *mti,
763                                 struct thandle *th)
764 {
765         struct mdt_device *mdt = mti->mti_mdt;
766         struct ptlrpc_request *req = mdt_info_req(mti);
767         struct mdt_export_data *med;
768         struct lsd_client_data *lcd;
769         loff_t off;
770         int err;
771         __s32 rc = th->th_result;
772         __u64 *transno_p;
773
774         ENTRY;
775         LASSERT(req);
776         LASSERT(req->rq_export);
777         LASSERT(mdt);
778         med = &req->rq_export->exp_mdt_data;
779         LASSERT(med);
780         lcd = med->med_lcd;
781         /* if the export has already been failed, we have no last_rcvd slot */
782         if (req->rq_export->exp_failed) {
783                 CWARN("commit transaction for disconnected client %s: rc %d\n",
784                       req->rq_export->exp_client_uuid.uuid, rc);
785                 if (rc == 0)
786                         rc = -ENOTCONN;
787                 RETURN(rc);
788         }
789
790         off = med->med_lr_off;
791         mutex_down(&med->med_lcd_lock);
792         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_CLOSE ||
793             lustre_msg_get_opc(req->rq_reqmsg) == MDS_DONE_WRITING) {
794                 transno_p = &lcd->lcd_last_close_transno;
795                 lcd->lcd_last_close_xid = req->rq_xid;
796                 lcd->lcd_last_close_result = rc;
797         } else {
798                 transno_p = &lcd->lcd_last_transno;
799                 lcd->lcd_last_xid = req->rq_xid;
800                 lcd->lcd_last_result = rc;
801                 /*XXX: save intent_disposition in mdt_thread_info?
802                  * also there is bug - intent_dispostion is __u64,
803                  * see struct ldlm_reply->lock_policy_res1; */
804                 lcd->lcd_last_data = mti->mti_opdata;
805         }
806
807         /*
808          * When we store zero transno in lcd we can lost last transno value
809          * because lcd contains 0, but lsd is not yet written
810          * The server data should be updated also if the latest 
811          * transno is rewritten by zero. See the bug 11125 for details.
812          */
813         if (mti->mti_transno == 0 && 
814             *transno_p == mdt->mdt_last_transno)
815                 mdt_server_data_update(mti->mti_env, mdt);
816
817         *transno_p = mti->mti_transno;
818
819         if (off <= 0) {
820                 CERROR("client idx %d has offset %lld\n", med->med_lr_idx, off);
821                 err = -EINVAL;
822         } else {
823                 err = mdt_last_rcvd_write(mti->mti_env, mdt, lcd, &off, th);
824         }
825         mutex_up(&med->med_lcd_lock);
826         RETURN(err);
827 }
828
829 extern struct lu_context_key mdt_thread_key;
830
831 /* add credits for last_rcvd update */
832 static int mdt_txn_start_cb(const struct lu_env *env,
833                             struct txn_param *param, void *cookie)
834 {
835         param->tp_credits += MDT_TXN_LAST_RCVD_WRITE_CREDITS;
836         return 0;
837 }
838
839 /* Update last_rcvd records with latests transaction data */
840 static int mdt_txn_stop_cb(const struct lu_env *env,
841                            struct thandle *txn, void *cookie)
842 {
843         struct mdt_device *mdt = cookie;
844         struct mdt_txn_info *txi;
845         struct mdt_thread_info *mti;
846         struct ptlrpc_request *req;
847
848         /* transno in two contexts - for commit_cb and for thread */
849         txi = lu_context_key_get(&txn->th_ctx, &mdt_txn_key);
850         mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key);
851         req = mdt_info_req(mti);
852
853         if (mti->mti_mdt == NULL || req == NULL || mti->mti_no_need_trans) {
854                 txi->txi_transno = 0;
855                 mti->mti_no_need_trans = 0;
856                 return 0;
857         }
858
859         if (mti->mti_has_trans) {
860                 /* XXX: currently there are allowed cases, but the wrong cases
861                  * are also possible, so better check is needed here */
862                 CDEBUG(D_INFO, "More than one transaction "LPU64"\n", mti->mti_transno);
863                 return 0;
864         }
865
866         mti->mti_has_trans = 1;
867         spin_lock(&mdt->mdt_transno_lock);
868         if (txn->th_result != 0) {
869                 if (mti->mti_transno != 0) {
870                         CERROR("Replay transno "LPU64" failed: rc %i\n",
871                                mti->mti_transno, txn->th_result);
872                         mti->mti_transno = 0;
873                 }
874         } else if (mti->mti_transno == 0) {
875                 mti->mti_transno = ++ mdt->mdt_last_transno;
876         } else {
877                 /* should be replay */
878                 if (mti->mti_transno > mdt->mdt_last_transno)
879                         mdt->mdt_last_transno = mti->mti_transno;
880         }
881
882         /* sometimes the reply message has not been successfully packed */
883         LASSERT(req != NULL && req->rq_repmsg != NULL);
884
885         /* filling reply data */
886         CDEBUG(D_INODE, "transno = %llu, last_committed = %llu\n",
887                mti->mti_transno, req->rq_export->exp_obd->obd_last_committed);
888
889         req->rq_transno = mti->mti_transno;
890         lustre_msg_set_transno(req->rq_repmsg, mti->mti_transno);
891         lustre_msg_set_last_xid(req->rq_repmsg,
892                          lcd_last_xid(req->rq_export->exp_mdt_data.med_lcd));
893         /* save transno for the commit callback */
894         txi->txi_transno = mti->mti_transno;
895         spin_unlock(&mdt->mdt_transno_lock);
896
897         return mdt_last_rcvd_update(mti, txn);
898 }
899
900 /* commit callback, need to update last_commited value */
901 static int mdt_txn_commit_cb(const struct lu_env *env,
902                              struct thandle *txn, void *cookie)
903 {
904         struct mdt_device *mdt = cookie;
905         struct obd_device *obd = md2lu_dev(&mdt->mdt_md_dev)->ld_obd;
906         struct mdt_txn_info *txi;
907         int i;
908
909         txi = lu_context_key_get(&txn->th_ctx, &mdt_txn_key);
910
911         /* copy of obd_transno_commit_cb() but with locking */
912         spin_lock(&mdt->mdt_transno_lock);
913         if (txi->txi_transno > obd->obd_last_committed) {
914                 obd->obd_last_committed = txi->txi_transno;
915                 spin_unlock(&mdt->mdt_transno_lock);
916                 ptlrpc_commit_replies(obd);
917         } else
918                 spin_unlock(&mdt->mdt_transno_lock);
919
920         if (txi->txi_transno)
921                 CDEBUG(D_HA, "%s: transno "LPD64" is committed\n",
922                        obd->obd_name, txi->txi_transno);
923
924         /* iterate through all additional callbacks */
925         for (i = 0; i < txi->txi_cb_count; i++) {
926                 txi->txi_cb[i].mdt_cb_func(mdt, txi->txi_transno,
927                                            txi->txi_cb[i].mdt_cb_data, 0);
928         }
929         return 0;
930 }
931
932 int mdt_fs_setup(const struct lu_env *env, struct mdt_device *mdt,
933                  struct obd_device *obd)
934 {
935         struct lu_fid fid;
936         struct dt_object *o;
937         int rc = 0;
938         ENTRY;
939
940         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FS_SETUP))
941                 RETURN(-ENOENT);
942
943         /* prepare transactions callbacks */
944         mdt->mdt_txn_cb.dtc_txn_start = mdt_txn_start_cb;
945         mdt->mdt_txn_cb.dtc_txn_stop = mdt_txn_stop_cb;
946         mdt->mdt_txn_cb.dtc_txn_commit = mdt_txn_commit_cb;
947         mdt->mdt_txn_cb.dtc_cookie = mdt;
948         CFS_INIT_LIST_HEAD(&mdt->mdt_txn_cb.dtc_linkage);
949
950         dt_txn_callback_add(mdt->mdt_bottom, &mdt->mdt_txn_cb);
951
952         o = dt_store_open(env, mdt->mdt_bottom, LAST_RCVD, &fid);
953         if (!IS_ERR(o)) {
954                 mdt->mdt_last_rcvd = o;
955                 rc = mdt_server_data_init(env, mdt);
956                 if (rc)
957                         GOTO(put_last_rcvd, rc);
958         } else {
959                 rc = PTR_ERR(o);
960                 CERROR("cannot open %s: rc = %d\n", LAST_RCVD, rc);
961                 RETURN(rc);
962         }
963
964         o = dt_store_open(env, mdt->mdt_bottom, CAPA_KEYS, &fid);
965         if (!IS_ERR(o)) {
966                 mdt->mdt_ck_obj = o;
967                 rc = mdt_capa_keys_init(env, mdt);
968                 if (rc)
969                         GOTO(put_ck_object, rc);
970         } else {
971                 rc = PTR_ERR(o);
972                 CERROR("cannot open %s: rc = %d\n", CAPA_KEYS, rc);
973                 GOTO(put_last_rcvd, rc);
974         }
975         RETURN(0);
976
977 put_ck_object:
978         lu_object_put(env, &o->do_lu);
979         mdt->mdt_ck_obj = NULL;
980 put_last_rcvd:
981         lu_object_put(env, &mdt->mdt_last_rcvd->do_lu);
982         mdt->mdt_last_rcvd = NULL;
983         return rc;
984 }
985
986
987 void mdt_fs_cleanup(const struct lu_env *env, struct mdt_device *mdt)
988 {
989         ENTRY;
990
991         /* Remove transaction callback */
992         dt_txn_callback_del(mdt->mdt_bottom, &mdt->mdt_txn_cb);
993         if (mdt->mdt_last_rcvd)
994                 lu_object_put(env, &mdt->mdt_last_rcvd->do_lu);
995         mdt->mdt_last_rcvd = NULL;
996         if (mdt->mdt_ck_obj)
997                 lu_object_put(env, &mdt->mdt_ck_obj->do_lu);
998         mdt->mdt_ck_obj = NULL;
999         EXIT;
1000 }
1001
1002 /* reconstruction code */
1003 static void mdt_steal_ack_locks(struct ptlrpc_request *req)
1004 {
1005         struct obd_export         *exp = req->rq_export;
1006         struct list_head          *tmp;
1007         struct ptlrpc_reply_state *oldrep;
1008         struct ptlrpc_service     *svc;
1009         int                        i;
1010
1011         /* CAVEAT EMPTOR: spinlock order */
1012         spin_lock(&exp->exp_lock);
1013         list_for_each (tmp, &exp->exp_outstanding_replies) {
1014                 oldrep = list_entry(tmp, struct ptlrpc_reply_state,rs_exp_list);
1015
1016                 if (oldrep->rs_xid != req->rq_xid)
1017                         continue;
1018
1019                 if (lustre_msg_get_opc(oldrep->rs_msg) !=
1020                     lustre_msg_get_opc(req->rq_reqmsg))
1021                         CERROR ("Resent req xid "LPX64" has mismatched opc: "
1022                                 "new %d old %d\n", req->rq_xid,
1023                                 lustre_msg_get_opc(req->rq_reqmsg),
1024                                 lustre_msg_get_opc(oldrep->rs_msg));
1025
1026                 svc = oldrep->rs_service;
1027                 spin_lock (&svc->srv_lock);
1028
1029                 list_del_init (&oldrep->rs_exp_list);
1030
1031                 CWARN("Stealing %d locks from rs %p x"LPD64".t"LPD64
1032                       " o%d NID %s\n",
1033                       oldrep->rs_nlocks, oldrep,
1034                       oldrep->rs_xid, oldrep->rs_transno,
1035                       lustre_msg_get_opc(oldrep->rs_msg),
1036                       libcfs_nid2str(exp->exp_connection->c_peer.nid));
1037
1038                 for (i = 0; i < oldrep->rs_nlocks; i++)
1039                         ptlrpc_save_lock(req,
1040                                          &oldrep->rs_locks[i],
1041                                          oldrep->rs_modes[i]);
1042                 oldrep->rs_nlocks = 0;
1043
1044                 DEBUG_REQ(D_HA, req, "stole locks for");
1045                 ptlrpc_schedule_difficult_reply (oldrep);
1046
1047                 spin_unlock (&svc->srv_lock);
1048                 break;
1049         }
1050         spin_unlock(&exp->exp_lock);
1051 }
1052
1053 void mdt_req_from_lcd(struct ptlrpc_request *req,
1054                       struct lsd_client_data *lcd)
1055 {
1056         DEBUG_REQ(D_HA, req, "restoring transno "LPD64"/status %d",
1057                   lcd->lcd_last_transno, lcd->lcd_last_result);
1058
1059         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_CLOSE ||
1060             lustre_msg_get_opc(req->rq_repmsg) == MDS_DONE_WRITING) {
1061                 req->rq_transno = lcd->lcd_last_close_transno;
1062                 req->rq_status = lcd->lcd_last_close_result;
1063                 lustre_msg_set_transno(req->rq_repmsg, req->rq_transno);
1064                 lustre_msg_set_status(req->rq_repmsg, req->rq_status);
1065         } else {
1066                 req->rq_transno = lcd->lcd_last_transno;
1067                 req->rq_status = lcd->lcd_last_result;
1068                 lustre_msg_set_transno(req->rq_repmsg, req->rq_transno);
1069                 lustre_msg_set_status(req->rq_repmsg, req->rq_status);
1070         }
1071         mdt_steal_ack_locks(req);
1072 }
1073
1074 void mdt_reconstruct_generic(struct mdt_thread_info *mti,
1075                              struct mdt_lock_handle *lhc)
1076 {
1077         struct ptlrpc_request *req = mdt_info_req(mti);
1078         struct mdt_export_data *med = &req->rq_export->exp_mdt_data;
1079
1080         return mdt_req_from_lcd(req, med->med_lcd);
1081 }
1082
1083 static void mdt_reconstruct_create(struct mdt_thread_info *mti,
1084                                    struct mdt_lock_handle *lhc)
1085 {
1086         struct ptlrpc_request  *req = mdt_info_req(mti);
1087         struct obd_export *exp = req->rq_export;
1088         struct mdt_export_data *med = &exp->exp_mdt_data;
1089         struct mdt_device *mdt = mti->mti_mdt;
1090         struct mdt_object *child;
1091         struct mdt_body *body;
1092         int rc;
1093
1094         mdt_req_from_lcd(req, med->med_lcd);
1095         if (req->rq_status)
1096                 return;
1097
1098         /* if no error, so child was created with requested fid */
1099         child = mdt_object_find(mti->mti_env, mdt, mti->mti_rr.rr_fid2);
1100         if (IS_ERR(child)) {
1101                 rc = PTR_ERR(child);
1102                 LCONSOLE_WARN("Child "DFID" lookup error %d."
1103                               " Evicting client %s with export %s.\n",
1104                               PFID(mdt_object_fid(child)), rc,
1105                               obd_uuid2str(&exp->exp_client_uuid),
1106                               obd_export_nid2str(exp));
1107                 mdt_export_evict(exp);
1108                 EXIT;
1109                 return;
1110         }
1111
1112         body = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
1113         rc = mo_attr_get(mti->mti_env, mdt_object_child(child), &mti->mti_attr);
1114         if (rc == -EREMOTE) {
1115                 /* object was created on remote server */
1116                 req->rq_status = rc;
1117                 body->valid |= OBD_MD_MDS;
1118         }
1119         mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr, mdt_object_fid(child));
1120         mdt_object_put(mti->mti_env, child);
1121 }
1122
1123 static void mdt_reconstruct_setattr(struct mdt_thread_info *mti,
1124                                     struct mdt_lock_handle *lhc)
1125 {
1126         struct ptlrpc_request  *req = mdt_info_req(mti);
1127         struct obd_export *exp = req->rq_export;
1128         struct mdt_export_data *med = &exp->exp_mdt_data;
1129         struct mdt_device *mdt = mti->mti_mdt;
1130         struct mdt_object *obj;
1131         struct mdt_body *body;
1132
1133         mdt_req_from_lcd(req, med->med_lcd);
1134         if (req->rq_status)
1135                 return;
1136
1137         body = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
1138         obj = mdt_object_find(mti->mti_env, mdt, mti->mti_rr.rr_fid1);
1139         if (IS_ERR(obj)) {
1140                 int rc = PTR_ERR(obj);
1141                 LCONSOLE_WARN(""DFID" lookup error %d."
1142                               " Evicting client %s with export %s.\n",
1143                               PFID(mdt_object_fid(obj)), rc,
1144                               obd_uuid2str(&exp->exp_client_uuid),
1145                               obd_export_nid2str(exp));
1146                 mdt_export_evict(exp);
1147                 EXIT;
1148                 return;
1149         }
1150         mo_attr_get(mti->mti_env, mdt_object_child(obj), &mti->mti_attr);
1151         mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr,
1152                            mdt_object_fid(obj));
1153         if (mti->mti_epoch && (mti->mti_epoch->flags & MF_EPOCH_OPEN)) {
1154                 struct mdt_file_data *mfd;
1155                 struct mdt_body *repbody;
1156
1157                 repbody = req_capsule_server_get(mti->mti_pill, &RMF_MDT_BODY);
1158                 repbody->ioepoch = obj->mot_ioepoch;
1159                 spin_lock(&med->med_open_lock);
1160                 list_for_each_entry(mfd, &med->med_open_head, mfd_list) {
1161                         if (mfd->mfd_xid == req->rq_xid)
1162                                 break;
1163                 }
1164                 LASSERT(&mfd->mfd_list != &med->med_open_head);
1165                 spin_unlock(&med->med_open_lock);
1166                 repbody->handle.cookie = mfd->mfd_handle.h_cookie;
1167         }
1168
1169         mdt_object_put(mti->mti_env, obj);
1170 }
1171
1172 static void mdt_reconstruct_setxattr(struct mdt_thread_info *mti,
1173                                      struct mdt_lock_handle *lhc)
1174 {
1175         /* reply nothing */
1176         req_capsule_shrink(mti->mti_pill, &RMF_EADATA, 0, RCL_SERVER);
1177 }
1178
1179 typedef void (*mdt_reconstructor)(struct mdt_thread_info *mti,
1180                                   struct mdt_lock_handle *lhc);
1181
1182 static mdt_reconstructor reconstructors[REINT_MAX] = {
1183         [REINT_SETATTR]  = mdt_reconstruct_setattr,
1184         [REINT_CREATE]   = mdt_reconstruct_create,
1185         [REINT_LINK]     = mdt_reconstruct_generic,
1186         [REINT_UNLINK]   = mdt_reconstruct_generic,
1187         [REINT_RENAME]   = mdt_reconstruct_generic,
1188         [REINT_OPEN]     = mdt_reconstruct_open,
1189         [REINT_SETXATTR] = mdt_reconstruct_setxattr
1190 };
1191
1192 void mdt_reconstruct(struct mdt_thread_info *mti,
1193                      struct mdt_lock_handle *lhc)
1194 {
1195         ENTRY;
1196         reconstructors[mti->mti_rr.rr_opcode](mti, lhc);
1197         EXIT;
1198 }