1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Lustre Lite routines to issue a secondary close after writeback
6 * Copyright (c) 2001-2003 Cluster File Systems, Inc.
8 * This file is part of Lustre, http://www.lustre.org.
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.
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.
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.
24 #include <linux/module.h>
26 #define DEBUG_SUBSYSTEM S_LLITE
28 #include <linux/lustre_mds.h>
29 #include <linux/lustre_lite.h>
30 #include "llite_internal.h"
32 /* record that a write is in flight */
33 void llap_write_pending(struct inode *inode, struct ll_async_page *llap)
35 struct ll_inode_info *lli = ll_i2info(inode);
36 spin_lock(&lli->lli_lock);
37 list_add(&llap->llap_pending_write, &lli->lli_pending_write_llaps);
38 spin_unlock(&lli->lli_lock);
41 /* record that a write has completed */
42 void llap_write_complete(struct inode *inode, struct ll_async_page *llap)
44 struct ll_inode_info *lli = ll_i2info(inode);
45 spin_lock(&lli->lli_lock);
46 if (!list_empty(&llap->llap_pending_write))
47 list_del_init(&llap->llap_pending_write);
48 spin_unlock(&lli->lli_lock);
51 void ll_open_complete(struct inode *inode)
53 struct ll_inode_info *lli = ll_i2info(inode);
54 spin_lock(&lli->lli_lock);
55 lli->lli_send_done_writing = 0;
56 spin_unlock(&lli->lli_lock);
59 /* if we close with writes in flight then we want the completion or cancelation
60 * of those writes to send a DONE_WRITING rpc to the MDS */
61 int ll_is_inode_dirty(struct inode *inode)
63 struct ll_inode_info *lli = ll_i2info(inode);
67 spin_lock(&lli->lli_lock);
68 if (!list_empty(&lli->lli_pending_write_llaps))
70 spin_unlock(&lli->lli_lock);
74 void ll_try_done_writing(struct inode *inode)
76 struct ll_inode_info *lli = ll_i2info(inode);
77 struct ll_close_queue *lcq = ll_i2sbi(inode)->ll_lcq;
79 spin_lock(&lli->lli_lock);
81 if (lli->lli_send_done_writing &&
82 list_empty(&lli->lli_pending_write_llaps)) {
84 spin_lock(&lcq->lcq_lock);
85 if (list_empty(&lli->lli_close_item)) {
86 CDEBUG(D_INODE, "adding inode %lu/%u to close list\n",
87 inode->i_ino, inode->i_generation);
88 LASSERT(igrab(inode) == inode);
89 list_add_tail(&lli->lli_close_item, &lcq->lcq_list);
90 wake_up(&lcq->lcq_waitq);
92 spin_unlock(&lcq->lcq_lock);
95 spin_unlock(&lli->lli_lock);
98 /* The MDS needs us to get the real file attributes, then send a DONE_WRITING */
99 void ll_queue_done_writing(struct inode *inode)
101 struct ll_inode_info *lli = ll_i2info(inode);
104 spin_lock(&lli->lli_lock);
105 lli->lli_send_done_writing = 1;
106 spin_unlock(&lli->lli_lock);
108 ll_try_done_writing(inode);
113 /* If we know the file size and have the cookies:
114 * - send a DONE_WRITING rpc
117 * - get a whole-file lock
118 * - get the authoritative size and all cookies with GETATTRs
119 * - send a DONE_WRITING rpc
121 static void ll_close_done_writing(struct inode *inode)
123 struct ll_inode_info *lli = ll_i2info(inode);
124 ldlm_policy_data_t policy = { .l_extent = {0, OBD_OBJECT_EOF } };
125 struct lustre_handle lockh = { 0 };
128 int rc, ast_flags = 0;
131 memset(&obdo, 0, sizeof(obdo));
132 if (test_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags))
135 rc = ll_extent_lock(NULL, inode, lli->lli_smd, LCK_PW, &policy, &lockh,
137 if (rc != ELDLM_OK) {
138 CERROR("lock acquisition failed (%d): unable to send "
139 "DONE_WRITING for inode %lu/%u\n", rc, inode->i_ino,
140 inode->i_generation);
144 rc = ll_lsm_getattr(ll_i2obdexp(inode), lli->lli_smd, &obdo);
146 CERROR("inode_getattr failed (%d): unable to send DONE_WRITING "
147 "for inode %lu/%u\n", rc, inode->i_ino,
148 inode->i_generation);
149 ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
153 obdo_refresh_inode(inode, &obdo, valid);
155 CDEBUG(D_INODE, "objid "LPX64" size %Lu, blocks %lu, blksize %lu\n",
156 lli->lli_smd->lsm_object_id, inode->i_size, inode->i_blocks,
159 set_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags);
161 rc = ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
163 CERROR("unlock failed (%d)? proceeding anyways...\n", rc);
166 obdo.o_id = inode->i_ino;
167 obdo.o_size = inode->i_size;
168 obdo.o_blocks = inode->i_blocks;
169 obdo.o_valid = OBD_MD_FLID | OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
171 rc = md_done_writing(ll_i2sbi(inode)->ll_mdc_exp, &obdo);
176 static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
178 struct ll_inode_info *lli = NULL;
180 spin_lock(&lcq->lcq_lock);
182 if (lcq->lcq_list.next == NULL)
184 else if (!list_empty(&lcq->lcq_list)) {
185 lli = list_entry(lcq->lcq_list.next, struct ll_inode_info,
187 list_del(&lli->lli_close_item);
190 spin_unlock(&lcq->lcq_lock);
194 static int ll_close_thread(void *arg)
196 struct ll_close_queue *lcq = arg;
199 /* XXX boiler-plate */
201 char name[sizeof(current->comm)];
203 snprintf(name, sizeof(name) - 1, "ll_close");
204 kportal_daemonize(name);
205 SIGNAL_MASK_LOCK(current, flags);
206 sigfillset(¤t->blocked);
208 SIGNAL_MASK_UNLOCK(current, flags);
211 complete(&lcq->lcq_comp);
214 struct l_wait_info lwi = { 0 };
215 struct ll_inode_info *lli;
216 //struct inode *inode;
218 l_wait_event_exclusive(lcq->lcq_waitq,
219 (lli = ll_close_next_lli(lcq)) != NULL,
224 //inode = ll_info2i(lli);
225 //ll_close_done_writing(inode);
229 complete(&lcq->lcq_comp);
233 int ll_close_thread_start(struct ll_close_queue **lcq_ret)
235 struct ll_close_queue *lcq;
238 OBD_ALLOC(lcq, sizeof(*lcq));
242 spin_lock_init(&lcq->lcq_lock);
243 INIT_LIST_HEAD(&lcq->lcq_list);
244 init_waitqueue_head(&lcq->lcq_waitq);
245 init_completion(&lcq->lcq_comp);
247 pid = kernel_thread(ll_close_thread, lcq, 0);
249 OBD_FREE(lcq, sizeof(*lcq));
253 wait_for_completion(&lcq->lcq_comp);
258 void ll_close_thread_shutdown(struct ll_close_queue *lcq)
260 init_completion(&lcq->lcq_comp);
261 lcq->lcq_list.next = NULL;
262 wake_up(&lcq->lcq_waitq);
263 wait_for_completion(&lcq->lcq_comp);
264 OBD_FREE(lcq, sizeof(*lcq));