1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
6 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
8 * This program is free software; you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License version 2 only,
10 * as published by the Free Software Foundation.
12 * This program is distributed in the hope that it will be useful, but
13 * WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * General Public License version 2 for more details (a copy is included
16 * in the LICENSE file that accompanied this code).
18 * You should have received a copy of the GNU General Public License
19 * version 2 along with this program; If not, see
20 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
22 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23 * CA 95054 USA or visit www.sun.com if you need additional information or
29 * Copyright 2008 Sun Microsystems, Inc. All rights reserved
30 * Use is subject to license terms.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lustre/llite/llite_close.c
38 * Lustre Lite routines to issue a secondary close after writeback
41 #include <linux/module.h>
43 #define DEBUG_SUBSYSTEM S_LLITE
45 #include <lustre_lite.h>
46 #include "llite_internal.h"
48 #ifdef HAVE_CLOSE_THREAD
49 /* record that a write is in flight */
50 void llap_write_pending(struct inode *inode, struct ll_async_page *llap)
52 struct ll_inode_info *lli = ll_i2info(inode);
53 spin_lock(&lli->lli_lock);
54 list_add(&llap->llap_pending_write, &lli->lli_pending_write_llaps);
55 spin_unlock(&lli->lli_lock);
58 /* record that a write has completed */
59 void llap_write_complete(struct inode *inode, struct ll_async_page *llap)
61 struct ll_inode_info *lli = ll_i2info(inode);
62 spin_lock(&lli->lli_lock);
63 list_del_init(&llap->llap_pending_write);
64 spin_unlock(&lli->lli_lock);
67 void ll_open_complete(struct inode *inode)
69 struct ll_inode_info *lli = ll_i2info(inode);
70 spin_lock(&lli->lli_lock);
71 lli->lli_send_done_writing = 0;
72 spin_unlock(&lli->lli_lock);
75 /* if we close with writes in flight then we want the completion or cancelation
76 * of those writes to send a DONE_WRITING rpc to the MDS */
77 int ll_is_inode_dirty(struct inode *inode)
79 struct ll_inode_info *lli = ll_i2info(inode);
83 spin_lock(&lli->lli_lock);
84 if (!list_empty(&lli->lli_pending_write_llaps))
86 spin_unlock(&lli->lli_lock);
90 void ll_try_done_writing(struct inode *inode)
92 struct ll_inode_info *lli = ll_i2info(inode);
93 struct ll_close_queue *lcq = ll_i2sbi(inode)->ll_lcq;
95 spin_lock(&lli->lli_lock);
97 if (lli->lli_send_done_writing &&
98 list_empty(&lli->lli_pending_write_llaps)) {
100 spin_lock(&lcq->lcq_lock);
101 if (list_empty(&lli->lli_close_item)) {
102 CDEBUG(D_INODE, "adding inode %lu/%u to close list\n",
103 inode->i_ino, inode->i_generation);
105 list_add_tail(&lli->lli_close_item, &lcq->lcq_list);
106 wake_up(&lcq->lcq_waitq);
108 spin_unlock(&lcq->lcq_lock);
111 spin_unlock(&lli->lli_lock);
114 /* The MDS needs us to get the real file attributes, then send a DONE_WRITING */
115 void ll_queue_done_writing(struct inode *inode)
117 struct ll_inode_info *lli = ll_i2info(inode);
120 spin_lock(&lli->lli_lock);
121 lli->lli_send_done_writing = 1;
122 spin_unlock(&lli->lli_lock);
124 ll_try_done_writing(inode);
128 /* If we know the file size and have the cookies:
129 * - send a DONE_WRITING rpc
132 * - get a whole-file lock
133 * - get the authoritative size and all cookies with GETATTRs
134 * - send a DONE_WRITING rpc
136 static void ll_close_done_writing(struct inode *inode)
138 struct ll_inode_info *lli = ll_i2info(inode);
139 ldlm_policy_data_t policy = { .l_extent = {0, OBD_OBJECT_EOF } };
140 struct lustre_handle lockh = { 0 };
142 struct mdc_op_data data = { { 0 } };
144 int rc, ast_flags = 0;
147 memset(&obdo, 0, sizeof(obdo));
148 if (test_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags))
151 rc = ll_extent_lock(NULL, inode, lli->lli_smd, LCK_PW, &policy, &lockh,
154 CERROR("lock acquisition failed (%d): unable to send "
155 "DONE_WRITING for inode %lu/%u\n", rc, inode->i_ino,
156 inode->i_generation);
160 rc = ll_lsm_getattr(ll_i2obdexp(inode), lli->lli_smd, &obdo);
162 CERROR("inode_getattr failed (%d): unable to send DONE_WRITING "
163 "for inode %lu/%u\n", rc, inode->i_ino,
164 inode->i_generation);
165 ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
169 obdo_refresh_inode(inode, &obdo, valid);
171 CDEBUG(D_INODE, "objid "LPX64" size %Lu, blocks %lu, blksize %lu\n",
172 lli->lli_smd->lsm_object_id, i_size_read(inode), inode->i_blocks,
173 1<<inode->i_blkbits);
175 set_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags);
177 rc = ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
179 CERROR("unlock failed (%d)? proceeding anyways...\n", rc);
182 obdo.o_id = inode->i_ino;
183 obdo.o_size = i_size_read(inode);
184 obdo.o_blocks = inode->i_blocks;
185 obdo.o_valid = OBD_MD_FLID | OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
187 ll_inode2fid(&data.fid1, inode);
188 rc = mdc_done_writing(ll_i2sbi(inode)->ll_mdc_exp, &data, &obdo);
193 static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
195 struct ll_inode_info *lli = NULL;
197 spin_lock(&lcq->lcq_lock);
199 if (lcq->lcq_list.next == NULL)
201 else if (!list_empty(&lcq->lcq_list)) {
202 lli = list_entry(lcq->lcq_list.next, struct ll_inode_info,
204 list_del(&lli->lli_close_item);
207 spin_unlock(&lcq->lcq_lock);
211 static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
213 if (lcq->lcq_list.next == NULL)
220 static int ll_close_thread(void *arg)
222 struct ll_close_queue *lcq = arg;
226 char name[CFS_CURPROC_COMM_MAX];
227 snprintf(name, sizeof(name) - 1, "ll_close");
231 complete(&lcq->lcq_comp);
234 struct l_wait_info lwi = { 0 };
235 struct ll_inode_info *lli;
236 //struct inode *inode;
238 l_wait_event_exclusive(lcq->lcq_waitq,
239 (lli = ll_close_next_lli(lcq)) != NULL,
244 //inode = ll_info2i(lli);
245 //ll_close_done_writing(inode);
249 complete(&lcq->lcq_comp);
253 int ll_close_thread_start(struct ll_close_queue **lcq_ret)
255 struct ll_close_queue *lcq;
258 OBD_FAIL_RETURN(OBD_FAIL_LDLM_CLOSE_THREAD, -EINTR);
259 OBD_ALLOC(lcq, sizeof(*lcq));
263 spin_lock_init(&lcq->lcq_lock);
264 INIT_LIST_HEAD(&lcq->lcq_list);
265 init_waitqueue_head(&lcq->lcq_waitq);
266 init_completion(&lcq->lcq_comp);
268 pid = kernel_thread(ll_close_thread, lcq, 0);
270 OBD_FREE(lcq, sizeof(*lcq));
274 wait_for_completion(&lcq->lcq_comp);
279 void ll_close_thread_shutdown(struct ll_close_queue *lcq)
281 init_completion(&lcq->lcq_comp);
282 lcq->lcq_list.next = NULL;
283 wake_up(&lcq->lcq_waitq);
284 wait_for_completion(&lcq->lcq_comp);
285 OBD_FREE(lcq, sizeof(*lcq));