Whamcloud - gitweb
e52d30346c194c682aa944dd37f6def211e9c086
[fs/lustre-release.git] / lustre / target / barrier.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9
10  * This program is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2016, Intel Corporation.
24  *
25  * lustre/target/barrier.c
26  *
27  * Currently, the Lustre barrier is implemented as write barrier on all MDTs.
28  * For each MDT in the system, when it starts, it registers a barrier instance
29  * that will be used in handling subsequent barrier requests.
30  *
31  * Author: Fan, Yong <fan.yong@intel.com>
32  */
33
34 #define DEBUG_SUBSYSTEM S_SNAPSHOT
35
36 #include <linux/percpu_counter.h>
37
38 #include <lustre/lustre_idl.h>
39 #include <dt_object.h>
40 #include <obd.h>
41 #include <obd_class.h>
42 #include <lustre_barrier.h>
43 #include <lustre/lustre_barrier_user.h>
44
45 static LIST_HEAD(barrier_instance_list);
46 static DEFINE_SPINLOCK(barrier_instance_lock);
47
48 struct barrier_instance {
49         struct list_head         bi_link;
50         struct dt_device        *bi_bottom;
51         struct dt_device        *bi_next;
52         wait_queue_head_t        bi_waitq;
53         rwlock_t                 bi_rwlock;
54         struct percpu_counter    bi_writers;
55         atomic_t                 bi_ref;
56         time_t                   bi_deadline;
57         __u32                    bi_status;
58 };
59
60 static inline char *barrier_barrier2name(struct barrier_instance *barrier)
61 {
62         return barrier->bi_bottom->dd_lu_dev.ld_obd->obd_name;
63 }
64
65 static inline __u32 barrier_dev_idx(struct barrier_instance *barrier)
66 {
67         return lu_site2seq(barrier->bi_bottom->dd_lu_dev.ld_site)->ss_node_id;
68 }
69
70 static void barrier_instance_cleanup(struct barrier_instance *barrier)
71 {
72         LASSERT(list_empty(&barrier->bi_link));
73
74         percpu_counter_destroy(&barrier->bi_writers);
75         OBD_FREE_PTR(barrier);
76 }
77
78 static inline void barrier_instance_put(struct barrier_instance *barrier)
79 {
80         if (atomic_dec_and_test(&barrier->bi_ref))
81                 barrier_instance_cleanup(barrier);
82 }
83
84 static struct barrier_instance *
85 barrier_instance_find_locked(struct dt_device *key)
86 {
87         struct barrier_instance *barrier;
88
89         list_for_each_entry(barrier, &barrier_instance_list, bi_link) {
90                 if (barrier->bi_bottom == key)
91                         return barrier;
92         }
93
94         return NULL;
95 }
96
97 static void barrier_instance_add(struct barrier_instance *barrier)
98 {
99         struct barrier_instance *tmp;
100
101         spin_lock(&barrier_instance_lock);
102         tmp = barrier_instance_find_locked(barrier->bi_bottom);
103         LASSERT(!tmp);
104
105         list_add_tail(&barrier->bi_link, &barrier_instance_list);
106         spin_unlock(&barrier_instance_lock);
107 }
108
109 static struct barrier_instance *barrier_instance_find(struct dt_device *key)
110 {
111         struct barrier_instance *barrier;
112
113         spin_lock(&barrier_instance_lock);
114         barrier = barrier_instance_find_locked(key);
115         if (barrier)
116                 atomic_inc(&barrier->bi_ref);
117         spin_unlock(&barrier_instance_lock);
118
119         return barrier;
120 }
121
122 static void barrier_set(struct barrier_instance *barrier, __u32 status)
123 {
124         if (barrier->bi_status != status) {
125                 CDEBUG(D_SNAPSHOT, "%s: change barrier status from %u to %u\n",
126                        barrier_barrier2name(barrier),
127                        barrier->bi_status, status);
128
129                 barrier->bi_status = status;
130         }
131 }
132
133 /**
134  * Create the barrier for the given instance.
135  *
136  * We use two-phases barrier to guarantee that after the barrier setup:
137  * 1) All the MDT side pending async modification have been flushed.
138  * 2) Any subsequent modification will be blocked.
139  * 3) All async transactions on the MDTs have been committed.
140  *
141  * For phase1, we do the following:
142  *
143  * Firstly, it sets barrier flag on the instance that will block subsequent
144  * modifications from clients. (Note: server sponsored modification will be
145  * allowed for flush pending modifications)
146  *
147  * Secondly, it will flush all pending modification via dt_sync(), such as
148  * async OST-object destroy, async OST-object owner changes, and so on.
149  *
150  * If there are some on-handling clients sponsored modifications during the
151  * barrier freezing, then related modifications may cause pending requests
152  * after the first dt_sync(), so call dt_sync() again after all on-handling
153  * modifications done.
154  *
155  * With the phase1 barrier set, all pending cross-servers modification have
156  * been flushed to remote servers, and any new modification will be blocked.
157  * But it does not guarantees that all the updates have been committed to
158  * storage on remote servers. So when all the instances have done phase1
159  * barrier successfully, the MGS will notify all instances to do the phase2
160  * barrier as following:
161  *
162  * Every barrier instance will call dt_sync() to make all async transactions
163  * to be committed locally.
164  *
165  * \param[in] env       pointer to the thread context
166  * \param[in] barrier   pointer to the barrier instance
167  * \param[in] phase1    indicate whether it is phase1 barrier or not
168  *
169  * \retval              positive number for timeout
170  * \retval              0 for success
171  * \retval              negative error number on failure
172  */
173 static int barrier_freeze(const struct lu_env *env,
174                           struct barrier_instance *barrier, bool phase1)
175 {
176         int left;
177         int rc = 0;
178         __s64 inflight = 0;
179         ENTRY;
180
181         write_lock(&barrier->bi_rwlock);
182         barrier_set(barrier, phase1 ? BS_FREEZING_P1 : BS_FREEZING_P2);
183
184         /* Avoid out-of-order execution the barrier_set()
185          * and the check of inflight modifications count. */
186         smp_mb();
187
188         if (phase1)
189                 inflight = percpu_counter_sum(&barrier->bi_writers);
190         write_unlock(&barrier->bi_rwlock);
191
192         rc = dt_sync(env, barrier->bi_next);
193         if (rc)
194                 RETURN(rc);
195
196         LASSERT(barrier->bi_deadline != 0);
197
198         left = barrier->bi_deadline - cfs_time_current_sec();
199         if (left <= 0)
200                 RETURN(1);
201
202         if (phase1 && inflight != 0) {
203                 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(left),
204                                                      NULL, NULL);
205
206                 rc = l_wait_event(barrier->bi_waitq,
207                                   percpu_counter_sum(&barrier->bi_writers) == 0,
208                                   &lwi);
209                 if (rc)
210                         RETURN(1);
211
212                 /* sync again after all inflight modifications done. */
213                 rc = dt_sync(env, barrier->bi_next);
214                 if (rc)
215                         RETURN(rc);
216
217                 if (cfs_time_beforeq(barrier->bi_deadline,
218                                      cfs_time_current_sec()))
219                         RETURN(1);
220         }
221
222         CDEBUG(D_SNAPSHOT, "%s: barrier freezing %s done.\n",
223                barrier_barrier2name(barrier), phase1 ? "phase1" : "phase2");
224
225         if (!phase1)
226                 barrier_set(barrier, BS_FROZEN);
227
228         RETURN(0);
229 }
230
231 void barrier_init(void)
232 {
233 }
234
235 void barrier_fini(void)
236 {
237         LASSERT(list_empty(&barrier_instance_list));
238 }
239
240 bool barrier_entry(struct dt_device *key)
241 {
242         struct barrier_instance *barrier;
243         bool entered = false;
244         ENTRY;
245
246         barrier = barrier_instance_find(key);
247         if (unlikely(!barrier))
248                 /* Fail open */
249                 RETURN(true);
250
251         read_lock(&barrier->bi_rwlock);
252         if (likely(barrier->bi_status != BS_FREEZING_P1 &&
253                    barrier->bi_status != BS_FREEZING_P2 &&
254                    barrier->bi_status != BS_FROZEN) ||
255             cfs_time_beforeq(barrier->bi_deadline, cfs_time_current_sec())) {
256                 percpu_counter_inc(&barrier->bi_writers);
257                 entered = true;
258         }
259         read_unlock(&barrier->bi_rwlock);
260
261         barrier_instance_put(barrier);
262         return entered;
263 }
264 EXPORT_SYMBOL(barrier_entry);
265
266 void barrier_exit(struct dt_device *key)
267 {
268         struct barrier_instance *barrier;
269
270         barrier = barrier_instance_find(key);
271         if (likely(barrier)) {
272                 percpu_counter_dec(&barrier->bi_writers);
273
274                 /* Avoid out-of-order execution the decreasing inflight
275                  * modifications count and the check of barrier status. */
276                 smp_mb();
277
278                 if (unlikely(barrier->bi_status == BS_FREEZING_P1))
279                         wake_up_all(&barrier->bi_waitq);
280                 barrier_instance_put(barrier);
281         }
282 }
283 EXPORT_SYMBOL(barrier_exit);
284
285 int barrier_handler(struct dt_device *key, struct ptlrpc_request *req)
286 {
287         struct ldlm_gl_barrier_desc *desc;
288         struct barrier_instance *barrier;
289         struct barrier_lvb *lvb;
290         struct lu_env env;
291         int rc = 0;
292         ENTRY;
293
294         /* glimpse on barrier locks always packs a glimpse descriptor */
295         req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_DESC_CALLBACK);
296         desc = req_capsule_client_get(&req->rq_pill, &RMF_DLM_GL_DESC);
297         if (!desc)
298                 GOTO(out, rc = -EPROTO);
299
300         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
301                               sizeof(struct barrier_lvb));
302         rc = req_capsule_server_pack(&req->rq_pill);
303         if (rc)
304                 GOTO(out, rc);
305
306         lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB);
307         barrier = barrier_instance_find(key);
308         if (!barrier)
309                 GOTO(out, rc = -ENODEV);
310
311         rc = lu_env_init(&env, LCT_MD_THREAD | LCT_DT_THREAD);
312         if (rc)
313                 GOTO(out_barrier, rc);
314
315         CDEBUG(D_SNAPSHOT,
316                "%s: handling barrier request: status %u, timeout %u\n",
317                barrier_barrier2name(barrier),
318                desc->lgbd_status, desc->lgbd_timeout);
319
320         switch (desc->lgbd_status) {
321         case BS_RESCAN:
322                 barrier_set(barrier, BS_INIT);
323                 break;
324         case BS_FREEZING_P1:
325         case BS_FREEZING_P2:
326                 barrier->bi_deadline = cfs_time_current_sec() +
327                                         desc->lgbd_timeout;
328                 rc = barrier_freeze(&env, barrier,
329                                     desc->lgbd_status == BS_FREEZING_P1);
330                 break;
331         case BS_THAWING:
332         case BS_FAILED:
333         case BS_EXPIRED:
334                 barrier_set(barrier, BS_THAWED);
335                 break;
336         default:
337                 CWARN("%s: unexpected barrier status %u\n",
338                       barrier_barrier2name(barrier), desc->lgbd_status);
339                 rc = -EINVAL;
340                 break;
341         }
342
343         GOTO(fini, rc);
344
345 fini:
346         lu_env_fini(&env);
347
348 out_barrier:
349         if (rc < 0)
350                 barrier_set(barrier, BS_FAILED);
351         else if (rc > 0)
352                 barrier_set(barrier, BS_EXPIRED);
353
354         lvb->lvb_status = barrier->bi_status;
355         lvb->lvb_index = barrier_dev_idx(barrier);
356
357         CDEBUG(D_SNAPSHOT, "%s: handled barrier request: status %u, "
358                "deadline %lu: rc = %d\n", barrier_barrier2name(barrier),
359                lvb->lvb_status, barrier->bi_deadline, rc);
360
361         barrier_instance_put(barrier);
362         rc = 0;
363
364 out:
365         req->rq_status = rc;
366         return rc;
367 }
368 EXPORT_SYMBOL(barrier_handler);
369
370 int barrier_register(struct dt_device *key, struct dt_device *next)
371 {
372         struct barrier_instance *barrier;
373         int rc;
374         ENTRY;
375
376         OBD_ALLOC_PTR(barrier);
377         if (!barrier)
378                 RETURN(-ENOMEM);
379
380         INIT_LIST_HEAD(&barrier->bi_link);
381         barrier->bi_bottom = key;
382         barrier->bi_next = next;
383         init_waitqueue_head(&barrier->bi_waitq);
384         rwlock_init(&barrier->bi_rwlock);
385         atomic_set(&barrier->bi_ref, 1);
386 #ifdef HAVE_PERCPU_COUNTER_INIT_GFP_FLAG
387         rc = percpu_counter_init(&barrier->bi_writers, 0, GFP_KERNEL);
388 #else
389         rc = percpu_counter_init(&barrier->bi_writers, 0);
390 #endif
391         if (rc)
392                 barrier_instance_cleanup(barrier);
393         else
394                 barrier_instance_add(barrier);
395
396         RETURN(rc);
397 }
398 EXPORT_SYMBOL(barrier_register);
399
400 void barrier_deregister(struct dt_device *key)
401 {
402         struct barrier_instance *barrier;
403
404         spin_lock(&barrier_instance_lock);
405         barrier = barrier_instance_find_locked(key);
406         if (barrier)
407                 list_del_init(&barrier->bi_link);
408         spin_unlock(&barrier_instance_lock);
409
410         if (barrier)
411                 barrier_instance_put(barrier);
412 }
413 EXPORT_SYMBOL(barrier_deregister);