lustre/liblustre/rw.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Lustre Light Super operations
   5  *
   6  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   7  *
   8  *   This file is part of Lustre, http://www.lustre.org.
   9  *
  10  *   Lustre is free software; you can redistribute it and/or
  11  *   modify it under the terms of version 2 of the GNU General Public
  12  *   License as published by the Free Software Foundation.
  13  *
  14  *   Lustre is distributed in the hope that it will be useful,
  15  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  *   GNU General Public License for more details.
  18  *
  19  *   You should have received a copy of the GNU General Public License
  20  *   along with Lustre; if not, write to the Free Software
  21  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  22  */
  23
  24 #define DEBUG_SUBSYSTEM S_LLITE
  25
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <error.h>
  29 #include <assert.h>
  30 #include <time.h>
  31 #include <sys/types.h>
  32 #include <sys/queue.h>
  33
  34 #include <sysio.h>
  35 #include <fs.h>
  36 #include <mount.h>
  37 #include <inode.h>
  38 #include <file.h>
  39
  40 #include "llite_lib.h"
  41
  42 int llu_iop_iodone(struct ioctx *ioctxp __IS_UNUSED)
  43 {
  44         return 1;
  45 }
  46
  47 /*
  48  * this grabs a lock and manually implements behaviour that makes it look
  49  * like the OST is returning the file size with each lock acquisition
  50  */
  51 int llu_extent_lock(struct ll_file_data *fd, struct inode *inode,
  52                    struct lov_stripe_md *lsm,
  53                    int mode, struct ldlm_extent *extent,
  54                    struct lustre_handle *lockh)
  55 {
  56 #if 0
  57         struct ll_inode_info *lli = ll_i2info(inode);
  58         int rc;
  59         ENTRY;
  60
  61         rc = ll_extent_lock_no_validate(fd, inode, lsm, mode, extent, lockh);
  62         if (rc != ELDLM_OK)
  63                 RETURN(rc);
  64
  65         /* always do a getattr for the first person to pop out of lock
  66          * acquisition.. the DID_GETATTR flag and semaphore serialize
  67          * this initial race.  we used to make a decision based on whether
  68          * the lock was matched or acquired, but the matcher could win the
  69          * waking race with the first issuer so that was no good..
  70          */
  71         if (test_bit(LLI_F_DID_GETATTR, &lli->lli_flags))
  72                 RETURN(ELDLM_OK);
  73
  74         down(&lli->lli_getattr_sem);
  75
  76         if (!test_bit(LLI_F_DID_GETATTR, &lli->lli_flags)) {
  77                 rc = ll_inode_getattr(inode, lsm, fd ? &fd->fd_ost_och : NULL);
  78                 if (rc == 0) {
  79                         set_bit(LLI_F_DID_GETATTR, &lli->lli_flags);
  80                 } else {
  81                         /* XXX can this fail? */
  82                         ll_extent_unlock(fd, inode, lsm, mode, lockh);
  83                 }
  84         }
  85
  86         up(&lli->lli_getattr_sem);
  87         RETURN(rc);
  88 #else
  89         return ELDLM_OK;
  90 #endif
  91 }
  92
  93 int ll_extent_unlock(struct ll_file_data *fd, struct inode *inode,
  94                 struct lov_stripe_md *lsm, int mode,
  95                 struct lustre_handle *lockh)
  96 {
  97 #if 0
  98         struct ll_sb_info *sbi = ll_i2sbi(inode);
  99         int rc;
 100         ENTRY;
 101
 102         /* XXX phil: can we do this?  won't it screw the file size up? */
 103         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 104             (sbi->ll_flags & LL_SBI_NOLCK))
 105                 RETURN(0);
 106
 107         rc = obd_cancel(&sbi->ll_osc_conn, lsm, mode, lockh);
 108
 109         RETURN(rc);
 110 #else
 111         return 0;
 112 #endif
 113 }
 114
 115 static int llu_brw(int cmd, struct inode *inode, struct page *page, int flags)
 116 {
 117         struct llu_inode_info *lli = llu_i2info(inode);
 118         struct lov_stripe_md *lsm = lli->lli_smd;
 119         struct brw_page pg;
 120         int rc;
 121         ENTRY;
 122
 123         pg.pg = page;
 124         pg.off = ((obd_off)page->index) << PAGE_SHIFT;
 125
 126         /* FIXME FIXME FIXME FIXME FIXME FIXME FIXME FIXME FIXME */
 127 #if 0
 128         if (cmd == OBD_BRW_WRITE && (pg.off + PAGE_SIZE > lli->lli_st_size))
 129                 pg.count = lli->lli_st_size % PAGE_SIZE;
 130         else
 131 #endif
 132                 pg.count = PAGE_SIZE;
 133
 134         CDEBUG(D_PAGE, "%s %d bytes ino %lu at "LPU64"/"LPX64"\n",
 135                cmd & OBD_BRW_WRITE ? "write" : "read", pg.count, lli->lli_st_ino,
 136                pg.off, pg.off);
 137         if (pg.count == 0) {
 138                 LBUG();
 139         }
 140
 141         pg.flag = flags;
 142
 143         rc = obd_brw(cmd, llu_i2obdconn(inode), lsm, 1, &pg, set, NULL);
 144         if (rc) {
 145                 CERROR("error from obd_brw: rc = %d\n", rc);
 146         }
 147
 148         RETURN(rc);
 149 }
 150
 151 static int llu_prepare_write(struct inode *inode, struct page *page,
 152                              unsigned from, unsigned to)
 153 {
 154         struct llu_inode_info *lli = llu_i2info(inode);
 155         obd_off offset = ((obd_off)page->index) << PAGE_SHIFT;
 156         int rc = 0;
 157         ENTRY;
 158
 159 #if 0
 160         if (!PageLocked(page))
 161                 LBUG();
 162
 163         if (PageUptodate(page))
 164                 RETURN(0);
 165
 166         //POISON(addr + from, 0xca, to - from);
 167 #endif
 168         /* We're completely overwriting an existing page, so _don't_ set it up
 169          * to date until commit_write */
 170         if (from == 0 && to == PAGE_SIZE)
 171                 RETURN(0);
 172
 173         /* If are writing to a new page, no need to read old data.
 174          * the extent locking and getattr procedures in ll_file_write have
 175          * guaranteed that i_size is stable enough for our zeroing needs */
 176         if (lli->lli_st_size <= offset) {
 177                 memset(kmap(page), 0, PAGE_SIZE);
 178                 kunmap(page);
 179                 GOTO(prepare_done, rc = 0);
 180         }
 181
 182         rc = llu_brw(OBD_BRW_READ, inode, page, 0);
 183
 184         EXIT;
 185
 186  prepare_done:
 187         return rc;
 188 }
 189
 190 static int llu_commit_write(struct inode *inode, struct page *page,
 191                             unsigned from, unsigned to)
 192 {
 193         struct llu_inode_info *lli = llu_i2info(inode);
 194         loff_t size;
 195         int rc;
 196         ENTRY;
 197 #if 0
 198         LASSERT(inode == file->f_dentry->d_inode);
 199         LASSERT(PageLocked(page));
 200
 201         CDEBUG(D_INODE, "inode %p is writing page %p from %d to %d at %lu\n",
 202                inode, page, from, to, page->index);
 203         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu,from=%d,to=%d\n",
 204                inode->i_ino, from, to);
 205         /* to match full page case in prepare_write */
 206         SetPageUptodate(page);
 207         /* mark the page dirty, put it on mapping->dirty,
 208          * mark the inode PAGES_DIRTY, put it on sb->dirty */
 209         set_page_dirty(page);
 210 #endif
 211         rc = llu_brw(OBD_BRW_WRITE, inode, page, 0);
 212         if (rc)
 213                 return rc;
 214
 215         /* this is matched by a hack in obdo_to_inode at the moment */
 216         size = (((obd_off)page->index) << PAGE_SHIFT) + to;
 217         if (size > lli->lli_st_size)
 218                 lli->lli_st_size = size;
 219
 220         RETURN(0);
 221 } /* ll_commit_write */
 222
 223 ssize_t
 224 llu_generic_file_write(struct inode *inode, const char *buf,
 225                        size_t count, loff_t pos)
 226 {
 227         struct page     *page;
 228         ssize_t         written;
 229         long            status = 0;
 230         int             err;
 231         unsigned        bytes;
 232
 233         if ((ssize_t) count < 0)
 234                 return -EINVAL;
 235 #if 0
 236         down(&inode->i_sem);
 237 #endif
 238         if (pos < 0)
 239                 return -EINVAL;
 240
 241         written = 0;
 242
 243 #if 0
 244         remove_suid(inode);
 245         update_inode_times(inode);
 246 #endif
 247         do {
 248                 unsigned long index, offset;
 249                 char *kaddr;
 250
 251                 /*
 252                  * Try to find the page in the cache. If it isn't there,
 253                  * allocate a free page.
 254                  */
 255                 offset = (pos & (PAGE_CACHE_SIZE -1)); /* Within page */
 256                 index = pos >> PAGE_CACHE_SHIFT;
 257                 bytes = PAGE_CACHE_SIZE - offset;
 258                 if (bytes > count) {
 259                         bytes = count;
 260                 }
 261
 262                 status = -ENOMEM;       /* we'll assign it later anyway */
 263                 page = __grab_cache_page(index);
 264                 if (!page)
 265                         break;
 266
 267                 kaddr = kmap(page);
 268                 status = llu_prepare_write(inode, page, offset, offset+bytes);
 269                 if (status)
 270                         goto sync_failure;
 271
 272                 memcpy(kaddr+offset, buf, bytes);
 273
 274                 status = llu_commit_write(inode, page, offset, offset+bytes);
 275                 if (!status)
 276                         status = bytes;
 277
 278                 if (status >= 0) {
 279                         written += status;
 280                         count -= status;
 281                         pos += status;
 282                         buf += status;
 283                 }
 284 unlock:
 285                 kunmap(page);
 286                 page_cache_release(page);
 287
 288                 if (status < 0)
 289                         break;
 290         } while (count);
 291 done:
 292         err = written ? written : status;
 293
 294 #if 0
 295         up(&inode->i_sem);
 296 #endif
 297         return err;
 298
 299         status = -EFAULT;
 300         goto unlock;
 301
 302 sync_failure:
 303         /*
 304          * If blocksize < pagesize, prepare_write() may have instantiated a
 305          * few blocks outside i_size.  Trim these off again.
 306          */
 307         kunmap(page);
 308         page_cache_release(page);
 309         goto done;
 310 }
 311
 312 ssize_t llu_file_write(struct inode *inode, const struct iovec *iovec,
 313                        size_t iovlen, loff_t pos)
 314 {
 315         struct llu_inode_info *lli = llu_i2info(inode);
 316         struct ll_file_data *fd = lli->lli_file_data; /* XXX not ready don't use it now */
 317         struct lustre_handle lockh = { 0 };
 318         struct lov_stripe_md *lsm = lli->lli_smd;
 319         struct ldlm_extent extent;
 320         ldlm_error_t err;
 321         ssize_t retval = 0;
 322         ENTRY;
 323
 324         /* XXX consider other types later */
 325         if (!S_ISREG(lli->lli_st_mode))
 326                 LBUG();
 327 #if 0
 328         CDEBUG(D_VFSTRACE, "VFS Op:inode=%lu,size="LPSZ",offset=%Ld\n",
 329                inode->i_ino, count, *ppos);
 330
 331         /*
 332          * sleep doing some writeback work of this mount's dirty data
 333          * if the VM thinks we're low on memory.. other dirtying code
 334          * paths should think about doing this, too, but they should be
 335          * careful not to hold locked pages while they do so.  like
 336          * ll_prepare_write.  *cough*
 337          */
 338         ll_check_dirty(inode->i_sb);
 339 #endif
 340         while (iovlen--) {
 341                 const char *buf = iovec[iovlen].iov_base;
 342                 size_t count = iovec[iovlen].iov_len;
 343
 344                 /* POSIX, but surprised the VFS doesn't check this already */
 345                 if (count == 0)
 346                         continue;
 347
 348 #if 0
 349                 if (!S_ISBLK(lli->lli_st_mode) && file->f_flags & O_APPEND) {
 350                         extent.start = 0;
 351                         extent.end = OBD_OBJECT_EOF;
 352                 } else  {
 353                         extent.start = *ppos;
 354                         extent.end = *ppos + count - 1;
 355                 }
 356 #else
 357                 extent.start = pos;
 358                 extent.end = pos + count - 1;
 359 #endif
 360
 361                 err = llu_extent_lock(fd, inode, lsm, LCK_PW, &extent, &lockh);
 362                 if (err != ELDLM_OK)
 363                         RETURN(-ENOLCK);
 364
 365 #if 0
 366                 if (!S_ISBLK(inode->i_mode) && file->f_flags & O_APPEND)
 367                         *ppos = inode->i_size;
 368
 369                 CDEBUG(D_INFO, "Writing inode %lu, "LPSZ" bytes, offset %Lu\n",
 370                        inode->i_ino, count, *ppos);
 371 #endif
 372                 retval += llu_generic_file_write(inode, buf, count, pos);
 373         }
 374
 375         /* XXX errors? */
 376         ll_extent_unlock(fd, inode, lsm, LCK_PW, &lockh);
 377         return(retval);
 378 }
 379
 380 static void llu_update_atime(struct inode *inode)
 381 {
 382 #if 0
 383         struct llu_inode_info *lli = llu_i2info(inode);
 384
 385 #ifdef USE_ATIME
 386         struct iattr attr;
 387
 388         attr.ia_atime = LTIME_S(CURRENT_TIME);
 389         attr.ia_valid = ATTR_ATIME;
 390
 391         if (lli->lli_st_atime == attr.ia_atime) return;
 392         if (IS_RDONLY(inode)) return;
 393         if (IS_NOATIME(inode)) return;
 394
 395         /* ll_inode_setattr() sets inode->i_atime from attr.ia_atime */
 396         llu_inode_setattr(inode, &attr, 0);
 397 #else
 398         /* update atime, but don't explicitly write it out just this change */
 399         inode->i_atime = CURRENT_TIME;
 400 #endif
 401 #endif
 402 }
 403
 404 static size_t llu_generic_file_read(struct inode *inode, char *buf,
 405                                     size_t count, loff_t pos)
 406 {
 407         struct llu_inode_info *lli = llu_i2info(inode);
 408         unsigned long index, offset;
 409         int error = 0;
 410         size_t readed = 0;
 411
 412         index = pos >> PAGE_CACHE_SHIFT;
 413         offset = pos & ~PAGE_CACHE_MASK;
 414
 415         do {
 416                 struct page *page;
 417                 unsigned long end_index, nr;
 418
 419                 end_index = lli->lli_st_size >> PAGE_CACHE_SHIFT;
 420
 421                 if (index > end_index)
 422                         break;
 423                 nr = PAGE_CACHE_SIZE;
 424                 if (index == end_index) {
 425                         nr = lli->lli_st_size & ~PAGE_CACHE_MASK;
 426                         if (nr <= offset)
 427                                 break;
 428                 }
 429
 430                 nr = nr - offset;
 431                 if (nr > count)
 432                         nr = count;
 433
 434                 page = grab_cache_page(index);
 435                 if (!page) {
 436                         error = -ENOMEM;
 437                         break;
 438                 }
 439
 440                 error = llu_brw(OBD_BRW_READ, inode, page, 0);
 441                 if (error) {
 442                         page_cache_release(page);
 443                         break;
 444                 }
 445
 446                 memcpy(buf, kmap(page)+offset, nr);
 447                 offset += nr;
 448                 index += offset >> PAGE_CACHE_SHIFT;
 449                 offset &= ~PAGE_CACHE_MASK;
 450                 readed += nr;
 451                 count -= nr;
 452
 453                 page_cache_release(page);
 454         } while (count);
 455
 456         if (error)
 457                 return error;
 458         return readed;
 459 }
 460
 461 ssize_t llu_file_read(struct inode *inode, const struct iovec *iovec,
 462                        size_t iovlen, loff_t pos)
 463 {
 464         struct llu_inode_info *lli = llu_i2info(inode);
 465         struct ll_file_data *fd = lli->lli_file_data;
 466         struct lov_stripe_md *lsm = lli->lli_smd;
 467         struct lustre_handle lockh = { 0 };
 468 #if 0
 469         struct ll_read_extent rextent;
 470 #else
 471         struct ldlm_extent extent;
 472 #endif
 473         ldlm_error_t err;
 474         ssize_t retval = 0;
 475         ENTRY;
 476
 477         while (iovlen--) {
 478                 char *buf = iovec[iovlen].iov_base;
 479                 size_t count = iovec[iovlen].iov_len;
 480
 481                 /* "If nbyte is 0, read() will return 0 and have no other results."
 482                  *                      -- Single Unix Spec */
 483                 if (count == 0)
 484                         RETURN(0);
 485
 486 #if 0
 487                 rextent.re_extent.start = pos;
 488                 rextent.re_extent.end = pos + count - 1;
 489 #else
 490                 extent.start = pos;
 491                 extent.end = pos + count - 1;
 492 #endif
 493                 err = llu_extent_lock(fd, inode, lsm, LCK_PR, &extent, &lockh);
 494                 if (err != ELDLM_OK)
 495                         RETURN(-ENOLCK);
 496 #if 0
 497                 rextent.re_task = current;
 498                 spin_lock(&lli->lli_read_extent_lock);
 499                 list_add(&rextent.re_lli_item, &lli->lli_read_extents);
 500                 spin_unlock(&lli->lli_read_extent_lock);
 501 #endif
 502                 CDEBUG(D_INFO, "Reading inode %lu, "LPSZ" bytes, offset %Ld\n",
 503                        lli->lli_st_ino, count, pos);
 504                 retval = llu_generic_file_read(inode, buf, count, pos);
 505 #if 0
 506                 spin_lock(&lli->lli_read_extent_lock);
 507                 list_del(&rextent.re_lli_item);
 508                 spin_unlock(&lli->lli_read_extent_lock);
 509 #endif
 510         }
 511
 512         if (retval > 0)
 513                 llu_update_atime(inode);
 514
 515         /* XXX errors? */
 516         ll_extent_unlock(fd, inode, lsm, LCK_PR, &lockh);
 517         RETURN(retval);
 518 }
 519