lustre/llite/llite_close.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Lustre Lite routines to issue a secondary close after writeback
   5  *
   6  *  Copyright (c) 2001-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 #include <linux/module.h>
  25
  26 #define DEBUG_SUBSYSTEM S_LLITE
  27
  28 #include <lustre_mdc.h>
  29 #include <lustre_lite.h>
  30 #include "llite_internal.h"
  31
  32 /* record that a write is in flight */
  33 void llap_write_pending(struct inode *inode, struct ll_async_page *llap)
  34 {
  35         struct ll_inode_info *lli = ll_i2info(inode);
  36
  37         ENTRY;
  38         spin_lock(&lli->lli_lock);
  39         lli->lli_flags |= LLIF_SOM_DIRTY;
  40         if (llap && list_empty(&llap->llap_pending_write))
  41                 list_add(&llap->llap_pending_write,
  42                          &lli->lli_pending_write_llaps);
  43         spin_unlock(&lli->lli_lock);
  44         EXIT;
  45 }
  46
  47 /* record that a write has completed */
  48 int llap_write_complete(struct inode *inode, struct ll_async_page *llap)
  49 {
  50         struct ll_inode_info *lli = ll_i2info(inode);
  51         int rc = 0;
  52
  53         ENTRY;
  54         spin_lock(&lli->lli_lock);
  55         if (llap && !list_empty(&llap->llap_pending_write)) {
  56                 list_del_init(&llap->llap_pending_write);
  57                 rc = 1;
  58         }
  59         spin_unlock(&lli->lli_lock);
  60         RETURN(rc);
  61 }
  62
  63 /* Queue DONE_WRITING if
  64  * - done writing is allowed;
  65  * - inode has no no dirty pages; */
  66 void ll_queue_done_writing(struct inode *inode)
  67 {
  68         struct ll_inode_info *lli = ll_i2info(inode);
  69
  70         spin_lock(&lli->lli_lock);
  71         if ((lli->lli_flags & LLIF_DONE_WRITING) &&
  72             list_empty(&lli->lli_pending_write_llaps)) {
  73                 struct ll_close_queue *lcq = ll_i2sbi(inode)->ll_lcq;
  74
  75                 /* DONE_WRITING is allowed and inode has no dirty page. */
  76                 spin_lock(&lcq->lcq_lock);
  77                 LASSERT(list_empty(&lli->lli_close_list));
  78                 CDEBUG(D_INODE, "adding inode %lu/%u to close list\n",
  79                        inode->i_ino, inode->i_generation);
  80
  81                 list_add_tail(&lli->lli_close_list, &lcq->lcq_head);
  82                 wake_up(&lcq->lcq_waitq);
  83                 spin_unlock(&lcq->lcq_lock);
  84         }
  85         spin_unlock(&lli->lli_lock);
  86 }
  87
  88 /* CLOSE has already occured but has not closed epoch;
  89  * Let let DONE_WRITING to happen. */
  90 void ll_init_done_writing(struct inode *inode) {
  91         struct ll_inode_info *lli = ll_i2info(inode);
  92         spin_lock(&lli->lli_lock);
  93         if ((lli->lli_flags & LLIF_EPOCH_PENDING))
  94                 lli->lli_flags |= LLIF_DONE_WRITING;
  95         spin_unlock(&lli->lli_lock);
  96         ll_queue_done_writing(inode);
  97 }
  98
  99 /* Close epoch and send Size-on-MDS attribute update if possible.
 100  * Call this under @lli->lli_lock spinlock. */
 101 void ll_epoch_close(struct inode *inode, struct md_op_data *op_data)
 102 {
 103         struct ll_inode_info *lli = ll_i2info(inode);
 104         ENTRY;
 105
 106         CDEBUG(D_INODE, "Epoch "LPU64" closed on "DFID"\n",
 107                op_data->ioepoch, PFID(&lli->lli_fid));
 108         op_data->flags |= MF_EPOCH_CLOSE;
 109
 110         /* Pack Size-on-MDS inode attributes only if they has changed */
 111         if (!(lli->lli_flags & LLIF_SOM_DIRTY))
 112                 goto out;
 113
 114         /* There is already 1 pending DONE_WRITE, do not create another one --
 115          * close epoch with no attribute change. */
 116         if (lli->lli_flags & LLIF_EPOCH_PENDING)
 117                 goto out;
 118
 119         op_data->flags |= MF_SOM_CHANGE;
 120
 121         /* Check if Size-on-MDS attributes are valid. */
 122         if ((lli->lli_flags & LLIF_MDS_SIZE_LOCK) || !ll_local_size(inode)) {
 123                 /* Send Size-on-MDS Attributes if valid. */
 124                 op_data->attr.ia_valid |= ATTR_MTIME_SET | ATTR_CTIME_SET |
 125                                           ATTR_SIZE | ATTR_BLOCKS;
 126         }
 127 out:
 128         EXIT;
 129 }
 130
 131 int ll_sizeonmds_update(struct inode *inode, struct lustre_handle *fh)
 132 {
 133         struct ll_inode_info *lli = ll_i2info(inode);
 134         struct md_op_data *op_data;
 135         struct obdo *oa;
 136         int rc;
 137         ENTRY;
 138
 139         LASSERT(!(lli->lli_flags & LLIF_MDS_SIZE_LOCK));
 140
 141         oa = obdo_alloc();
 142         OBD_ALLOC_PTR(op_data);
 143         if (!oa || !op_data) {
 144                 CERROR("can't allocate memory for Size-on-MDS update.\n");
 145                 RETURN(-ENOMEM);
 146         }
 147         rc = ll_inode_getattr(inode, oa);
 148         if (rc) {
 149                 CERROR("inode_getattr failed (%d): unable to send a "
 150                        "Size-on-MDS attribute update for inode %lu/%u\n",
 151                        rc, inode->i_ino, inode->i_generation);
 152                 GOTO(out, rc);
 153         }
 154         CDEBUG(D_INODE, "Size-on-MDS update on "DFID"\n", PFID(&lli->lli_fid));
 155
 156         md_from_obdo(op_data, oa, oa->o_valid);
 157         memcpy(&op_data->handle, fh, sizeof(*fh));
 158
 159         op_data->ioepoch = lli->lli_ioepoch;
 160         op_data->flags |= MF_SOM_CHANGE;
 161
 162         rc = ll_md_setattr(inode, op_data);
 163         EXIT;
 164 out:
 165         if (oa)
 166                 obdo_free(oa);
 167         if (op_data)
 168                 OBD_FREE_PTR(op_data);
 169         return rc;
 170 }
 171
 172 /* Send a DONE_WRITING rpc, pack Size-on-MDS attributes into it, if possible */
 173 static void ll_done_writing(struct inode *inode)
 174 {
 175         struct ll_inode_info *lli = ll_i2info(inode);
 176         struct md_op_data *op_data;
 177         struct obd_client_handle *och;
 178         int rc;
 179         ENTRY;
 180
 181         OBD_ALLOC_PTR(op_data);
 182         if (op_data == NULL) {
 183                 CERROR("can't allocate op_data\n");
 184                 EXIT;
 185                 return;
 186         }
 187
 188         spin_lock(&lli->lli_lock);
 189         LASSERT(lli->lli_flags & LLIF_SOM_DIRTY);
 190
 191         och = lli->lli_pending_och;
 192         lli->lli_pending_och = NULL;
 193         lli->lli_flags &= ~(LLIF_DONE_WRITING | LLIF_EPOCH_PENDING);
 194         ll_epoch_close(inode, op_data);
 195         lli->lli_flags &= ~LLIF_SOM_DIRTY;
 196         spin_unlock(&lli->lli_lock);
 197
 198         ll_pack_inode2opdata(inode, op_data, &och->och_fh);
 199
 200         rc = md_done_writing(ll_i2sbi(inode)->ll_md_exp, op_data, och);
 201         OBD_FREE_PTR(op_data);
 202         if (rc == -EAGAIN) {
 203                 /* MDS has instructed us to obtain Size-on-MDS attribute from
 204                  * OSTs and send setattr to back to MDS. */
 205                 rc = ll_sizeonmds_update(inode, &och->och_fh);
 206         } else if (rc) {
 207                 CERROR("inode %lu mdc done_writing failed: rc = %d\n",
 208                        inode->i_ino, rc);
 209         }
 210         OBD_FREE_PTR(och);
 211         EXIT;
 212 }
 213
 214 static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
 215 {
 216         struct ll_inode_info *lli = NULL;
 217
 218         spin_lock(&lcq->lcq_lock);
 219
 220         if (!list_empty(&lcq->lcq_head)) {
 221                 lli = list_entry(lcq->lcq_head.next, struct ll_inode_info,
 222                                  lli_close_list);
 223                 list_del_init(&lli->lli_close_list);
 224         } else if (atomic_read(&lcq->lcq_stop))
 225                 lli = ERR_PTR(-1);
 226
 227         spin_unlock(&lcq->lcq_lock);
 228         return lli;
 229 }
 230
 231 static int ll_close_thread(void *arg)
 232 {
 233         struct ll_close_queue *lcq = arg;
 234         ENTRY;
 235
 236         {
 237                 char name[CFS_CURPROC_COMM_MAX];
 238                 snprintf(name, sizeof(name) - 1, "ll_close");
 239                 cfs_daemonize(name);
 240         }
 241
 242         complete(&lcq->lcq_comp);
 243
 244         while (1) {
 245                 struct l_wait_info lwi = { 0 };
 246                 struct ll_inode_info *lli;
 247                 struct inode *inode;
 248
 249                 l_wait_event_exclusive(lcq->lcq_waitq,
 250                                        (lli = ll_close_next_lli(lcq)) != NULL,
 251                                        &lwi);
 252                 if (IS_ERR(lli))
 253                         break;
 254
 255                 inode = ll_info2i(lli);
 256                 ll_done_writing(inode);
 257                 iput(inode);
 258         }
 259
 260         complete(&lcq->lcq_comp);
 261         RETURN(0);
 262 }
 263
 264 int ll_close_thread_start(struct ll_close_queue **lcq_ret)
 265 {
 266         struct ll_close_queue *lcq;
 267         pid_t pid;
 268
 269         OBD_ALLOC(lcq, sizeof(*lcq));
 270         if (lcq == NULL)
 271                 return -ENOMEM;
 272
 273         spin_lock_init(&lcq->lcq_lock);
 274         INIT_LIST_HEAD(&lcq->lcq_head);
 275         init_waitqueue_head(&lcq->lcq_waitq);
 276         init_completion(&lcq->lcq_comp);
 277
 278         pid = kernel_thread(ll_close_thread, lcq, 0);
 279         if (pid < 0) {
 280                 OBD_FREE(lcq, sizeof(*lcq));
 281                 return pid;
 282         }
 283
 284         wait_for_completion(&lcq->lcq_comp);
 285         *lcq_ret = lcq;
 286         return 0;
 287 }
 288
 289 void ll_close_thread_shutdown(struct ll_close_queue *lcq)
 290 {
 291         init_completion(&lcq->lcq_comp);
 292         atomic_inc(&lcq->lcq_stop);
 293         wake_up(&lcq->lcq_waitq);
 294         wait_for_completion(&lcq->lcq_comp);
 295         OBD_FREE(lcq, sizeof(*lcq));
 296 }