Whamcloud - gitweb
f4dfdf2eee678b81892481d1f603fa2882791dc8
[fs/lustre-release.git] / 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 <linux/lustre_mds.h>
29 #include <linux/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         spin_lock(&lli->lli_lock);
37         list_add(&llap->llap_pending_write, &lli->lli_pending_write_llaps);
38         spin_unlock(&lli->lli_lock);
39 }
40
41 /* record that a write has completed */
42 void llap_write_complete(struct inode *inode, struct ll_async_page *llap)
43 {
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);
49 }
50
51 void ll_open_complete(struct inode *inode)
52 {
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);
57 }
58
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)
62 {
63         struct ll_inode_info *lli = ll_i2info(inode);
64         int rc = 0;
65         ENTRY;
66
67         spin_lock(&lli->lli_lock);
68         if (!list_empty(&lli->lli_pending_write_llaps))
69                 rc = 1;
70         spin_unlock(&lli->lli_lock);
71         RETURN(rc);
72 }
73
74 void ll_try_done_writing(struct inode *inode)
75 {
76         struct ll_inode_info *lli = ll_i2info(inode);
77         struct ll_close_queue *lcq = ll_i2sbi(inode)->ll_lcq;
78
79         spin_lock(&lli->lli_lock);
80
81         if (lli->lli_send_done_writing &&
82             list_empty(&lli->lli_pending_write_llaps)) {
83
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);
91                 }
92                 spin_unlock(&lcq->lcq_lock);
93         }
94
95         spin_unlock(&lli->lli_lock);
96 }
97
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)
100 {
101         struct ll_inode_info *lli = ll_i2info(inode);
102         ENTRY;
103
104         spin_lock(&lli->lli_lock);
105         lli->lli_send_done_writing = 1;
106         spin_unlock(&lli->lli_lock);
107
108         ll_try_done_writing(inode);
109         EXIT;
110 }
111
112 #if 0
113 /* If we know the file size and have the cookies:
114  *  - send a DONE_WRITING rpc
115  *
116  * Otherwise:
117  *  - get a whole-file lock
118  *  - get the authoritative size and all cookies with GETATTRs
119  *  - send a DONE_WRITING rpc
120  */
121 static void ll_close_done_writing(struct inode *inode)
122 {
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 };
126         struct obdo obdo;
127         obd_flag valid;
128         int rc, ast_flags = 0;
129         ENTRY;
130
131         memset(&obdo, 0, sizeof(obdo));
132         if (test_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags))
133                 goto rpc;
134
135         rc = ll_extent_lock(NULL, inode, lli->lli_smd, LCK_PW, &policy, &lockh,
136                             ast_flags);
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);
141                 GOTO(out, rc);
142         }
143
144         rc = ll_lsm_getattr(ll_i2obdexp(inode), lli->lli_smd, &obdo);
145         if (rc) {
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);
150                 GOTO(out, rc);
151         }
152
153         obdo_refresh_inode(inode, &obdo, valid);
154
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,
157                inode->i_blksize);
158
159         set_bit(LLI_F_HAVE_OST_SIZE_LOCK, &lli->lli_flags);
160
161         rc = ll_extent_unlock(NULL, inode, lli->lli_smd, LCK_PW, &lockh);
162         if (rc != ELDLM_OK)
163                 CERROR("unlock failed (%d)?  proceeding anyways...\n", rc);
164
165  rpc:
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;
170
171         rc = md_done_writing(ll_i2sbi(inode)->ll_mdc_exp, &obdo);
172  out:
173 }
174 #endif
175
176 static struct ll_inode_info *ll_close_next_lli(struct ll_close_queue *lcq)
177 {
178         struct ll_inode_info *lli = NULL;
179
180         spin_lock(&lcq->lcq_lock);
181
182         if (lcq->lcq_list.next == NULL)
183                 lli = ERR_PTR(-1);
184         else if (!list_empty(&lcq->lcq_list)) {
185                 lli = list_entry(lcq->lcq_list.next, struct ll_inode_info,
186                                  lli_close_item);
187                 list_del(&lli->lli_close_item);
188         }
189
190         spin_unlock(&lcq->lcq_lock);
191         return lli;
192 }
193
194 static int ll_close_thread(void *arg)
195 {
196         struct ll_close_queue *lcq = arg;
197         ENTRY;
198
199         /* XXX boiler-plate */
200         {
201                 char name[sizeof(current->comm)];
202                 unsigned long flags;
203                 snprintf(name, sizeof(name) - 1, "ll_close");
204                 kportal_daemonize(name);
205                 SIGNAL_MASK_LOCK(current, flags);
206                 sigfillset(&current->blocked);
207                 RECALC_SIGPENDING;
208                 SIGNAL_MASK_UNLOCK(current, flags);
209         }
210
211         complete(&lcq->lcq_comp);
212
213         while (1) {
214                 struct l_wait_info lwi = { 0 };
215                 struct ll_inode_info *lli;
216                 //struct inode *inode;
217
218                 l_wait_event_exclusive(lcq->lcq_waitq,
219                                        (lli = ll_close_next_lli(lcq)) != NULL,
220                                        &lwi);
221                 if (IS_ERR(lli))
222                         break;
223
224                 //inode = ll_info2i(lli);
225                 //ll_close_done_writing(inode);
226                 //iput(inode);
227         }
228
229         complete(&lcq->lcq_comp);
230         RETURN(0);
231 }
232
233 int ll_close_thread_start(struct ll_close_queue **lcq_ret)
234 {
235         struct ll_close_queue *lcq;
236         pid_t pid;
237
238         OBD_ALLOC(lcq, sizeof(*lcq));
239         if (lcq == NULL)
240                 return -ENOMEM;
241
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);
246
247         pid = kernel_thread(ll_close_thread, lcq, 0);
248         if (pid < 0) {
249                 OBD_FREE(lcq, sizeof(*lcq));
250                 return pid;
251         }
252
253         wait_for_completion(&lcq->lcq_comp);
254         *lcq_ret = lcq;
255         return 0;
256 }
257
258 void ll_close_thread_shutdown(struct ll_close_queue *lcq)
259 {
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));
265 }