lustre/cmobd/cm_write.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002 Cluster File Systems, Inc. <info@clusterfs.com>
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 #define DEBUG_SUBSYSTEM S_CMOBD
  23
  24 #include <linux/version.h>
  25 #include <linux/init.h>
  26 #include <linux/obd_support.h>
  27 #include <linux/lustre_lib.h>
  28 #include <linux/lustre_net.h>
  29 #include <linux/lustre_idl.h>
  30 #include <linux/obd_class.h>
  31 #include <linux/lustre_mds.h>
  32 #include <linux/lustre_cmobd.h>
  33
  34 #include <asm/div64.h>
  35 #include <linux/pagemap.h>
  36
  37 #include "cm_internal.h"
  38
  39 extern kmem_cache_t *cmobd_extent_slab;
  40
  41 /* helper function to split an extent */
  42 static obd_count split_extent(struct ldlm_extent *ext, unsigned long interval)
  43 {
  44         obd_count buf_count, remainder;
  45         ENTRY;
  46
  47         buf_count = ext->end - ext->start + 1;
  48         LASSERT(buf_count > 0);
  49
  50         remainder = do_div(buf_count, interval);
  51         if (remainder)
  52                 buf_count++;
  53
  54         RETURN(buf_count);
  55 }
  56
  57 static int cmobd_ap_make_ready(void *data, int cmd)
  58 {
  59         struct cmobd_async_page *cmap = (struct cmobd_async_page *)data;
  60         struct page *page = cmap->cmap_page;
  61         ENTRY;
  62
  63         if (cmd == OBD_BRW_READ)
  64                 RETURN(0);
  65
  66         if (TryLockPage(page))
  67                 RETURN(-EAGAIN);
  68
  69         RETURN(0);
  70 }
  71
  72 static int cmobd_ap_refresh_count(void *data, int cmd)
  73 {
  74         struct cmobd_async_page *cmap = (struct cmobd_async_page *)data;
  75         struct page *page = cmap->cmap_page;
  76         struct inode *inode = page->mapping->host;
  77         ENTRY;
  78
  79         LASSERT(cmd != OBD_BRW_READ);
  80
  81         /* catch race with truncate */
  82         if (((loff_t)page->index << PAGE_SHIFT) >= inode->i_size)
  83                 RETURN(0);
  84
  85         /* catch sub-page write at end of file */
  86         if (((loff_t)page->index << PAGE_SHIFT) + PAGE_SIZE > inode->i_size)
  87                 RETURN(inode->i_size % PAGE_SIZE);
  88
  89         RETURN(PAGE_SIZE);
  90 }
  91
  92 static void cmobd_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
  93 {
  94         struct cmobd_async_page *cmap = (struct cmobd_async_page *)data;
  95         obd_valid valid_flags;
  96         struct inode *inode;
  97         ENTRY;
  98
  99         if (IS_ERR(cmap)) {
 100                 EXIT;
 101                 return;
 102         }
 103
 104         inode = cmap->cmap_page->mapping->host;
 105         oa->o_id = cmap->cmap_es->es_oa.o_id;
 106         oa->o_gr = cmap->cmap_es->es_oa.o_gr;
 107         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
 108         valid_flags = OBD_MD_FLTYPE | OBD_MD_FLATIME;
 109         if (cmd == OBD_BRW_WRITE) {
 110                 oa->o_valid |= OBD_MD_FLIFID;
 111
 112                 /* FIXME-UMKA: should be here some mds num and mds id? */
 113                 mdc_pack_id(obdo_id(oa), inode->i_ino, 0,
 114                             inode->i_mode, 0, 0);
 115                 valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME;
 116         }
 117
 118         obdo_from_inode(oa, inode, valid_flags);
 119
 120         EXIT;
 121         return;
 122 }
 123
 124 static void cmobd_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
 125 {
 126         struct cmobd_async_page *cmap = (struct cmobd_async_page *)data;
 127         struct cmobd_extent_set *set = cmap->cmap_es;
 128         unsigned long flags;
 129         struct page *page;
 130         int wakeup = 0;
 131         ENTRY;
 132
 133         page = cmap->cmap_page;
 134         LASSERT(PageLocked(page));
 135
 136         /* XXX */
 137         if (rc)
 138                 SetPageError(page);
 139
 140         spin_lock_irqsave(&set->es_lock, flags);
 141         LASSERT(!list_empty(&set->es_pages));
 142         LASSERT(!list_empty(&cmap->cmap_link));
 143
 144         list_del_init(&cmap->cmap_link);
 145         if (list_empty(&set->es_pages) && !set->es_count)
 146                 wakeup = 1;
 147         spin_unlock_irqrestore(&set->es_lock, flags);
 148
 149         obd_teardown_async_page(set->es_exp, set->es_lsm, NULL,
 150                                 cmap->cmap_cookie);
 151         OBD_FREE(cmap, sizeof(*cmap));
 152
 153         unlock_page(page);
 154         page_cache_release(page);
 155
 156         if (wakeup)
 157                 wake_up(&set->es_waitq);
 158         EXIT;
 159         return;
 160 }
 161
 162 static struct obd_async_page_ops cmobd_async_page_ops = {
 163         .ap_make_ready =        cmobd_ap_make_ready,
 164         .ap_refresh_count =     cmobd_ap_refresh_count,
 165         .ap_fill_obdo =         cmobd_ap_fill_obdo,
 166         .ap_completion =        cmobd_ap_completion,
 167 };
 168
 169 static int cmobd_send_pages(struct obd_device *obd,
 170                             struct niobuf_local *lnb,
 171                             obd_count oa_bufs,
 172                             struct cmobd_extent_set *set)
 173 {
 174         struct cm_obd *cmobd = &obd->u.cm;
 175         struct obd_export *exp = cmobd->master_exp;
 176         struct cmobd_async_page *cmap = NULL;
 177         obd_count i;
 178         int rc = 0;
 179         unsigned long flags;
 180         ENTRY;
 181
 182         for (i = 0; i < oa_bufs; i++, lnb++) {
 183
 184                 OBD_ALLOC(cmap, sizeof(*cmap));
 185                 if (cmap == NULL) {
 186                         CERROR("Not enought memory\n");
 187                         rc = -ENOMEM;
 188                         break;
 189                 }
 190                 INIT_LIST_HEAD(&cmap->cmap_link);
 191                 cmap->cmap_page = lnb->page;
 192                 cmap->cmap_es = set;
 193
 194                 rc = obd_prep_async_page(exp, set->es_lsm, NULL, lnb->page,
 195                                          lnb->offset, &cmobd_async_page_ops,
 196                                          cmap, &cmap->cmap_cookie);
 197                 if (rc) {
 198                         CERROR("cmobd prep async page failed page(%p) rc(%d)\n",
 199                                lnb->page, rc);
 200                         OBD_FREE(cmap, sizeof(*cmap));
 201                         break;
 202                 }
 203
 204                 LASSERT(cmap->cmap_page);
 205                 LASSERT(!PageLocked(cmap->cmap_page));
 206                 LASSERT(Page_Uptodate(cmap->cmap_page));
 207                 page_cache_get(cmap->cmap_page);
 208
 209                 spin_lock_irqsave(&set->es_lock, flags);
 210                 list_add_tail(&cmap->cmap_link, &set->es_pages);
 211                 spin_unlock_irqrestore(&set->es_lock, flags);
 212
 213                 rc = obd_queue_async_io(exp, set->es_lsm, NULL, cmap->cmap_cookie,
 214                                         OBD_BRW_WRITE, 0, 0, 0, 0);
 215                 if (rc) {  /* try sync io */
 216                         struct obd_io_group *oig;
 217
 218                         spin_lock_irqsave(&set->es_lock, flags);
 219                         list_del_init(&cmap->cmap_link);
 220                         spin_unlock_irqrestore(&set->es_lock, flags);
 221
 222                         lock_page(cmap->cmap_page);
 223
 224                         rc = oig_init(&oig);
 225                         if (rc)
 226                                 GOTO(free_page, rc);
 227
 228                         rc = obd_queue_group_io(exp, set->es_lsm, NULL, oig,
 229                                                 cmap->cmap_cookie,
 230                                                 OBD_BRW_WRITE, 0, lnb->len, 0,
 231                                                 ASYNC_READY | ASYNC_URGENT |
 232                                                 ASYNC_COUNT_STABLE |
 233                                                 ASYNC_GROUP_SYNC);
 234
 235                         if (rc)
 236                                 GOTO(free_oig, rc);
 237
 238                         rc = obd_trigger_group_io(exp, set->es_lsm, NULL, oig);
 239                         if (rc)
 240                                 GOTO(free_oig, rc);
 241
 242                         rc = oig_wait(oig);
 243 free_oig:
 244                         oig_release(oig);
 245 free_page:
 246                         unlock_page(cmap->cmap_page);
 247                         page_cache_release(cmap->cmap_page);
 248                         obd_teardown_async_page(exp, set->es_lsm, NULL,
 249                                                 cmap->cmap_cookie);
 250                         OBD_FREE(cmap, sizeof(*cmap));
 251                         if (rc) {
 252                                 CERROR("cmobd sync io failed\n");
 253                                 break;
 254                         }
 255                 }
 256         }
 257         RETURN(rc);
 258 }
 259
 260 static int cmobd_write_extent(struct obd_device *obd,
 261                               struct cmobd_extent_info *ei)
 262 {
 263         struct cmobd_extent_set *set = ei->ei_set;
 264         struct cm_obd *cmobd = &obd->u.cm;
 265         unsigned long flags;
 266         struct obd_ioobj ioo;
 267         struct niobuf_local *lnb;
 268         struct niobuf_remote *rnb;
 269         obd_count i, oa_bufs;
 270         struct obdo *oa;
 271         obd_off offset;
 272         int ret, rc = 0, wakeup = 0;
 273         ENTRY;
 274
 275         oa_bufs = split_extent(&ei->ei_extent, PAGE_SIZE);
 276         LASSERT(oa_bufs > 0);
 277
 278         OBD_ALLOC(lnb, oa_bufs * sizeof(struct niobuf_local));
 279         OBD_ALLOC(rnb, oa_bufs * sizeof(struct niobuf_remote));
 280         oa = obdo_alloc();
 281
 282         if (lnb == NULL || rnb == NULL || oa == NULL)
 283                 GOTO(out, rc = -ENOMEM);
 284
 285         LASSERT(ei->ei_extent.end >= ei->ei_extent.start);
 286         LASSERT((ei->ei_extent.start & (PAGE_SIZE -1)) == 0);
 287
 288         for (i = 0, offset = ei->ei_extent.start; i < oa_bufs;
 289              i++, offset += PAGE_SIZE) {
 290                 rnb[i].offset = offset;
 291                 rnb[i].len = MIN(PAGE_SIZE, ei->ei_extent.end - offset + 1);
 292         }
 293
 294         memcpy(oa, &set->es_oa, sizeof(*oa));
 295         obdo_to_ioobj(oa, &ioo);
 296         ioo.ioo_bufcnt = oa_bufs;
 297
 298         ret = obd_preprw(OBD_BRW_READ, cmobd->cache_exp, oa, 1, &ioo,
 299                          oa_bufs, rnb, lnb, NULL, NULL);
 300         if (ret)
 301                 GOTO(out, rc = ret);
 302
 303         rc = cmobd_send_pages(obd, lnb, oa_bufs, set);
 304         if (rc)
 305                 CERROR("cmobd_send_pages failed %d\n", rc);
 306
 307         rc = obd_commitrw(OBD_BRW_READ, cmobd->cache_exp, oa, 1, &ioo,
 308                           oa_bufs, lnb, NULL, ret);
 309
 310         /* countdown and wake up */
 311         spin_lock_irqsave(&set->es_lock, flags);
 312         LASSERT(set->es_count);
 313         set->es_count--;
 314         if (!set->es_count)
 315                 wakeup = 1;
 316         spin_unlock_irqrestore(&set->es_lock, flags);
 317
 318         if (wakeup)
 319                 wake_up(&set->es_waitq);
 320
 321         EXIT;
 322 out:
 323         if (lnb)
 324                 OBD_FREE(lnb, oa_bufs * sizeof(struct niobuf_local));
 325         if (rnb)
 326                 OBD_FREE(rnb, oa_bufs * sizeof(struct niobuf_remote));
 327         if (oa)
 328                 obdo_free(oa);
 329
 330         return rc;
 331 }
 332
 333 static struct cmobd_extent_info* get_next_ei(struct cmobd_write_service *ws)
 334 {
 335         struct cmobd_extent_info *ei = NULL;
 336         unsigned long flags;
 337         int wakeup = 0;
 338
 339         spin_lock_irqsave(&ws->ws_extent_lock, flags);
 340         if (!list_empty(&ws->ws_extents)) {
 341                 ei = list_entry(ws->ws_extents.next,
 342                                 struct cmobd_extent_info, ei_link);
 343                 list_del_init(&ei->ei_link);
 344                 ws->ws_nextents--;
 345                 if (ws->ws_nextents < CMOBD_MAX_EXTENTS)
 346                         wakeup = 1;
 347         }
 348         spin_unlock_irqrestore(&ws->ws_extent_lock, flags);
 349
 350         if (wakeup)
 351                 wake_up_all(&ws->ws_waitq_provider);
 352
 353         return ei;
 354 }
 355
 356 static int cmobd_write_main(void *arg)
 357 {
 358         struct ptlrpc_svc_data *data = (struct ptlrpc_svc_data *)arg;
 359         struct ptlrpc_thread   *thread = data->thread;
 360         struct obd_device *obd = data->dev;
 361         struct cm_obd *cmobd = &obd->u.cm;
 362         struct cmobd_write_service *ws = cmobd->write_srv;
 363         struct cmobd_extent_info *extent = NULL;
 364         unsigned long flags;
 365         int rc;
 366         ENTRY;
 367
 368         lock_kernel();
 369         ptlrpc_daemonize();
 370
 371         SIGNAL_MASK_LOCK(current, flags);
 372         sigfillset(&current->blocked);
 373         RECALC_SIGPENDING;
 374         SIGNAL_MASK_UNLOCK(current, flags);
 375
 376         LASSERTF(strlen(data->name) < sizeof(current->comm),
 377                  "name %d > len %d\n",strlen(data->name),sizeof(current->comm));
 378         THREAD_NAME(current->comm, sizeof(current->comm) - 1, "%s", data->name);
 379
 380         unlock_kernel();
 381
 382         thread->t_flags = SVC_RUNNING;
 383         wake_up(&thread->t_ctl_waitq);
 384
 385         /* Record that the thread is running */
 386         spin_lock_irqsave(&ws->ws_thread_lock, flags);
 387         ws->ws_nthreads++;
 388         spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 389
 390         while ((thread->t_flags & SVC_STOPPING) == 0) {
 391                 struct l_wait_info lwi = { 0 };
 392
 393                 l_wait_event_exclusive(ws->ws_waitq_consumer,
 394                                        ((thread->t_flags & SVC_STOPPING) ||
 395                                         ((extent = get_next_ei(ws)) !=
 396                                           NULL)),
 397                                        &lwi);
 398                 if (extent == NULL)
 399                         continue;
 400                 rc = cmobd_write_extent(obd, extent);
 401                 if (rc)
 402                         CERROR("write extent failed rc=%d\n", rc);
 403                 OBD_SLAB_FREE(extent, cmobd_extent_slab, sizeof(*extent));
 404                 extent = NULL;
 405         }
 406
 407         thread->t_flags = SVC_STOPPED;
 408         wake_up(&thread->t_ctl_waitq);
 409
 410         spin_lock_irqsave(&ws->ws_thread_lock, flags);
 411         ws->ws_nthreads--;                    /* must know immediately */
 412         spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 413
 414         RETURN(0);
 415 }
 416
 417 /* functions for manipulating cmobd write replay threads, similar with
 418  * ptlrpc threads functions */
 419 static int cmobd_start_thread(struct obd_device *obd, char *name)
 420 {
 421         struct cm_obd *cmobd = &obd->u.cm;
 422         struct cmobd_write_service *ws = cmobd->write_srv;
 423         struct l_wait_info lwi = { 0 };
 424         struct ptlrpc_svc_data d;
 425         struct ptlrpc_thread *thread;
 426         unsigned long flags;
 427         int rc;
 428         ENTRY;
 429
 430         OBD_ALLOC(thread, sizeof(*thread));
 431         if (thread == NULL)
 432                 RETURN(-ENOMEM);
 433         init_waitqueue_head(&thread->t_ctl_waitq);
 434
 435         d.dev = obd;
 436         d.svc = NULL;
 437         d.name = name;
 438         d.thread = thread;
 439
 440         spin_lock_irqsave(&ws->ws_thread_lock, flags);
 441         list_add(&thread->t_link, &ws->ws_threads);
 442         spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 443
 444         /* CLONE_VM and CLONE_FILES just avoid a needless copy, because we
 445          * just drop the VM and FILES in ptlrpc_daemonize() right away.
 446          */
 447         rc = kernel_thread(cmobd_write_main, &d, CLONE_VM | CLONE_FILES);
 448         if (rc < 0) {
 449                 CERROR("cannot start thread: %d\n", rc);
 450                 spin_lock_irqsave(&ws->ws_thread_lock, flags);
 451                 list_del_init(&thread->t_link);
 452                 spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 453                 OBD_FREE(thread, sizeof(*thread));
 454                 RETURN(rc);
 455         }
 456         l_wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_RUNNING, &lwi);
 457
 458         RETURN(0);
 459
 460 }
 461
 462 static void cmobd_stop_thread(struct obd_device *obd,
 463                               struct ptlrpc_thread *thread)
 464 {
 465         struct cm_obd *cmobd = &obd->u.cm;
 466         struct cmobd_write_service *ws = cmobd->write_srv;
 467         struct l_wait_info lwi = { 0 };
 468         unsigned long flags;
 469         ENTRY;
 470
 471         thread->t_flags = SVC_STOPPING;
 472         wake_up_all(&ws->ws_waitq_consumer);
 473
 474         l_wait_event(thread->t_ctl_waitq, (thread->t_flags & SVC_STOPPED),
 475                      &lwi);
 476
 477         spin_lock_irqsave(&ws->ws_thread_lock, flags);
 478         list_del(&thread->t_link);
 479         spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 480
 481         OBD_FREE(thread, sizeof(*thread));
 482         EXIT;
 483 }
 484
 485 static void cmobd_stop_all_threads(struct obd_device *obd)
 486 {
 487         struct cm_obd *cmobd = &obd->u.cm;
 488         struct cmobd_write_service *ws = cmobd->write_srv;
 489         unsigned long flags;
 490         struct ptlrpc_thread *thread;
 491         ENTRY;
 492
 493         spin_lock_irqsave(&ws->ws_thread_lock, flags);
 494         while (!list_empty(&ws->ws_threads)) {
 495                 thread = list_entry(ws->ws_threads.next,
 496                                     struct ptlrpc_thread, t_link);
 497
 498                 spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 499                 cmobd_stop_thread(obd, thread);
 500                 spin_lock_irqsave(&ws->ws_thread_lock, flags);
 501         }
 502
 503         spin_unlock_irqrestore(&ws->ws_thread_lock, flags);
 504         EXIT;
 505 }
 506
 507 static int cmobd_start_n_threads(struct obd_device *obd, int num_threads,
 508                                  char *base_name)
 509 {
 510         int i, rc = 0;
 511         ENTRY;
 512
 513         for (i = 0; i < num_threads; i++) {
 514                 char name[32];
 515                 snprintf(name, sizeof(name) - 1, "%s_%02d", base_name, i);
 516                 rc = cmobd_start_thread(obd, name);
 517                 if (rc) {
 518                         CERROR("cannot start %s thread #%d: rc %d\n", base_name,
 519                                i, rc);
 520                         cmobd_stop_all_threads(obd);
 521                 }
 522         }
 523         RETURN(rc);
 524 }
 525
 526 void cmobd_cleanup_write_srv(struct obd_device *obd)
 527 {
 528         struct cm_obd *cmobd = &obd->u.cm;
 529         struct list_head *pos, *n;
 530         struct cmobd_extent_info *ei;
 531         ENTRY;
 532
 533         cmobd_stop_all_threads(obd);
 534
 535         list_for_each_safe(pos, n, &cmobd->write_srv->ws_extents) {
 536                 ei = list_entry(pos, struct cmobd_extent_info, ei_link);
 537                 list_del_init(&ei->ei_link);
 538                 OBD_FREE(ei, sizeof(*ei));
 539         }
 540         OBD_FREE(cmobd->write_srv, sizeof(*cmobd->write_srv));
 541         EXIT;
 542 }
 543
 544 int cmobd_init_write_srv(struct obd_device *obd)
 545 {
 546         struct cm_obd *cmobd = &obd->u.cm;
 547         struct cmobd_write_service *ws;
 548         int rc;
 549         ENTRY;
 550
 551         OBD_ALLOC(cmobd->write_srv, sizeof(*cmobd->write_srv));
 552         if (cmobd->write_srv == NULL)
 553                 RETURN(-ENOMEM);
 554         ws = cmobd->write_srv;
 555
 556         INIT_LIST_HEAD(&ws->ws_threads);
 557         spin_lock_init(&ws->ws_thread_lock);
 558         ws->ws_nthreads = 0;
 559
 560         INIT_LIST_HEAD(&ws->ws_extents);
 561         spin_lock_init(&ws->ws_extent_lock);
 562         ws->ws_nextents = 0;
 563         init_waitqueue_head(&ws->ws_waitq_provider);
 564         init_waitqueue_head(&ws->ws_waitq_consumer);
 565
 566         rc = cmobd_start_n_threads(obd, CMOBD_NUM_THREADS, "cm_write");
 567         if (rc)
 568                 cmobd_cleanup_write_srv(obd);
 569
 570         RETURN(rc);
 571 }
 572
 573 static int extent_queue_full(struct cmobd_write_service *ws)
 574 {
 575         unsigned long flags;
 576         int full = 0;
 577
 578         spin_lock_irqsave(&ws->ws_extent_lock, flags);
 579         full = (ws->ws_nextents >= CMOBD_MAX_EXTENTS) ? 1 : 0;
 580         spin_unlock_irqrestore(&ws->ws_extent_lock, flags);
 581
 582         return full;
 583 }
 584
 585 static void cmobd_queue_extent(struct obd_device *obd,
 586                                struct cmobd_extent_info *ex)
 587 {
 588         struct cm_obd *cmobd = &obd->u.cm;
 589         struct cmobd_write_service *ws = cmobd->write_srv;
 590         struct cmobd_extent_set *set = ex->ei_set;
 591         unsigned long flags;
 592         struct l_wait_info lwi = { 0 };
 593         ENTRY;
 594
 595 wait:
 596         l_wait_event(ws->ws_waitq_provider, !extent_queue_full(ws), &lwi);
 597
 598         spin_lock_irqsave(&ws->ws_extent_lock, flags);
 599         if (ws->ws_nextents >= CMOBD_MAX_EXTENTS) {
 600                 spin_unlock_irqrestore(&ws->ws_extent_lock, flags);
 601                 goto wait;
 602         }
 603         list_add_tail(&ex->ei_link, &ws->ws_extents);
 604         ws->ws_nextents++;
 605         spin_unlock_irqrestore(&ws->ws_extent_lock, flags);
 606
 607         spin_lock_irqsave(&set->es_lock, flags);
 608         set->es_count++;
 609         spin_unlock_irqrestore(&set->es_lock, flags);
 610
 611         wake_up_all(&ws->ws_waitq_consumer);
 612
 613         EXIT;
 614 }
 615
 616 static obd_size cmobd_id2size(struct obd_export *exp, obd_id id, obd_gr grp)
 617 {
 618         struct lvfs_run_ctxt saved;
 619         struct dentry *de = NULL;
 620         obd_size size;
 621         ENTRY;
 622
 623         push_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
 624
 625         de = obd_lvfs_id2dentry(exp, id, 0, grp);
 626         LASSERT(de);
 627
 628         size = de->d_inode->i_size;
 629
 630         dput(de);
 631         pop_ctxt(&saved, &exp->exp_obd->obd_lvfs_ctxt, NULL);
 632
 633         RETURN(size);
 634 }
 635
 636 static int extent_set_done(struct cmobd_extent_set *set, int phase)
 637 {
 638         int done = 0;
 639         unsigned long flags;
 640
 641         spin_lock_irqsave(&set->es_lock, flags);
 642         if (phase == 1)
 643                 done = set->es_count ? 0 : 1;
 644         else if (phase == 2)
 645                 done = (!set->es_count && list_empty(&set->es_pages)) ? 1 : 0;
 646         spin_unlock_irqrestore(&set->es_lock, flags);
 647
 648         return done;
 649 }
 650
 651 int cmobd_replay_write(struct obd_device *obd, struct obdo *oa,
 652                        struct ldlm_extent *ext)
 653 {
 654         struct cm_obd *cmobd = &obd->u.cm;
 655         struct lov_stripe_md *lsm = NULL;
 656         struct cmobd_extent_set set;
 657         struct cmobd_extent_info *ex;
 658         struct l_wait_info lwi = { 0 };
 659         struct list_head *pos, *n;
 660         struct cmobd_async_page *cmap;
 661         unsigned long flags;
 662         obd_count i, buf_count;
 663         obd_off start;
 664         int rc = 0;
 665         ENTRY;
 666
 667         rc = cmobd_dummy_lsm(&lsm, cmobd->master_desc.ld_tgt_count, oa,
 668                              (__u32)cmobd->master_desc.ld_default_stripe_size);
 669         if (rc)
 670                 RETURN(-ENOMEM);
 671
 672         set.es_extent.start = ext->start;
 673         set.es_extent.end = ext->end;
 674         set.es_lsm = lsm;
 675         set.es_exp = cmobd->master_exp;
 676         set.es_ext_sz = CMOBD_MAX_EXTENT_SZ;
 677         set.es_count = 0;
 678         memcpy(&set.es_oa, oa, sizeof(*oa));
 679
 680         INIT_LIST_HEAD(&set.es_pages);
 681         spin_lock_init(&set.es_lock);
 682         init_waitqueue_head(&set.es_waitq);
 683
 684         if (set.es_extent.end < set.es_extent.start) {
 685                 CDEBUG(D_HA, "illegal extent in write replay\n");
 686                 GOTO(out, rc = -EINVAL);
 687         }
 688         /* start of extent is extended to page boundaries */
 689         set.es_extent.start -= set.es_extent.start & ~PAGE_MASK;
 690         /* if the end of extent is EOF, set it as file size */
 691         if (set.es_extent.end == OBD_OBJECT_EOF) {
 692                 set.es_extent.end = cmobd_id2size(cmobd->cache_exp,
 693                                                   oa->o_id, oa->o_gr) - 1;
 694                 if (set.es_extent.end <= 0)
 695                         GOTO(out, rc = 0);
 696         }
 697
 698         buf_count = split_extent(&set.es_extent, set.es_ext_sz);
 699         for (i = 0, start = set.es_extent.start; i < buf_count;
 700              i++, start += set.es_ext_sz) {
 701                 OBD_SLAB_ALLOC(ex, cmobd_extent_slab, SLAB_NOFS, sizeof(*ex));
 702                 if (ex == NULL) {
 703                         CERROR("not enough memory\n");
 704                         break;
 705                 }
 706
 707                 INIT_LIST_HEAD(&ex->ei_link);
 708                 ex->ei_set = &set;
 709                 ex->ei_extent.start = start;
 710                 ex->ei_extent.end = start + set.es_ext_sz - 1;
 711                 if (ex->ei_extent.end > set.es_extent.end)
 712                         ex->ei_extent.end = set.es_extent.end;
 713
 714                 cmobd_queue_extent(obd, ex);
 715         }
 716
 717         l_wait_event(set.es_waitq, extent_set_done(&set, 1), &lwi);
 718
 719         /* fire remaining ios */
 720         spin_lock_irqsave(&set.es_lock, flags);
 721         list_for_each_safe (pos, n, &set.es_pages) {
 722                 cmap = list_entry(pos, struct cmobd_async_page, cmap_link);
 723
 724                 /* locked pages are in flight */
 725                 if (PageLocked(cmap->cmap_page))
 726                         continue;
 727
 728                 spin_unlock_irqrestore(&set.es_lock, flags);
 729                 rc = obd_set_async_flags(set.es_exp, set.es_lsm, NULL,
 730                                          cmap->cmap_cookie,
 731                                          ASYNC_URGENT);
 732                 if (rc)
 733                         CERROR("cmobd set async flags failed\n");
 734                 spin_lock_irqsave(&set.es_lock, flags);
 735                 break;
 736         }
 737         spin_unlock_irqrestore(&set.es_lock, flags);
 738
 739         l_wait_event(set.es_waitq, extent_set_done(&set, 2), &lwi);
 740 out:
 741         cmobd_free_lsm(&lsm);
 742         RETURN(rc);
 743 }