fs/gfs2/lops.c

   1 /*
   2  * Copyright (C) Sistina Software, Inc.  1997-2003 All rights reserved.
   3  * Copyright (C) 2004-2006 Red Hat, Inc.  All rights reserved.
   4  *
   5  * This copyrighted material is made available to anyone wishing to use,
   6  * modify, copy, or redistribute it subject to the terms and conditions
   7  * of the GNU General Public License version 2.
   8  */
   9
  10 #include <linux/sched.h>
  11 #include <linux/slab.h>
  12 #include <linux/spinlock.h>
  13 #include <linux/completion.h>
  14 #include <linux/buffer_head.h>
  15 #include <linux/gfs2_ondisk.h>
  16 #include <linux/lm_interface.h>
  17
  18 #include "gfs2.h"
  19 #include "incore.h"
  20 #include "glock.h"
  21 #include "log.h"
  22 #include "lops.h"
  23 #include "meta_io.h"
  24 #include "recovery.h"
  25 #include "rgrp.h"
  26 #include "trans.h"
  27 #include "util.h"
  28
  29 static void glock_lo_add(struct gfs2_sbd *sdp, struct gfs2_log_element *le)
  30 {
  31         struct gfs2_glock *gl;
  32         struct gfs2_trans *tr = current->journal_info;
  33
  34         tr->tr_touched = 1;
  35
  36         if (!list_empty(&le->le_list))
  37                 return;
  38
  39         gl = container_of(le, struct gfs2_glock, gl_le);
  40         if (gfs2_assert_withdraw(sdp, gfs2_glock_is_held_excl(gl)))
  41                 return;
  42         gfs2_glock_hold(gl);
  43         set_bit(GLF_DIRTY, &gl->gl_flags);
  44
  45         gfs2_log_lock(sdp);
  46         sdp->sd_log_num_gl++;
  47         list_add(&le->le_list, &sdp->sd_log_le_gl);
  48         gfs2_log_unlock(sdp);
  49 }
  50
  51 static void glock_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_ail *ai)
  52 {
  53         struct list_head *head = &sdp->sd_log_le_gl;
  54         struct gfs2_glock *gl;
  55
  56         while (!list_empty(head)) {
  57                 gl = list_entry(head->next, struct gfs2_glock, gl_le.le_list);
  58                 list_del_init(&gl->gl_le.le_list);
  59                 sdp->sd_log_num_gl--;
  60
  61                 gfs2_assert_withdraw(sdp, gfs2_glock_is_held_excl(gl));
  62                 gfs2_glock_put(gl);
  63         }
  64         gfs2_assert_warn(sdp, !sdp->sd_log_num_gl);
  65 }
  66
  67 static void buf_lo_add(struct gfs2_sbd *sdp, struct gfs2_log_element *le)
  68 {
  69         struct gfs2_bufdata *bd = container_of(le, struct gfs2_bufdata, bd_le);
  70         struct gfs2_trans *tr;
  71
  72         gfs2_log_lock(sdp);
  73         if (!list_empty(&bd->bd_list_tr)) {
  74                 gfs2_log_unlock(sdp);
  75                 return;
  76         }
  77         tr = current->journal_info;
  78         tr->tr_touched = 1;
  79         tr->tr_num_buf++;
  80         list_add(&bd->bd_list_tr, &tr->tr_list_buf);
  81         gfs2_log_unlock(sdp);
  82
  83         if (!list_empty(&le->le_list))
  84                 return;
  85
  86         gfs2_trans_add_gl(bd->bd_gl);
  87
  88         gfs2_meta_check(sdp, bd->bd_bh);
  89         gfs2_pin(sdp, bd->bd_bh);
  90         gfs2_log_lock(sdp);
  91         sdp->sd_log_num_buf++;
  92         list_add(&le->le_list, &sdp->sd_log_le_buf);
  93         gfs2_log_unlock(sdp);
  94
  95         tr->tr_num_buf_new++;
  96 }
  97
  98 static void buf_lo_incore_commit(struct gfs2_sbd *sdp, struct gfs2_trans *tr)
  99 {
 100         struct list_head *head = &tr->tr_list_buf;
 101         struct gfs2_bufdata *bd;
 102
 103         gfs2_log_lock(sdp);
 104         while (!list_empty(head)) {
 105                 bd = list_entry(head->next, struct gfs2_bufdata, bd_list_tr);
 106                 list_del_init(&bd->bd_list_tr);
 107                 tr->tr_num_buf--;
 108         }
 109         gfs2_log_unlock(sdp);
 110         gfs2_assert_warn(sdp, !tr->tr_num_buf);
 111 }
 112
 113 static void buf_lo_before_commit(struct gfs2_sbd *sdp)
 114 {
 115         struct buffer_head *bh;
 116         struct gfs2_log_descriptor *ld;
 117         struct gfs2_bufdata *bd1 = NULL, *bd2;
 118         unsigned int total = sdp->sd_log_num_buf;
 119         unsigned int offset = sizeof(struct gfs2_log_descriptor);
 120         unsigned int limit;
 121         unsigned int num;
 122         unsigned n;
 123         __be64 *ptr;
 124
 125         offset += sizeof(__be64) - 1;
 126         offset &= ~(sizeof(__be64) - 1);
 127         limit = (sdp->sd_sb.sb_bsize - offset)/sizeof(__be64);
 128         /* for 4k blocks, limit = 503 */
 129
 130         bd1 = bd2 = list_prepare_entry(bd1, &sdp->sd_log_le_buf, bd_le.le_list);
 131         while(total) {
 132                 num = total;
 133                 if (total > limit)
 134                         num = limit;
 135                 bh = gfs2_log_get_buf(sdp);
 136                 sdp->sd_log_num_hdrs++;
 137                 ld = (struct gfs2_log_descriptor *)bh->b_data;
 138                 ptr = (__be64 *)(bh->b_data + offset);
 139                 ld->ld_header.mh_magic = cpu_to_be32(GFS2_MAGIC);
 140                 ld->ld_header.mh_type = cpu_to_be32(GFS2_METATYPE_LD);
 141                 ld->ld_header.mh_format = cpu_to_be32(GFS2_FORMAT_LD);
 142                 ld->ld_type = cpu_to_be32(GFS2_LOG_DESC_METADATA);
 143                 ld->ld_length = cpu_to_be32(num + 1);
 144                 ld->ld_data1 = cpu_to_be32(num);
 145                 ld->ld_data2 = cpu_to_be32(0);
 146                 memset(ld->ld_reserved, 0, sizeof(ld->ld_reserved));
 147
 148                 n = 0;
 149                 list_for_each_entry_continue(bd1, &sdp->sd_log_le_buf,
 150                                              bd_le.le_list) {
 151                         *ptr++ = cpu_to_be64(bd1->bd_bh->b_blocknr);
 152                         if (++n >= num)
 153                                 break;
 154                 }
 155
 156                 set_buffer_dirty(bh);
 157                 ll_rw_block(WRITE, 1, &bh);
 158
 159                 n = 0;
 160                 list_for_each_entry_continue(bd2, &sdp->sd_log_le_buf,
 161                                              bd_le.le_list) {
 162                         bh = gfs2_log_fake_buf(sdp, bd2->bd_bh);
 163                         set_buffer_dirty(bh);
 164                         ll_rw_block(WRITE, 1, &bh);
 165                         if (++n >= num)
 166                                 break;
 167                 }
 168
 169                 total -= num;
 170         }
 171 }
 172
 173 static void buf_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_ail *ai)
 174 {
 175         struct list_head *head = &sdp->sd_log_le_buf;
 176         struct gfs2_bufdata *bd;
 177
 178         while (!list_empty(head)) {
 179                 bd = list_entry(head->next, struct gfs2_bufdata, bd_le.le_list);
 180                 list_del_init(&bd->bd_le.le_list);
 181                 sdp->sd_log_num_buf--;
 182
 183                 gfs2_unpin(sdp, bd->bd_bh, ai);
 184         }
 185         gfs2_assert_warn(sdp, !sdp->sd_log_num_buf);
 186 }
 187
 188 static void buf_lo_before_scan(struct gfs2_jdesc *jd,
 189                                struct gfs2_log_header_host *head, int pass)
 190 {
 191         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 192
 193         if (pass != 0)
 194                 return;
 195
 196         sdp->sd_found_blocks = 0;
 197         sdp->sd_replayed_blocks = 0;
 198 }
 199
 200 static int buf_lo_scan_elements(struct gfs2_jdesc *jd, unsigned int start,
 201                                 struct gfs2_log_descriptor *ld, __be64 *ptr,
 202                                 int pass)
 203 {
 204         struct gfs2_inode *ip = GFS2_I(jd->jd_inode);
 205         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 206         struct gfs2_glock *gl = ip->i_gl;
 207         unsigned int blks = be32_to_cpu(ld->ld_data1);
 208         struct buffer_head *bh_log, *bh_ip;
 209         u64 blkno;
 210         int error = 0;
 211
 212         if (pass != 1 || be32_to_cpu(ld->ld_type) != GFS2_LOG_DESC_METADATA)
 213                 return 0;
 214
 215         gfs2_replay_incr_blk(sdp, &start);
 216
 217         for (; blks; gfs2_replay_incr_blk(sdp, &start), blks--) {
 218                 blkno = be64_to_cpu(*ptr++);
 219
 220                 sdp->sd_found_blocks++;
 221
 222                 if (gfs2_revoke_check(sdp, blkno, start))
 223                         continue;
 224
 225                 error = gfs2_replay_read_block(jd, start, &bh_log);
 226                 if (error)
 227                         return error;
 228
 229                 bh_ip = gfs2_meta_new(gl, blkno);
 230                 memcpy(bh_ip->b_data, bh_log->b_data, bh_log->b_size);
 231
 232                 if (gfs2_meta_check(sdp, bh_ip))
 233                         error = -EIO;
 234                 else
 235                         mark_buffer_dirty(bh_ip);
 236
 237                 brelse(bh_log);
 238                 brelse(bh_ip);
 239
 240                 if (error)
 241                         break;
 242
 243                 sdp->sd_replayed_blocks++;
 244         }
 245
 246         return error;
 247 }
 248
 249 static void buf_lo_after_scan(struct gfs2_jdesc *jd, int error, int pass)
 250 {
 251         struct gfs2_inode *ip = GFS2_I(jd->jd_inode);
 252         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 253
 254         if (error) {
 255                 gfs2_meta_sync(ip->i_gl);
 256                 return;
 257         }
 258         if (pass != 1)
 259                 return;
 260
 261         gfs2_meta_sync(ip->i_gl);
 262
 263         fs_info(sdp, "jid=%u: Replayed %u of %u blocks\n",
 264                 jd->jd_jid, sdp->sd_replayed_blocks, sdp->sd_found_blocks);
 265 }
 266
 267 static void revoke_lo_add(struct gfs2_sbd *sdp, struct gfs2_log_element *le)
 268 {
 269         struct gfs2_trans *tr;
 270
 271         tr = current->journal_info;
 272         tr->tr_touched = 1;
 273         tr->tr_num_revoke++;
 274
 275         gfs2_log_lock(sdp);
 276         sdp->sd_log_num_revoke++;
 277         list_add(&le->le_list, &sdp->sd_log_le_revoke);
 278         gfs2_log_unlock(sdp);
 279 }
 280
 281 static void revoke_lo_before_commit(struct gfs2_sbd *sdp)
 282 {
 283         struct gfs2_log_descriptor *ld;
 284         struct gfs2_meta_header *mh;
 285         struct buffer_head *bh;
 286         unsigned int offset;
 287         struct list_head *head = &sdp->sd_log_le_revoke;
 288         struct gfs2_revoke *rv;
 289
 290         if (!sdp->sd_log_num_revoke)
 291                 return;
 292
 293         bh = gfs2_log_get_buf(sdp);
 294         ld = (struct gfs2_log_descriptor *)bh->b_data;
 295         ld->ld_header.mh_magic = cpu_to_be32(GFS2_MAGIC);
 296         ld->ld_header.mh_type = cpu_to_be32(GFS2_METATYPE_LD);
 297         ld->ld_header.mh_format = cpu_to_be32(GFS2_FORMAT_LD);
 298         ld->ld_type = cpu_to_be32(GFS2_LOG_DESC_REVOKE);
 299         ld->ld_length = cpu_to_be32(gfs2_struct2blk(sdp, sdp->sd_log_num_revoke,
 300                                                     sizeof(u64)));
 301         ld->ld_data1 = cpu_to_be32(sdp->sd_log_num_revoke);
 302         ld->ld_data2 = cpu_to_be32(0);
 303         memset(ld->ld_reserved, 0, sizeof(ld->ld_reserved));
 304         offset = sizeof(struct gfs2_log_descriptor);
 305
 306         while (!list_empty(head)) {
 307                 rv = list_entry(head->next, struct gfs2_revoke, rv_le.le_list);
 308                 list_del_init(&rv->rv_le.le_list);
 309                 sdp->sd_log_num_revoke--;
 310
 311                 if (offset + sizeof(u64) > sdp->sd_sb.sb_bsize) {
 312                         set_buffer_dirty(bh);
 313                         ll_rw_block(WRITE, 1, &bh);
 314
 315                         bh = gfs2_log_get_buf(sdp);
 316                         mh = (struct gfs2_meta_header *)bh->b_data;
 317                         mh->mh_magic = cpu_to_be32(GFS2_MAGIC);
 318                         mh->mh_type = cpu_to_be32(GFS2_METATYPE_LB);
 319                         mh->mh_format = cpu_to_be32(GFS2_FORMAT_LB);
 320                         offset = sizeof(struct gfs2_meta_header);
 321                 }
 322
 323                 *(__be64 *)(bh->b_data + offset) = cpu_to_be64(rv->rv_blkno);
 324                 kfree(rv);
 325
 326                 offset += sizeof(u64);
 327         }
 328         gfs2_assert_withdraw(sdp, !sdp->sd_log_num_revoke);
 329
 330         set_buffer_dirty(bh);
 331         ll_rw_block(WRITE, 1, &bh);
 332 }
 333
 334 static void revoke_lo_before_scan(struct gfs2_jdesc *jd,
 335                                   struct gfs2_log_header_host *head, int pass)
 336 {
 337         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 338
 339         if (pass != 0)
 340                 return;
 341
 342         sdp->sd_found_revokes = 0;
 343         sdp->sd_replay_tail = head->lh_tail;
 344 }
 345
 346 static int revoke_lo_scan_elements(struct gfs2_jdesc *jd, unsigned int start,
 347                                    struct gfs2_log_descriptor *ld, __be64 *ptr,
 348                                    int pass)
 349 {
 350         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 351         unsigned int blks = be32_to_cpu(ld->ld_length);
 352         unsigned int revokes = be32_to_cpu(ld->ld_data1);
 353         struct buffer_head *bh;
 354         unsigned int offset;
 355         u64 blkno;
 356         int first = 1;
 357         int error;
 358
 359         if (pass != 0 || be32_to_cpu(ld->ld_type) != GFS2_LOG_DESC_REVOKE)
 360                 return 0;
 361
 362         offset = sizeof(struct gfs2_log_descriptor);
 363
 364         for (; blks; gfs2_replay_incr_blk(sdp, &start), blks--) {
 365                 error = gfs2_replay_read_block(jd, start, &bh);
 366                 if (error)
 367                         return error;
 368
 369                 if (!first)
 370                         gfs2_metatype_check(sdp, bh, GFS2_METATYPE_LB);
 371
 372                 while (offset + sizeof(u64) <= sdp->sd_sb.sb_bsize) {
 373                         blkno = be64_to_cpu(*(__be64 *)(bh->b_data + offset));
 374
 375                         error = gfs2_revoke_add(sdp, blkno, start);
 376                         if (error < 0)
 377                                 return error;
 378                         else if (error)
 379                                 sdp->sd_found_revokes++;
 380
 381                         if (!--revokes)
 382                                 break;
 383                         offset += sizeof(u64);
 384                 }
 385
 386                 brelse(bh);
 387                 offset = sizeof(struct gfs2_meta_header);
 388                 first = 0;
 389         }
 390
 391         return 0;
 392 }
 393
 394 static void revoke_lo_after_scan(struct gfs2_jdesc *jd, int error, int pass)
 395 {
 396         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 397
 398         if (error) {
 399                 gfs2_revoke_clean(sdp);
 400                 return;
 401         }
 402         if (pass != 1)
 403                 return;
 404
 405         fs_info(sdp, "jid=%u: Found %u revoke tags\n",
 406                 jd->jd_jid, sdp->sd_found_revokes);
 407
 408         gfs2_revoke_clean(sdp);
 409 }
 410
 411 static void rg_lo_add(struct gfs2_sbd *sdp, struct gfs2_log_element *le)
 412 {
 413         struct gfs2_rgrpd *rgd;
 414         struct gfs2_trans *tr = current->journal_info;
 415
 416         tr->tr_touched = 1;
 417
 418         if (!list_empty(&le->le_list))
 419                 return;
 420
 421         rgd = container_of(le, struct gfs2_rgrpd, rd_le);
 422         gfs2_rgrp_bh_hold(rgd);
 423
 424         gfs2_log_lock(sdp);
 425         sdp->sd_log_num_rg++;
 426         list_add(&le->le_list, &sdp->sd_log_le_rg);
 427         gfs2_log_unlock(sdp);
 428 }
 429
 430 static void rg_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_ail *ai)
 431 {
 432         struct list_head *head = &sdp->sd_log_le_rg;
 433         struct gfs2_rgrpd *rgd;
 434
 435         while (!list_empty(head)) {
 436                 rgd = list_entry(head->next, struct gfs2_rgrpd, rd_le.le_list);
 437                 list_del_init(&rgd->rd_le.le_list);
 438                 sdp->sd_log_num_rg--;
 439
 440                 gfs2_rgrp_repolish_clones(rgd);
 441                 gfs2_rgrp_bh_put(rgd);
 442         }
 443         gfs2_assert_warn(sdp, !sdp->sd_log_num_rg);
 444 }
 445
 446 /**
 447  * databuf_lo_add - Add a databuf to the transaction.
 448  *
 449  * This is used in two distinct cases:
 450  * i) In ordered write mode
 451  *    We put the data buffer on a list so that we can ensure that its
 452  *    synced to disk at the right time
 453  * ii) In journaled data mode
 454  *    We need to journal the data block in the same way as metadata in
 455  *    the functions above. The difference is that here we have a tag
 456  *    which is two __be64's being the block number (as per meta data)
 457  *    and a flag which says whether the data block needs escaping or
 458  *    not. This means we need a new log entry for each 251 or so data
 459  *    blocks, which isn't an enormous overhead but twice as much as
 460  *    for normal metadata blocks.
 461  */
 462 static void databuf_lo_add(struct gfs2_sbd *sdp, struct gfs2_log_element *le)
 463 {
 464         struct gfs2_bufdata *bd = container_of(le, struct gfs2_bufdata, bd_le);
 465         struct gfs2_trans *tr = current->journal_info;
 466         struct address_space *mapping = bd->bd_bh->b_page->mapping;
 467         struct gfs2_inode *ip = GFS2_I(mapping->host);
 468
 469         gfs2_log_lock(sdp);
 470         tr->tr_touched = 1;
 471         if (list_empty(&bd->bd_list_tr) &&
 472             (ip->i_di.di_flags & GFS2_DIF_JDATA)) {
 473                 tr->tr_num_buf++;
 474                 list_add(&bd->bd_list_tr, &tr->tr_list_buf);
 475                 gfs2_log_unlock(sdp);
 476                 gfs2_pin(sdp, bd->bd_bh);
 477                 tr->tr_num_buf_new++;
 478         } else {
 479                 gfs2_log_unlock(sdp);
 480         }
 481         gfs2_trans_add_gl(bd->bd_gl);
 482         gfs2_log_lock(sdp);
 483         if (list_empty(&le->le_list)) {
 484                 if (ip->i_di.di_flags & GFS2_DIF_JDATA)
 485                         sdp->sd_log_num_jdata++;
 486                 sdp->sd_log_num_databuf++;
 487                 list_add(&le->le_list, &sdp->sd_log_le_databuf);
 488         }
 489         gfs2_log_unlock(sdp);
 490 }
 491
 492 static int gfs2_check_magic(struct buffer_head *bh)
 493 {
 494         struct page *page = bh->b_page;
 495         void *kaddr;
 496         __be32 *ptr;
 497         int rv = 0;
 498
 499         kaddr = kmap_atomic(page, KM_USER0);
 500         ptr = kaddr + bh_offset(bh);
 501         if (*ptr == cpu_to_be32(GFS2_MAGIC))
 502                 rv = 1;
 503         kunmap_atomic(kaddr, KM_USER0);
 504
 505         return rv;
 506 }
 507
 508 /**
 509  * databuf_lo_before_commit - Scan the data buffers, writing as we go
 510  *
 511  * Here we scan through the lists of buffers and make the assumption
 512  * that any buffer thats been pinned is being journaled, and that
 513  * any unpinned buffer is an ordered write data buffer and therefore
 514  * will be written back rather than journaled.
 515  */
 516 static void databuf_lo_before_commit(struct gfs2_sbd *sdp)
 517 {
 518         LIST_HEAD(started);
 519         struct gfs2_bufdata *bd1 = NULL, *bd2, *bdt;
 520         struct buffer_head *bh = NULL,*bh1 = NULL;
 521         unsigned int offset = sizeof(struct gfs2_log_descriptor);
 522         struct gfs2_log_descriptor *ld;
 523         unsigned int limit;
 524         unsigned int total_dbuf = sdp->sd_log_num_databuf;
 525         unsigned int total_jdata = sdp->sd_log_num_jdata;
 526         unsigned int num, n;
 527         __be64 *ptr = NULL;
 528
 529         offset += 2*sizeof(__be64) - 1;
 530         offset &= ~(2*sizeof(__be64) - 1);
 531         limit = (sdp->sd_sb.sb_bsize - offset)/sizeof(__be64);
 532
 533         /*
 534          * Start writing ordered buffers, write journaled buffers
 535          * into the log along with a header
 536          */
 537         gfs2_log_lock(sdp);
 538         bd2 = bd1 = list_prepare_entry(bd1, &sdp->sd_log_le_databuf,
 539                                        bd_le.le_list);
 540         while(total_dbuf) {
 541                 num = total_jdata;
 542                 if (num > limit)
 543                         num = limit;
 544                 n = 0;
 545                 list_for_each_entry_safe_continue(bd1, bdt,
 546                                                   &sdp->sd_log_le_databuf,
 547                                                   bd_le.le_list) {
 548                         /* store off the buffer head in a local ptr since
 549                          * gfs2_bufdata might change when we drop the log lock
 550                          */
 551                         bh1 = bd1->bd_bh;
 552
 553                         /* An ordered write buffer */
 554                         if (bh1 && !buffer_pinned(bh1)) {
 555                                 list_move(&bd1->bd_le.le_list, &started);
 556                                 if (bd1 == bd2) {
 557                                         bd2 = NULL;
 558                                         bd2 = list_prepare_entry(bd2,
 559                                                         &sdp->sd_log_le_databuf,
 560                                                         bd_le.le_list);
 561                                 }
 562                                 total_dbuf--;
 563                                 if (bh1) {
 564                                         if (buffer_dirty(bh1)) {
 565                                                 get_bh(bh1);
 566
 567                                                 gfs2_log_unlock(sdp);
 568
 569                                                 ll_rw_block(SWRITE, 1, &bh1);
 570                                                 brelse(bh1);
 571
 572                                                 gfs2_log_lock(sdp);
 573                                         }
 574                                         continue;
 575                                 }
 576                                 continue;
 577                         } else if (bh1) { /* A journaled buffer */
 578                                 int magic;
 579                                 gfs2_log_unlock(sdp);
 580                                 if (!bh) {
 581                                         bh = gfs2_log_get_buf(sdp);
 582                                         sdp->sd_log_num_hdrs++;
 583                                         ld = (struct gfs2_log_descriptor *)
 584                                              bh->b_data;
 585                                         ptr = (__be64 *)(bh->b_data + offset);
 586                                         ld->ld_header.mh_magic =
 587                                                 cpu_to_be32(GFS2_MAGIC);
 588                                         ld->ld_header.mh_type =
 589                                                 cpu_to_be32(GFS2_METATYPE_LD);
 590                                         ld->ld_header.mh_format =
 591                                                 cpu_to_be32(GFS2_FORMAT_LD);
 592                                         ld->ld_type =
 593                                                 cpu_to_be32(GFS2_LOG_DESC_JDATA);
 594                                         ld->ld_length = cpu_to_be32(num + 1);
 595                                         ld->ld_data1 = cpu_to_be32(num);
 596                                         ld->ld_data2 = cpu_to_be32(0);
 597                                         memset(ld->ld_reserved, 0, sizeof(ld->ld_reserved));
 598                                 }
 599                                 magic = gfs2_check_magic(bh1);
 600                                 *ptr++ = cpu_to_be64(bh1->b_blocknr);
 601                                 *ptr++ = cpu_to_be64((__u64)magic);
 602                                 clear_buffer_escaped(bh1);
 603                                 if (unlikely(magic != 0))
 604                                         set_buffer_escaped(bh1);
 605                                 gfs2_log_lock(sdp);
 606                                 if (n++ > num)
 607                                         break;
 608                         } else if (!bh1) {
 609                                 total_dbuf--;
 610                                 sdp->sd_log_num_databuf--;
 611                                 list_del_init(&bd1->bd_le.le_list);
 612                                 if (bd1 == bd2) {
 613                                         bd2 = NULL;
 614                                         bd2 = list_prepare_entry(bd2,
 615                                                 &sdp->sd_log_le_databuf,
 616                                                 bd_le.le_list);
 617                                 }
 618                                 kmem_cache_free(gfs2_bufdata_cachep, bd1);
 619                         }
 620                 }
 621                 gfs2_log_unlock(sdp);
 622                 if (bh) {
 623                         set_buffer_dirty(bh);
 624                         ll_rw_block(WRITE, 1, &bh);
 625                         bh = NULL;
 626                 }
 627                 n = 0;
 628                 gfs2_log_lock(sdp);
 629                 list_for_each_entry_continue(bd2, &sdp->sd_log_le_databuf,
 630                                              bd_le.le_list) {
 631                         if (!bd2->bd_bh)
 632                                 continue;
 633                         /* copy buffer if it needs escaping */
 634                         gfs2_log_unlock(sdp);
 635                         if (unlikely(buffer_escaped(bd2->bd_bh))) {
 636                                 void *kaddr;
 637                                 struct page *page = bd2->bd_bh->b_page;
 638                                 bh = gfs2_log_get_buf(sdp);
 639                                 kaddr = kmap_atomic(page, KM_USER0);
 640                                 memcpy(bh->b_data,
 641                                        kaddr + bh_offset(bd2->bd_bh),
 642                                        sdp->sd_sb.sb_bsize);
 643                                 kunmap_atomic(kaddr, KM_USER0);
 644                                 *(__be32 *)bh->b_data = 0;
 645                         } else {
 646                                 bh = gfs2_log_fake_buf(sdp, bd2->bd_bh);
 647                         }
 648                         set_buffer_dirty(bh);
 649                         ll_rw_block(WRITE, 1, &bh);
 650                         gfs2_log_lock(sdp);
 651                         if (++n >= num)
 652                                 break;
 653                 }
 654                 bh = NULL;
 655                 total_dbuf -= num;
 656                 total_jdata -= num;
 657         }
 658         gfs2_log_unlock(sdp);
 659
 660         /* Wait on all ordered buffers */
 661         while (!list_empty(&started)) {
 662                 gfs2_log_lock(sdp);
 663                 bd1 = list_entry(started.next, struct gfs2_bufdata,
 664                                  bd_le.le_list);
 665                 list_del_init(&bd1->bd_le.le_list);
 666                 sdp->sd_log_num_databuf--;
 667                 bh = bd1->bd_bh;
 668                 if (bh) {
 669                         bh->b_private = NULL;
 670                         get_bh(bh);
 671                         gfs2_log_unlock(sdp);
 672                         wait_on_buffer(bh);
 673                         brelse(bh);
 674                 } else
 675                         gfs2_log_unlock(sdp);
 676
 677                 kmem_cache_free(gfs2_bufdata_cachep, bd1);
 678         }
 679
 680         /* We've removed all the ordered write bufs here, so only jdata left */
 681         gfs2_assert_warn(sdp, sdp->sd_log_num_databuf == sdp->sd_log_num_jdata);
 682 }
 683
 684 static int databuf_lo_scan_elements(struct gfs2_jdesc *jd, unsigned int start,
 685                                     struct gfs2_log_descriptor *ld,
 686                                     __be64 *ptr, int pass)
 687 {
 688         struct gfs2_inode *ip = GFS2_I(jd->jd_inode);
 689         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 690         struct gfs2_glock *gl = ip->i_gl;
 691         unsigned int blks = be32_to_cpu(ld->ld_data1);
 692         struct buffer_head *bh_log, *bh_ip;
 693         u64 blkno;
 694         u64 esc;
 695         int error = 0;
 696
 697         if (pass != 1 || be32_to_cpu(ld->ld_type) != GFS2_LOG_DESC_JDATA)
 698                 return 0;
 699
 700         gfs2_replay_incr_blk(sdp, &start);
 701         for (; blks; gfs2_replay_incr_blk(sdp, &start), blks--) {
 702                 blkno = be64_to_cpu(*ptr++);
 703                 esc = be64_to_cpu(*ptr++);
 704
 705                 sdp->sd_found_blocks++;
 706
 707                 if (gfs2_revoke_check(sdp, blkno, start))
 708                         continue;
 709
 710                 error = gfs2_replay_read_block(jd, start, &bh_log);
 711                 if (error)
 712                         return error;
 713
 714                 bh_ip = gfs2_meta_new(gl, blkno);
 715                 memcpy(bh_ip->b_data, bh_log->b_data, bh_log->b_size);
 716
 717                 /* Unescape */
 718                 if (esc) {
 719                         __be32 *eptr = (__be32 *)bh_ip->b_data;
 720                         *eptr = cpu_to_be32(GFS2_MAGIC);
 721                 }
 722                 mark_buffer_dirty(bh_ip);
 723
 724                 brelse(bh_log);
 725                 brelse(bh_ip);
 726                 if (error)
 727                         break;
 728
 729                 sdp->sd_replayed_blocks++;
 730         }
 731
 732         return error;
 733 }
 734
 735 /* FIXME: sort out accounting for log blocks etc. */
 736
 737 static void databuf_lo_after_scan(struct gfs2_jdesc *jd, int error, int pass)
 738 {
 739         struct gfs2_inode *ip = GFS2_I(jd->jd_inode);
 740         struct gfs2_sbd *sdp = GFS2_SB(jd->jd_inode);
 741
 742         if (error) {
 743                 gfs2_meta_sync(ip->i_gl);
 744                 return;
 745         }
 746         if (pass != 1)
 747                 return;
 748
 749         /* data sync? */
 750         gfs2_meta_sync(ip->i_gl);
 751
 752         fs_info(sdp, "jid=%u: Replayed %u of %u data blocks\n",
 753                 jd->jd_jid, sdp->sd_replayed_blocks, sdp->sd_found_blocks);
 754 }
 755
 756 static void databuf_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_ail *ai)
 757 {
 758         struct list_head *head = &sdp->sd_log_le_databuf;
 759         struct gfs2_bufdata *bd;
 760
 761         while (!list_empty(head)) {
 762                 bd = list_entry(head->next, struct gfs2_bufdata, bd_le.le_list);
 763                 list_del_init(&bd->bd_le.le_list);
 764                 sdp->sd_log_num_databuf--;
 765                 sdp->sd_log_num_jdata--;
 766                 gfs2_unpin(sdp, bd->bd_bh, ai);
 767         }
 768         gfs2_assert_warn(sdp, !sdp->sd_log_num_databuf);
 769         gfs2_assert_warn(sdp, !sdp->sd_log_num_jdata);
 770 }
 771
 772
 773 const struct gfs2_log_operations gfs2_glock_lops = {
 774         .lo_add = glock_lo_add,
 775         .lo_after_commit = glock_lo_after_commit,
 776         .lo_name = "glock",
 777 };
 778
 779 const struct gfs2_log_operations gfs2_buf_lops = {
 780         .lo_add = buf_lo_add,
 781         .lo_incore_commit = buf_lo_incore_commit,
 782         .lo_before_commit = buf_lo_before_commit,
 783         .lo_after_commit = buf_lo_after_commit,
 784         .lo_before_scan = buf_lo_before_scan,
 785         .lo_scan_elements = buf_lo_scan_elements,
 786         .lo_after_scan = buf_lo_after_scan,
 787         .lo_name = "buf",
 788 };
 789
 790 const struct gfs2_log_operations gfs2_revoke_lops = {
 791         .lo_add = revoke_lo_add,
 792         .lo_before_commit = revoke_lo_before_commit,
 793         .lo_before_scan = revoke_lo_before_scan,
 794         .lo_scan_elements = revoke_lo_scan_elements,
 795         .lo_after_scan = revoke_lo_after_scan,
 796         .lo_name = "revoke",
 797 };
 798
 799 const struct gfs2_log_operations gfs2_rg_lops = {
 800         .lo_add = rg_lo_add,
 801         .lo_after_commit = rg_lo_after_commit,
 802         .lo_name = "rg",
 803 };
 804
 805 const struct gfs2_log_operations gfs2_databuf_lops = {
 806         .lo_add = databuf_lo_add,
 807         .lo_incore_commit = buf_lo_incore_commit,
 808         .lo_before_commit = databuf_lo_before_commit,
 809         .lo_after_commit = databuf_lo_after_commit,
 810         .lo_scan_elements = databuf_lo_scan_elements,
 811         .lo_after_scan = databuf_lo_after_scan,
 812         .lo_name = "databuf",
 813 };
 814
 815 const struct gfs2_log_operations *gfs2_log_ops[] = {
 816         &gfs2_glock_lops,
 817         &gfs2_buf_lops,
 818         &gfs2_revoke_lops,
 819         &gfs2_rg_lops,
 820         &gfs2_databuf_lops,
 821         NULL,
 822 };
 823