1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
5 * Lustre Metadata Target (mdt) open/close file handling
7 * Copyright (C) 2002-2006 Cluster File Systems, Inc.
8 * Author: Huang Hua <huanghua@clusterfs.com>
9 * Author; Pershin Mike <tappro@clusterfs.com>
11 * This file is part of the Lustre file system, http://www.lustre.org
12 * Lustre is a trademark of Cluster File Systems, Inc.
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.
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.
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.
30 # define EXPORT_SYMTAB
32 #define DEBUG_SUBSYSTEM S_MDS
34 #include "mdt_internal.h"
36 /* Add client data to the MDS. We use a bitmap to locate a free space
37 * in the last_rcvd file if cl_off is -1 (i.e. a new client).
38 * Otherwise, we just have to read the data from the last_rcvd file and
41 * It should not be possible to fail adding an existing client - otherwise
42 * mdt_init_server_data() callsite needs to be fixed.
44 int mdt_client_add(const struct lu_context *ctxt,
45 struct mdt_device *mdt,
46 struct mdt_export_data *med,
49 unsigned long *bitmap = mdt->mdt_client_bitmap;
50 struct mdt_client_data *mcd = med->med_mcd;
51 struct mdt_server_data *msd = &mdt->mdt_msd;
52 int new_client = (cl_idx == -1);
56 LASSERT(bitmap != NULL);
57 LASSERTF(cl_idx > -2, "%d\n", cl_idx);
59 /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
60 * there's no need for extra complication here
63 cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
65 if (cl_idx >= LR_MAX_CLIENTS ||
66 OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_CLIENT_ADD)) {
67 CERROR("no room for clients - fix LR_MAX_CLIENTS\n");
70 if (test_and_set_bit(cl_idx, bitmap)) {
71 cl_idx = find_next_zero_bit(bitmap, LR_MAX_CLIENTS,
76 if (test_and_set_bit(cl_idx, bitmap)) {
77 CERROR("MDS client %d: bit already set in bitmap!!\n",
83 CDEBUG(D_INFO, "client at idx %d with UUID '%s' added\n",
84 cl_idx, med->med_mcd->mcd_uuid);
86 med->med_lr_idx = cl_idx;
87 med->med_lr_off = le32_to_cpu(msd->msd_client_start) +
88 (cl_idx * le16_to_cpu(msd->msd_client_size));
89 LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
92 struct dt_object *last = mdt->mdt_last_rcvd;
93 loff_t off = med->med_lr_off;
94 rc = last->do_body_ops->dbo_write(ctxt, last, mcd,
95 sizeof(*mcd), &off, NULL);
96 CDEBUG(D_INFO, "wrote client mcd at idx %u off %llu (len %u)"
98 cl_idx, off, sizeof(mcd), rc);
99 if (rc == sizeof(*mcd))
107 int mdt_update_server_data(const struct lu_context *ctxt,
108 struct mdt_device *mdt,
111 struct mdt_server_data *msd = &mdt->mdt_msd;
114 struct dt_object *last = mdt->mdt_last_rcvd;
117 CDEBUG(D_INODE, "MDS mount_count is "LPU64", last_transno is "LPU64"\n",
118 mdt->mdt_mount_count, mdt->mdt_last_transno);
120 msd->msd_last_transno = cpu_to_le64(mdt->mdt_last_transno);
121 rc = last->do_body_ops->dbo_write(ctxt, last, msd,
122 sizeof(*msd), &off, NULL);
123 if (rc == sizeof(*msd))
131 int mdt_client_free(const struct lu_context *ctxt,
132 struct mdt_device *mdt,
133 struct mdt_export_data *med)
135 struct mdt_client_data *mcd = med->med_mcd;
136 struct dt_object *last = mdt->mdt_last_rcvd;
144 CDEBUG(D_INFO, "freeing client at idx %u, offset %lld\n",
145 med->med_lr_idx, med->med_lr_off);
147 off = med->med_lr_off;
149 /* Don't clear med_lr_idx here as it is likely also unset. At worst
150 * we leak a client slot that will be cleaned on the next recovery. */
152 CERROR("client idx %d has offset %lld\n",
153 med->med_lr_idx, off);
154 GOTO(free, rc = -EINVAL);
157 /* Clear the bit _after_ zeroing out the client so we don't
158 race with mdt_client_add and zero out new clients.*/
159 if (!test_bit(med->med_lr_idx, mdt->mdt_client_bitmap)) {
160 CERROR("MDT client %u: bit already clear in bitmap!!\n",
165 memset(mcd, 0, sizeof *mcd);
166 rc = last->do_body_ops->dbo_write(ctxt, last, mcd,
167 sizeof(*mcd), &off, NULL);
168 if (rc == sizeof(*mcd))
173 CDEBUG(rc == 0 ? D_INFO : D_ERROR,
174 "zeroing out client idx %u in %s rc %d\n",
175 med->med_lr_idx, LAST_RCVD, rc);
177 if (!test_and_clear_bit(med->med_lr_idx, mdt->mdt_client_bitmap)) {
178 CERROR("MDS client %u: bit already clear in bitmap!!\n",
183 /* Make sure the server's last_transno is up to date. Do this
184 * after the client is freed so we know all the client's
185 * transactions have been committed. */
186 mdt_update_server_data(ctxt, mdt, 0);
195 static int mdt_init_server_data(const struct lu_context *ctxt,
196 struct mdt_device *mdt)
198 struct mdt_server_data *msd = &mdt->mdt_msd;
199 struct mdt_client_data *mcd = NULL;
200 struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
202 unsigned long last_rcvd_size = 0; // = getsize(mdt->mdt_last)
206 struct mdt_thread_info *info;
207 struct dt_object *last = mdt->mdt_last_rcvd;
211 /* ensure padding in the struct is the correct size */
212 LASSERT(offsetof(struct mdt_server_data, msd_padding) +
213 sizeof(msd->msd_padding) == LR_SERVER_SIZE);
214 LASSERT(offsetof(struct mdt_client_data, mcd_padding) +
215 sizeof(mcd->mcd_padding) == LR_CLIENT_SIZE);
217 info = lu_context_key_get(ctxt, &mdt_thread_key);
218 LASSERT(info != NULL);
219 la = &info->mti_attr.ma_attr;
221 rc = last->do_ops->do_attr_get(ctxt, last, la);
224 last_rcvd_size = la->la_size;
226 if (last_rcvd_size == 0) {
227 LCONSOLE_WARN("%s: new disk, initializing\n", obd->obd_name);
229 memcpy(msd->msd_uuid, obd->obd_uuid.uuid,sizeof(msd->msd_uuid));
230 msd->msd_last_transno = 0;
231 mount_count = msd->msd_mount_count = 0;
232 msd->msd_server_size = cpu_to_le32(LR_SERVER_SIZE);
233 msd->msd_client_start = cpu_to_le32(LR_CLIENT_START);
234 msd->msd_client_size = cpu_to_le16(LR_CLIENT_SIZE);
235 msd->msd_feature_rocompat = cpu_to_le32(OBD_ROCOMPAT_LOVOBJID);
236 msd->msd_feature_incompat = cpu_to_le32(OBD_INCOMPAT_MDT |
237 OBD_INCOMPAT_COMMON_LR);
239 rc = last->do_body_ops->dbo_read(ctxt, last, msd,
241 if (rc == sizeof(*msd))
247 CERROR("error reading MDS %s: rc %d\n", LAST_RCVD, rc);
250 if (strcmp(msd->msd_uuid, obd->obd_uuid.uuid) != 0) {
251 LCONSOLE_ERROR("Trying to start OBD %s using the wrong"
252 " disk %s. Were the /dev/ assignments "
254 obd->obd_uuid.uuid, msd->msd_uuid);
255 GOTO(out, rc = -EINVAL);
257 mount_count = le64_to_cpu(msd->msd_mount_count);
260 if (msd->msd_feature_incompat & ~cpu_to_le32(MDT_INCOMPAT_SUPP)) {
261 CERROR("%s: unsupported incompat filesystem feature(s) %x\n",
262 obd->obd_name, le32_to_cpu(msd->msd_feature_incompat) &
264 GOTO(out, rc = -EINVAL);
266 if (msd->msd_feature_rocompat & ~cpu_to_le32(MDT_ROCOMPAT_SUPP)) {
267 CERROR("%s: unsupported read-only filesystem feature(s) %x\n",
268 obd->obd_name, le32_to_cpu(msd->msd_feature_rocompat) &
270 /* Do something like remount filesystem read-only */
271 GOTO(out, rc = -EINVAL);
273 if (!(msd->msd_feature_incompat & cpu_to_le32(OBD_INCOMPAT_COMMON_LR))){
274 CDEBUG(D_WARNING, "using old last_rcvd format\n");
275 msd->msd_mount_count = msd->msd_last_transno;
276 msd->msd_last_transno = msd->msd_unused;
277 /* If we update the last_rcvd, we can never go back to
278 an old install, so leave this in the old format for now.
279 msd->msd_feature_incompat |= cpu_to_le32(LR_INCOMPAT_COMMON_LR);
282 msd->msd_feature_compat = cpu_to_le32(OBD_COMPAT_MDT);
284 mdt->mdt_last_transno = le64_to_cpu(msd->msd_last_transno);
286 CDEBUG(D_INODE, "========BEGIN DUMPING LAST_RCVD========\n");
287 CDEBUG(D_INODE, "%s: server last_transno: "LPU64"\n",
288 obd->obd_name, mdt->mdt_last_transno);
289 CDEBUG(D_INODE, "%s: server mount_count: "LPU64"\n",
290 obd->obd_name, mount_count + 1);
291 CDEBUG(D_INODE, "%s: server data size: %u\n",
292 obd->obd_name, le32_to_cpu(msd->msd_server_size));
293 CDEBUG(D_INODE, "%s: per-client data start: %u\n",
294 obd->obd_name, le32_to_cpu(msd->msd_client_start));
295 CDEBUG(D_INODE, "%s: per-client data size: %u\n",
296 obd->obd_name, le32_to_cpu(msd->msd_client_size));
297 CDEBUG(D_INODE, "%s: last_rcvd size: %lu\n",
298 obd->obd_name, last_rcvd_size);
299 CDEBUG(D_INODE, "%s: last_rcvd clients: %lu\n", obd->obd_name,
300 last_rcvd_size <= le32_to_cpu(msd->msd_client_start) ? 0 :
301 (last_rcvd_size - le32_to_cpu(msd->msd_client_start)) /
302 le16_to_cpu(msd->msd_client_size));
303 CDEBUG(D_INODE, "========END DUMPING LAST_RCVD========\n");
305 if (!msd->msd_server_size || !msd->msd_client_start ||
306 !msd->msd_client_size) {
307 CERROR("Bad last_rcvd contents!\n");
308 GOTO(out, rc = -EINVAL);
311 /* When we do a clean MDS shutdown, we save the last_transno into
312 * the header. If we find clients with higher last_transno values
313 * then those clients may need recovery done. */
314 for (cl_idx = 0, off = le32_to_cpu(msd->msd_client_start);
315 off < last_rcvd_size; cl_idx++) {
317 struct obd_export *exp;
318 struct mdt_export_data *med;
323 GOTO(err_client, rc = -ENOMEM);
326 off = le32_to_cpu(msd->msd_client_start) +
327 cl_idx * le16_to_cpu(msd->msd_client_size);
328 rc = last->do_body_ops->dbo_read(ctxt, last, mcd,
330 if (rc == sizeof(*mcd))
336 CERROR("error reading MDS %s idx %d, off %llu: rc %d\n",
337 LAST_RCVD, cl_idx, off, rc);
338 break; /* read error shouldn't cause startup to fail */
341 if (mcd->mcd_uuid[0] == '\0') {
342 CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
347 last_transno = le64_to_cpu(mcd->mcd_last_transno);
349 /* These exports are cleaned up by mdt_obd_disconnect(), so
350 * they need to be set up like real exports as
351 * mdt_obd_connect() does.
353 CDEBUG(D_HA, "RCVRNG CLIENT uuid: %s idx: %d lr: "LPU64
354 " srv lr: "LPU64" lx: "LPU64"\n", mcd->mcd_uuid, cl_idx,
355 last_transno, le64_to_cpu(msd->msd_last_transno),
356 le64_to_cpu(mcd->mcd_last_xid));
358 exp = class_new_export(obd, (struct obd_uuid *)mcd->mcd_uuid);
360 GOTO(err_client, rc = PTR_ERR(exp));
362 med = &exp->exp_mdt_data;
364 rc = mdt_client_add(ctxt, mdt, med, cl_idx);
365 LASSERTF(rc == 0, "rc = %d\n", rc); /* can't fail existing */
368 exp->exp_replay_needed = 1;
369 exp->exp_connecting = 0;
370 obd->obd_recoverable_clients++;
371 obd->obd_max_recoverable_clients++;
372 class_export_put(exp);
374 CDEBUG(D_OTHER, "client at idx %d has last_transno = "LPU64"\n",
375 cl_idx, last_transno);
377 if (last_transno > mdt->mdt_last_transno)
378 mdt->mdt_last_transno = last_transno;
384 obd->obd_last_committed = mdt->mdt_last_transno;
386 if (obd->obd_recoverable_clients) {
387 CWARN("RECOVERY: service %s, %d recoverable clients, "
388 "last_transno "LPU64"\n", obd->obd_name,
389 obd->obd_recoverable_clients, mdt->mdt_last_transno);
390 obd->obd_next_recovery_transno = obd->obd_last_committed + 1;
391 obd->obd_recovering = 1;
392 obd->obd_recovery_start = CURRENT_SECONDS;
393 /* Only used for lprocfs_status */
394 obd->obd_recovery_end = obd->obd_recovery_start +
395 OBD_RECOVERY_TIMEOUT;
398 mdt->mdt_mount_count = mount_count + 1;
399 msd->msd_mount_count = cpu_to_le64(mdt->mdt_mount_count);
401 /* save it, so mount count and last_transno is current */
402 rc = mdt_update_server_data(ctxt, mdt, 1);
404 GOTO(err_client, rc);
409 class_disconnect_exports(obd);
415 * last_rcvd update callbacks
417 extern struct lu_context_key mdt_txn_key;
418 extern struct lu_context_key mdt_thread_key;
421 MDT_TXN_LAST_RCVD_CREDITS = 1
424 /* add credits for last_rcvd update */
425 static int mdt_txn_start_cb(const struct lu_context *ctx,
426 struct dt_device *dev,
427 struct txn_param *param, void *cookie)
429 param->tp_credits += MDT_TXN_LAST_RCVD_CREDITS;
433 /* Update last_rcvd records with latests transaction data */
434 static int mdt_txn_stop_cb(const struct lu_context *ctx,
435 struct dt_device *dev,
436 struct thandle *txn, void *cookie)
438 struct mdt_device *mdt = cookie;
439 struct mdt_txn_info *txni;
440 struct mdt_thread_info *mti;
442 /* transno in two contexts - for commit_cb and for thread */
443 txni = lu_context_key_get(&txn->th_ctx, &mdt_txn_key);
444 mti = lu_context_key_get(ctx, &mdt_thread_key);
445 /*TODO: checks for recovery cases, see mds_finish_transno */
446 spin_lock(&mdt->mdt_transno_lock);
447 if (mti->mti_transno == 0) {
448 mti->mti_transno = ++ mdt->mdt_last_transno;
451 CDEBUG(D_HA, "replaying transno: "LPD64" stopped\n",
453 if (mti->mti_transno > mdt->mdt_last_transno)
454 mdt->mdt_last_transno = mti->mti_transno;
456 spin_unlock(&mdt->mdt_transno_lock);
457 /* save transno for the commit callback */
458 txni->txi_transno = mti->mti_transno;
459 CDEBUG(D_INFO, "transno "LPD64" stopped\n", txni->txi_transno);
461 TODO: write last_rcvd
466 /* commit callback, need to update last_commited value */
467 static int mdt_txn_commit_cb(const struct lu_context *ctx,
468 struct dt_device *dev,
469 struct thandle *txn, void *cookie)
471 struct mdt_device *mdt = cookie;
472 struct obd_device *obd = md2lu_dev(&mdt->mdt_md_dev)->ld_obd;
473 struct mdt_txn_info *txi;
475 txi = lu_context_key_get(&txn->th_ctx, &mdt_txn_key);
476 if (txi->txi_transno > mdt->mdt_last_committed) {
477 mdt->mdt_last_committed = txi->txi_transno;
478 ptlrpc_commit_replies(obd);
480 CDEBUG(D_INFO, "%s: transno "LPD64" committed\n",
481 obd->obd_name, txi->txi_transno);
486 int mdt_fs_setup(const struct lu_context *ctxt,
487 struct mdt_device *mdt)
489 struct lu_fid last_fid;
490 struct dt_object *last;
494 /* prepare transactions callbacks */
495 mdt->mdt_txn_cb.dtc_txn_start = mdt_txn_start_cb;
496 mdt->mdt_txn_cb.dtc_txn_stop = mdt_txn_stop_cb;
497 mdt->mdt_txn_cb.dtc_txn_commit = mdt_txn_commit_cb;
498 mdt->mdt_txn_cb.dtc_cookie = mdt;
500 dt_txn_callback_add(mdt->mdt_bottom, &mdt->mdt_txn_cb);
502 last = dt_store_open(ctxt, mdt->mdt_bottom, LAST_RCVD, &last_fid);
504 mdt->mdt_last_rcvd = last;
505 rc = mdt_init_server_data(ctxt, mdt);
507 lu_object_put(ctxt, &last->do_lu);
508 mdt->mdt_last_rcvd = NULL;
512 CERROR("cannot open %s: rc = %d\n", LAST_RCVD, rc);
519 void mdt_fs_cleanup(const struct lu_context *ctxt,
520 struct mdt_device *mdt)
522 struct obd_device *obd = mdt->mdt_md_dev.md_lu_dev.ld_obd;
524 /* remove transaction callback */
525 dt_txn_callback_del(mdt->mdt_bottom, &mdt->mdt_txn_cb);
527 class_disconnect_exports(obd); /* cleans up client info too */
529 if (mdt->mdt_last_rcvd)
530 lu_object_put(ctxt, &mdt->mdt_last_rcvd->do_lu);
531 mdt->mdt_last_rcvd = NULL;