Whamcloud - gitweb
LU-14621 mdd: fix lock-tx order in mdd_xattr_merge()
[fs/lustre-release.git] / lustre / llite / pcc.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, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2017, DDN Storage Corporation.
24  */
25 /*
26  * Persistent Client Cache
27  *
28  * PCC is a new framework which provides a group of local cache on Lustre
29  * client side. It works in two modes: RW-PCC enables a read-write cache on the
30  * local SSDs of a single client; RO-PCC provides a read-only cache on the
31  * local SSDs of multiple clients. Less overhead is visible to the applications
32  * and network latencies and lock conflicts can be significantly reduced.
33  *
34  * For RW-PCC, no global namespace will be provided. Each client uses its own
35  * local storage as a cache for itself. Local file system is used to manage
36  * the data on local caches. Cached I/O is directed to local file system while
37  * normal I/O is directed to OSTs. RW-PCC uses HSM for data synchronization.
38  * It uses HSM copytool to restore file from local caches to Lustre OSTs. Each
39  * PCC has a copytool instance running with unique archive number. Any remote
40  * access from another Lustre client would trigger the data synchronization. If
41  * a client with RW-PCC goes offline, the cached data becomes inaccessible for
42  * other client temporarily. And after the RW-PCC client reboots and the
43  * copytool restarts, the data will be accessible again.
44  *
45  * Following is what will happen in different conditions for RW-PCC:
46  *
47  * > When file is being created on RW-PCC
48  *
49  * A normal HSM released file is created on MDT;
50  * An empty mirror file is created on local cache;
51  * The HSM status of the Lustre file will be set to archived and released;
52  * The archive number will be set to the proper value.
53  *
54  * > When file is being prefetched to RW-PCC
55  *
56  * An file is copied to the local cache;
57  * The HSM status of the Lustre file will be set to archived and released;
58  * The archive number will be set to the proper value.
59  *
60  * > When file is being accessed from PCC
61  *
62  * Data will be read directly from local cache;
63  * Metadata will be read from MDT, except file size;
64  * File size will be got from local cache.
65  *
66  * > When PCC cached file is being accessed on another client
67  *
68  * RW-PCC cached files are automatically restored when a process on another
69  * client tries to read or modify them. The corresponding I/O will block
70  * waiting for the released file to be restored. This is transparent to the
71  * process.
72  *
73  * For RW-PCC, when a file is being created, a rule-based policy is used to
74  * determine whether it will be cached. Rule-based caching of newly created
75  * files can determine which file can use a cache on PCC directly without any
76  * admission control.
77  *
78  * RW-PCC design can accelerate I/O intensive applications with one-to-one
79  * mappings between files and accessing clients. However, in several use cases,
80  * files will never be updated, but need to be read simultaneously from many
81  * clients. RO-PCC implements a read-only caching on Lustre clients using
82  * SSDs. RO-PCC is based on the same framework as RW-PCC, expect
83  * that no HSM mechanism is used.
84  *
85  * The main advantages to use this SSD cache on the Lustre clients via PCC
86  * is that:
87  * - The I/O stack becomes much simpler for the cached data, as there is no
88  *   interference with I/Os from other clients, which enables easier
89  *   performance optimizations;
90  * - The requirements on the HW inside the client nodes are small, any kind of
91  *   SSDs or even HDDs can be used as cache devices;
92  * - Caching reduces the pressure on the object storage targets (OSTs), as
93  *   small or random I/Os can be regularized to big sequential I/Os and
94  *   temporary files do not even need to be flushed to OSTs.
95  *
96  * PCC can accelerate applications with certain I/O patterns:
97  * - small-sized random writes (< 1MB) from a single client
98  * - repeated read of data that is larger than RAM
99  * - clients with high network latency
100  *
101  * Author: Li Xi <lixi@ddn.com>
102  * Author: Qian Yingjin <qian@ddn.com>
103  */
104
105 #define DEBUG_SUBSYSTEM S_LLITE
106
107 #include "pcc.h"
108 #include <linux/namei.h>
109 #include <linux/file.h>
110 #include <lustre_compat.h>
111 #include "llite_internal.h"
112
113 struct kmem_cache *pcc_inode_slab;
114
115 int pcc_super_init(struct pcc_super *super)
116 {
117         struct cred *cred;
118
119         super->pccs_cred = cred = prepare_creds();
120         if (!cred)
121                 return -ENOMEM;
122
123         /* Never override disk quota limits or use reserved space */
124         cap_lower(cred->cap_effective, CAP_SYS_RESOURCE);
125         init_rwsem(&super->pccs_rw_sem);
126         INIT_LIST_HEAD(&super->pccs_datasets);
127         super->pccs_generation = 1;
128
129         return 0;
130 }
131
132 /* Rule based auto caching */
133 static void pcc_id_list_free(struct list_head *id_list)
134 {
135         struct pcc_match_id *id, *n;
136
137         list_for_each_entry_safe(id, n, id_list, pmi_linkage) {
138                 list_del_init(&id->pmi_linkage);
139                 OBD_FREE_PTR(id);
140         }
141 }
142
143 static void pcc_fname_list_free(struct list_head *fname_list)
144 {
145         struct pcc_match_fname *fname, *n;
146
147         list_for_each_entry_safe(fname, n, fname_list, pmf_linkage) {
148                 OBD_FREE(fname->pmf_name, strlen(fname->pmf_name) + 1);
149                 list_del_init(&fname->pmf_linkage);
150                 OBD_FREE_PTR(fname);
151         }
152 }
153
154 static void pcc_expression_free(struct pcc_expression *expr)
155 {
156         LASSERT(expr->pe_field >= PCC_FIELD_UID &&
157                 expr->pe_field < PCC_FIELD_MAX);
158         switch (expr->pe_field) {
159         case PCC_FIELD_UID:
160         case PCC_FIELD_GID:
161         case PCC_FIELD_PROJID:
162                 pcc_id_list_free(&expr->pe_cond);
163                 break;
164         case PCC_FIELD_FNAME:
165                 pcc_fname_list_free(&expr->pe_cond);
166                 break;
167         default:
168                 LBUG();
169         }
170         OBD_FREE_PTR(expr);
171 }
172
173 static void pcc_conjunction_free(struct pcc_conjunction *conjunction)
174 {
175         struct pcc_expression *expression, *n;
176
177         LASSERT(list_empty(&conjunction->pc_linkage));
178         list_for_each_entry_safe(expression, n,
179                                  &conjunction->pc_expressions,
180                                  pe_linkage) {
181                 list_del_init(&expression->pe_linkage);
182                 pcc_expression_free(expression);
183         }
184         OBD_FREE_PTR(conjunction);
185 }
186
187 static void pcc_rule_conds_free(struct list_head *cond_list)
188 {
189         struct pcc_conjunction *conjunction, *n;
190
191         list_for_each_entry_safe(conjunction, n, cond_list, pc_linkage) {
192                 list_del_init(&conjunction->pc_linkage);
193                 pcc_conjunction_free(conjunction);
194         }
195 }
196
197 static void pcc_cmd_fini(struct pcc_cmd *cmd)
198 {
199         if (cmd->pccc_cmd == PCC_ADD_DATASET) {
200                 if (!list_empty(&cmd->u.pccc_add.pccc_conds))
201                         pcc_rule_conds_free(&cmd->u.pccc_add.pccc_conds);
202                 if (cmd->u.pccc_add.pccc_conds_str)
203                         OBD_FREE(cmd->u.pccc_add.pccc_conds_str,
204                                  strlen(cmd->u.pccc_add.pccc_conds_str) + 1);
205         }
206 }
207
208 #define PCC_DISJUNCTION_DELIM   (',')
209 #define PCC_CONJUNCTION_DELIM   ('&')
210 #define PCC_EXPRESSION_DELIM    ('=')
211
212 static int
213 pcc_fname_list_add(struct cfs_lstr *id, struct list_head *fname_list)
214 {
215         struct pcc_match_fname *fname;
216
217         OBD_ALLOC_PTR(fname);
218         if (fname == NULL)
219                 return -ENOMEM;
220
221         OBD_ALLOC(fname->pmf_name, id->ls_len + 1);
222         if (fname->pmf_name == NULL) {
223                 OBD_FREE_PTR(fname);
224                 return -ENOMEM;
225         }
226
227         memcpy(fname->pmf_name, id->ls_str, id->ls_len);
228         list_add_tail(&fname->pmf_linkage, fname_list);
229         return 0;
230 }
231
232 static int
233 pcc_fname_list_parse(char *str, int len, struct list_head *fname_list)
234 {
235         struct cfs_lstr src;
236         struct cfs_lstr res;
237         int rc = 0;
238
239         ENTRY;
240
241         src.ls_str = str;
242         src.ls_len = len;
243         INIT_LIST_HEAD(fname_list);
244         while (src.ls_str) {
245                 rc = cfs_gettok(&src, ' ', &res);
246                 if (rc == 0) {
247                         rc = -EINVAL;
248                         break;
249                 }
250                 rc = pcc_fname_list_add(&res, fname_list);
251                 if (rc)
252                         break;
253         }
254         if (rc)
255                 pcc_fname_list_free(fname_list);
256         RETURN(rc);
257 }
258
259 static int
260 pcc_id_list_parse(char *str, int len, struct list_head *id_list,
261                   enum pcc_field type)
262 {
263         struct cfs_lstr src;
264         struct cfs_lstr res;
265         int rc = 0;
266
267         ENTRY;
268
269         if (type != PCC_FIELD_UID && type != PCC_FIELD_GID &&
270             type != PCC_FIELD_PROJID)
271                 RETURN(-EINVAL);
272
273         src.ls_str = str;
274         src.ls_len = len;
275         INIT_LIST_HEAD(id_list);
276         while (src.ls_str) {
277                 struct pcc_match_id *id;
278                 __u32 id_val;
279
280                 if (cfs_gettok(&src, ' ', &res) == 0)
281                         GOTO(out, rc = -EINVAL);
282
283                 if (!cfs_str2num_check(res.ls_str, res.ls_len,
284                                        &id_val, 0, (u32)~0U))
285                         GOTO(out, rc = -EINVAL);
286
287                 OBD_ALLOC_PTR(id);
288                 if (id == NULL)
289                         GOTO(out, rc = -ENOMEM);
290
291                 id->pmi_id = id_val;
292                 list_add_tail(&id->pmi_linkage, id_list);
293         }
294 out:
295         if (rc)
296                 pcc_id_list_free(id_list);
297         RETURN(rc);
298 }
299
300 static inline bool
301 pcc_check_field(struct cfs_lstr *field, char *str)
302 {
303         int len = strlen(str);
304
305         return (field->ls_len == len &&
306                 strncmp(field->ls_str, str, len) == 0);
307 }
308
309 static int
310 pcc_expression_parse(struct cfs_lstr *src, struct list_head *cond_list)
311 {
312         struct pcc_expression *expr;
313         struct cfs_lstr field;
314         int rc = 0;
315
316         OBD_ALLOC_PTR(expr);
317         if (expr == NULL)
318                 return -ENOMEM;
319
320         rc = cfs_gettok(src, PCC_EXPRESSION_DELIM, &field);
321         if (rc == 0 || src->ls_len <= 2 || src->ls_str[0] != '{' ||
322             src->ls_str[src->ls_len - 1] != '}')
323                 GOTO(out, rc = -EINVAL);
324
325         /* Skip '{' and '}' */
326         src->ls_str++;
327         src->ls_len -= 2;
328
329         if (pcc_check_field(&field, "uid")) {
330                 if (pcc_id_list_parse(src->ls_str,
331                                       src->ls_len,
332                                       &expr->pe_cond,
333                                       PCC_FIELD_UID) < 0)
334                         GOTO(out, rc = -EINVAL);
335                 expr->pe_field = PCC_FIELD_UID;
336         } else if (pcc_check_field(&field, "gid")) {
337                 if (pcc_id_list_parse(src->ls_str,
338                                       src->ls_len,
339                                       &expr->pe_cond,
340                                       PCC_FIELD_GID) < 0)
341                         GOTO(out, rc = -EINVAL);
342                 expr->pe_field = PCC_FIELD_GID;
343         } else if (pcc_check_field(&field, "projid")) {
344                 if (pcc_id_list_parse(src->ls_str,
345                                       src->ls_len,
346                                       &expr->pe_cond,
347                                       PCC_FIELD_PROJID) < 0)
348                         GOTO(out, rc = -EINVAL);
349                 expr->pe_field = PCC_FIELD_PROJID;
350         } else if (pcc_check_field(&field, "fname")) {
351                 if (pcc_fname_list_parse(src->ls_str,
352                                          src->ls_len,
353                                          &expr->pe_cond) < 0)
354                         GOTO(out, rc = -EINVAL);
355                 expr->pe_field = PCC_FIELD_FNAME;
356         } else {
357                 GOTO(out, rc = -EINVAL);
358         }
359
360         list_add_tail(&expr->pe_linkage, cond_list);
361         return 0;
362 out:
363         OBD_FREE_PTR(expr);
364         return rc;
365 }
366
367 static int
368 pcc_conjunction_parse(struct cfs_lstr *src, struct list_head *cond_list)
369 {
370         struct pcc_conjunction *conjunction;
371         struct cfs_lstr expr;
372         int rc = 0;
373
374         OBD_ALLOC_PTR(conjunction);
375         if (conjunction == NULL)
376                 return -ENOMEM;
377
378         INIT_LIST_HEAD(&conjunction->pc_expressions);
379         list_add_tail(&conjunction->pc_linkage, cond_list);
380
381         while (src->ls_str) {
382                 rc = cfs_gettok(src, PCC_CONJUNCTION_DELIM, &expr);
383                 if (rc == 0) {
384                         rc = -EINVAL;
385                         break;
386                 }
387                 rc = pcc_expression_parse(&expr,
388                                           &conjunction->pc_expressions);
389                 if (rc)
390                         break;
391         }
392         return rc;
393 }
394
395 static int pcc_conds_parse(char *str, int len, struct list_head *cond_list)
396 {
397         struct cfs_lstr src;
398         struct cfs_lstr res;
399         int rc = 0;
400
401         src.ls_str = str;
402         src.ls_len = len;
403         INIT_LIST_HEAD(cond_list);
404         while (src.ls_str) {
405                 rc = cfs_gettok(&src, PCC_DISJUNCTION_DELIM, &res);
406                 if (rc == 0) {
407                         rc = -EINVAL;
408                         break;
409                 }
410                 rc = pcc_conjunction_parse(&res, cond_list);
411                 if (rc)
412                         break;
413         }
414         return rc;
415 }
416
417 static int pcc_id_parse(struct pcc_cmd *cmd, const char *id)
418 {
419         int rc;
420
421         OBD_ALLOC(cmd->u.pccc_add.pccc_conds_str, strlen(id) + 1);
422         if (cmd->u.pccc_add.pccc_conds_str == NULL)
423                 return -ENOMEM;
424
425         memcpy(cmd->u.pccc_add.pccc_conds_str, id, strlen(id));
426
427         rc = pcc_conds_parse(cmd->u.pccc_add.pccc_conds_str,
428                              strlen(cmd->u.pccc_add.pccc_conds_str),
429                              &cmd->u.pccc_add.pccc_conds);
430         if (rc)
431                 pcc_cmd_fini(cmd);
432
433         return rc;
434 }
435
436 static int
437 pcc_parse_value_pair(struct pcc_cmd *cmd, char *buffer)
438 {
439         char *key, *val;
440         unsigned long id;
441         int rc;
442
443         val = buffer;
444         key = strsep(&val, "=");
445         if (val == NULL || strlen(val) == 0)
446                 return -EINVAL;
447
448         /* Key of the value pair */
449         if (strcmp(key, "rwid") == 0) {
450                 rc = kstrtoul(val, 10, &id);
451                 if (rc)
452                         return rc;
453                 if (id <= 0)
454                         return -EINVAL;
455                 cmd->u.pccc_add.pccc_rwid = id;
456         } else if (strcmp(key, "roid") == 0) {
457                 rc = kstrtoul(val, 10, &id);
458                 if (rc)
459                         return rc;
460                 if (id <= 0)
461                         return -EINVAL;
462                 cmd->u.pccc_add.pccc_roid = id;
463         } else if (strcmp(key, "auto_attach") == 0) {
464                 rc = kstrtoul(val, 10, &id);
465                 if (rc)
466                         return rc;
467                 if (id == 0)
468                         cmd->u.pccc_add.pccc_flags &= ~PCC_DATASET_AUTO_ATTACH;
469         } else if (strcmp(key, "open_attach") == 0) {
470                 rc = kstrtoul(val, 10, &id);
471                 if (rc)
472                         return rc;
473                 if (id == 0)
474                         cmd->u.pccc_add.pccc_flags &= ~PCC_DATASET_OPEN_ATTACH;
475         } else if (strcmp(key, "io_attach") == 0) {
476                 rc = kstrtoul(val, 10, &id);
477                 if (rc)
478                         return rc;
479                 if (id == 0)
480                         cmd->u.pccc_add.pccc_flags &= ~PCC_DATASET_IO_ATTACH;
481         } else if (strcmp(key, "stat_attach") == 0) {
482                 rc = kstrtoul(val, 10, &id);
483                 if (rc)
484                         return rc;
485                 if (id == 0)
486                         cmd->u.pccc_add.pccc_flags &= ~PCC_DATASET_STAT_ATTACH;
487         } else if (strcmp(key, "rwpcc") == 0) {
488                 rc = kstrtoul(val, 10, &id);
489                 if (rc)
490                         return rc;
491                 if (id > 0)
492                         cmd->u.pccc_add.pccc_flags |= PCC_DATASET_RWPCC;
493         } else if (strcmp(key, "ropcc") == 0) {
494                 rc = kstrtoul(val, 10, &id);
495                 if (rc)
496                         return rc;
497                 if (id > 0)
498                         cmd->u.pccc_add.pccc_flags |= PCC_DATASET_ROPCC;
499         } else {
500                 return -EINVAL;
501         }
502
503         return 0;
504 }
505
506 static int
507 pcc_parse_value_pairs(struct pcc_cmd *cmd, char *buffer)
508 {
509         char *val;
510         char *token;
511         int rc;
512
513         switch (cmd->pccc_cmd) {
514         case PCC_ADD_DATASET:
515                 /* Enable auto attach by default */
516                 cmd->u.pccc_add.pccc_flags |= PCC_DATASET_AUTO_ATTACH;
517                 break;
518         case PCC_DEL_DATASET:
519         case PCC_CLEAR_ALL:
520                 break;
521         default:
522                 return -EINVAL;
523         }
524
525         val = buffer;
526         while (val != NULL && strlen(val) != 0) {
527                 token = strsep(&val, " ");
528                 rc = pcc_parse_value_pair(cmd, token);
529                 if (rc)
530                         return rc;
531         }
532
533         switch (cmd->pccc_cmd) {
534         case PCC_ADD_DATASET:
535                 if (cmd->u.pccc_add.pccc_flags & PCC_DATASET_RWPCC &&
536                     cmd->u.pccc_add.pccc_flags & PCC_DATASET_ROPCC)
537                         return -EINVAL;
538                 /*
539                  * By default, a PCC backend can provide caching service for
540                  * both RW-PCC and RO-PCC.
541                  */
542                 if ((cmd->u.pccc_add.pccc_flags & PCC_DATASET_PCC_ALL) == 0)
543                         cmd->u.pccc_add.pccc_flags |= PCC_DATASET_PCC_ALL;
544
545                 /* For RW-PCC, the value of @rwid must be non zero. */
546                 if (cmd->u.pccc_add.pccc_flags & PCC_DATASET_RWPCC &&
547                     cmd->u.pccc_add.pccc_rwid == 0)
548                         return -EINVAL;
549
550                 break;
551         case PCC_DEL_DATASET:
552         case PCC_CLEAR_ALL:
553                 break;
554         default:
555                 return -EINVAL;
556         }
557         return 0;
558 }
559
560 static void
561 pcc_dataset_rule_fini(struct pcc_match_rule *rule)
562 {
563         if (!list_empty(&rule->pmr_conds))
564                 pcc_rule_conds_free(&rule->pmr_conds);
565         LASSERT(rule->pmr_conds_str != NULL);
566         OBD_FREE(rule->pmr_conds_str, strlen(rule->pmr_conds_str) + 1);
567 }
568
569 static int
570 pcc_dataset_rule_init(struct pcc_match_rule *rule, struct pcc_cmd *cmd)
571 {
572         int rc = 0;
573
574         LASSERT(cmd->u.pccc_add.pccc_conds_str);
575         OBD_ALLOC(rule->pmr_conds_str,
576                   strlen(cmd->u.pccc_add.pccc_conds_str) + 1);
577         if (rule->pmr_conds_str == NULL)
578                 return -ENOMEM;
579
580         memcpy(rule->pmr_conds_str,
581                cmd->u.pccc_add.pccc_conds_str,
582                strlen(cmd->u.pccc_add.pccc_conds_str));
583
584         INIT_LIST_HEAD(&rule->pmr_conds);
585         if (!list_empty(&cmd->u.pccc_add.pccc_conds))
586                 rc = pcc_conds_parse(rule->pmr_conds_str,
587                                           strlen(rule->pmr_conds_str),
588                                           &rule->pmr_conds);
589
590         if (rc)
591                 pcc_dataset_rule_fini(rule);
592
593         return rc;
594 }
595
596 /* Rule Matching */
597 static int
598 pcc_id_list_match(struct list_head *id_list, __u32 id_val)
599 {
600         struct pcc_match_id *id;
601
602         list_for_each_entry(id, id_list, pmi_linkage) {
603                 if (id->pmi_id == id_val)
604                         return 1;
605         }
606         return 0;
607 }
608
609 static bool
610 cfs_match_wildcard(const char *pattern, const char *content)
611 {
612         if (*pattern == '\0' && *content == '\0')
613                 return true;
614
615         if (*pattern == '*' && *(pattern + 1) != '\0' && *content == '\0')
616                 return false;
617
618         while (*pattern == *content) {
619                 pattern++;
620                 content++;
621                 if (*pattern == '\0' && *content == '\0')
622                         return true;
623
624                 if (*pattern == '*' && *(pattern + 1) != '\0' &&
625                     *content == '\0')
626                         return false;
627         }
628
629         if (*pattern == '*')
630                 return (cfs_match_wildcard(pattern + 1, content) ||
631                         cfs_match_wildcard(pattern, content + 1));
632
633         return false;
634 }
635
636 static int
637 pcc_fname_list_match(struct list_head *fname_list, const char *name)
638 {
639         struct pcc_match_fname *fname;
640
641         list_for_each_entry(fname, fname_list, pmf_linkage) {
642                 if (cfs_match_wildcard(fname->pmf_name, name))
643                         return 1;
644         }
645         return 0;
646 }
647
648 static int
649 pcc_expression_match(struct pcc_expression *expr, struct pcc_matcher *matcher)
650 {
651         switch (expr->pe_field) {
652         case PCC_FIELD_UID:
653                 return pcc_id_list_match(&expr->pe_cond, matcher->pm_uid);
654         case PCC_FIELD_GID:
655                 return pcc_id_list_match(&expr->pe_cond, matcher->pm_gid);
656         case PCC_FIELD_PROJID:
657                 return pcc_id_list_match(&expr->pe_cond, matcher->pm_projid);
658         case PCC_FIELD_FNAME:
659                 return pcc_fname_list_match(&expr->pe_cond,
660                                             matcher->pm_name->name);
661         default:
662                 return 0;
663         }
664 }
665
666 static int
667 pcc_conjunction_match(struct pcc_conjunction *conjunction,
668                       struct pcc_matcher *matcher)
669 {
670         struct pcc_expression *expr;
671         int matched;
672
673         list_for_each_entry(expr, &conjunction->pc_expressions, pe_linkage) {
674                 matched = pcc_expression_match(expr, matcher);
675                 if (!matched)
676                         return 0;
677         }
678
679         return 1;
680 }
681
682 static int
683 pcc_cond_match(struct pcc_match_rule *rule, struct pcc_matcher *matcher)
684 {
685         struct pcc_conjunction *conjunction;
686         int matched;
687
688         list_for_each_entry(conjunction, &rule->pmr_conds, pc_linkage) {
689                 matched = pcc_conjunction_match(conjunction, matcher);
690                 if (matched)
691                         return 1;
692         }
693
694         return 0;
695 }
696
697 struct pcc_dataset*
698 pcc_dataset_match_get(struct pcc_super *super, struct pcc_matcher *matcher)
699 {
700         struct pcc_dataset *dataset;
701         struct pcc_dataset *selected = NULL;
702
703         down_read(&super->pccs_rw_sem);
704         list_for_each_entry(dataset, &super->pccs_datasets, pccd_linkage) {
705                 if (!(dataset->pccd_flags & PCC_DATASET_RWPCC))
706                         continue;
707
708                 if (pcc_cond_match(&dataset->pccd_rule, matcher)) {
709                         atomic_inc(&dataset->pccd_refcount);
710                         selected = dataset;
711                         break;
712                 }
713         }
714         up_read(&super->pccs_rw_sem);
715         if (selected)
716                 CDEBUG(D_CACHE, "PCC create, matched %s - %d:%d:%d:%s\n",
717                        dataset->pccd_rule.pmr_conds_str,
718                        matcher->pm_uid, matcher->pm_gid,
719                        matcher->pm_projid, matcher->pm_name->name);
720
721         return selected;
722 }
723
724 /**
725  * pcc_dataset_add - Add a Cache policy to control which files need be
726  * cached and where it will be cached.
727  *
728  * @super:      superblock of pcc
729  * @cmd:        pcc command
730  */
731 static int
732 pcc_dataset_add(struct pcc_super *super, struct pcc_cmd *cmd)
733 {
734         char *pathname = cmd->pccc_pathname;
735         struct pcc_dataset *dataset;
736         struct pcc_dataset *tmp;
737         bool found = false;
738         int rc;
739
740         OBD_ALLOC_PTR(dataset);
741         if (dataset == NULL)
742                 return -ENOMEM;
743
744         rc = kern_path(pathname, LOOKUP_DIRECTORY, &dataset->pccd_path);
745         if (unlikely(rc)) {
746                 OBD_FREE_PTR(dataset);
747                 return rc;
748         }
749         strncpy(dataset->pccd_pathname, pathname, PATH_MAX);
750         dataset->pccd_rwid = cmd->u.pccc_add.pccc_rwid;
751         dataset->pccd_roid = cmd->u.pccc_add.pccc_roid;
752         dataset->pccd_flags = cmd->u.pccc_add.pccc_flags;
753         atomic_set(&dataset->pccd_refcount, 1);
754
755         rc = pcc_dataset_rule_init(&dataset->pccd_rule, cmd);
756         if (rc) {
757                 pcc_dataset_put(dataset);
758                 return rc;
759         }
760
761         down_write(&super->pccs_rw_sem);
762         list_for_each_entry(tmp, &super->pccs_datasets, pccd_linkage) {
763                 if (strcmp(tmp->pccd_pathname, pathname) == 0 ||
764                     (dataset->pccd_rwid != 0 &&
765                      dataset->pccd_rwid == tmp->pccd_rwid) ||
766                     (dataset->pccd_roid != 0 &&
767                      dataset->pccd_roid == tmp->pccd_roid)) {
768                         found = true;
769                         break;
770                 }
771         }
772         if (!found)
773                 list_add(&dataset->pccd_linkage, &super->pccs_datasets);
774         up_write(&super->pccs_rw_sem);
775
776         if (found) {
777                 pcc_dataset_put(dataset);
778                 rc = -EEXIST;
779         }
780
781         return rc;
782 }
783
784 struct pcc_dataset *
785 pcc_dataset_get(struct pcc_super *super, enum lu_pcc_type type, __u32 id)
786 {
787         struct pcc_dataset *dataset;
788         struct pcc_dataset *selected = NULL;
789
790         if (id == 0)
791                 return NULL;
792
793         /*
794          * archive ID (read-write ID) or read-only ID is unique in the list,
795          * we just return last added one as first priority.
796          */
797         down_read(&super->pccs_rw_sem);
798         list_for_each_entry(dataset, &super->pccs_datasets, pccd_linkage) {
799                 if (type == LU_PCC_READWRITE && (dataset->pccd_rwid != id ||
800                     !(dataset->pccd_flags & PCC_DATASET_RWPCC)))
801                         continue;
802                 atomic_inc(&dataset->pccd_refcount);
803                 selected = dataset;
804                 break;
805         }
806         up_read(&super->pccs_rw_sem);
807         if (selected)
808                 CDEBUG(D_CACHE, "matched id %u, PCC mode %d\n", id, type);
809
810         return selected;
811 }
812
813 void
814 pcc_dataset_put(struct pcc_dataset *dataset)
815 {
816         if (atomic_dec_and_test(&dataset->pccd_refcount)) {
817                 pcc_dataset_rule_fini(&dataset->pccd_rule);
818                 path_put(&dataset->pccd_path);
819                 OBD_FREE_PTR(dataset);
820         }
821 }
822
823 static int
824 pcc_dataset_del(struct pcc_super *super, char *pathname)
825 {
826         struct list_head *l, *tmp;
827         struct pcc_dataset *dataset;
828         int rc = -ENOENT;
829
830         down_write(&super->pccs_rw_sem);
831         list_for_each_safe(l, tmp, &super->pccs_datasets) {
832                 dataset = list_entry(l, struct pcc_dataset, pccd_linkage);
833                 if (strcmp(dataset->pccd_pathname, pathname) == 0) {
834                         list_del_init(&dataset->pccd_linkage);
835                         pcc_dataset_put(dataset);
836                         super->pccs_generation++;
837                         rc = 0;
838                         break;
839                 }
840         }
841         up_write(&super->pccs_rw_sem);
842         return rc;
843 }
844
845 static void
846 pcc_dataset_dump(struct pcc_dataset *dataset, struct seq_file *m)
847 {
848         seq_printf(m, "%s:\n", dataset->pccd_pathname);
849         seq_printf(m, "  rwid: %u\n", dataset->pccd_rwid);
850         seq_printf(m, "  flags: %x\n", dataset->pccd_flags);
851         seq_printf(m, "  autocache: %s\n", dataset->pccd_rule.pmr_conds_str);
852 }
853
854 int
855 pcc_super_dump(struct pcc_super *super, struct seq_file *m)
856 {
857         struct pcc_dataset *dataset;
858
859         down_read(&super->pccs_rw_sem);
860         list_for_each_entry(dataset, &super->pccs_datasets, pccd_linkage) {
861                 pcc_dataset_dump(dataset, m);
862         }
863         up_read(&super->pccs_rw_sem);
864         return 0;
865 }
866
867 static void pcc_remove_datasets(struct pcc_super *super)
868 {
869         struct pcc_dataset *dataset, *tmp;
870
871         down_write(&super->pccs_rw_sem);
872         list_for_each_entry_safe(dataset, tmp,
873                                  &super->pccs_datasets, pccd_linkage) {
874                 list_del(&dataset->pccd_linkage);
875                 pcc_dataset_put(dataset);
876         }
877         super->pccs_generation++;
878         up_write(&super->pccs_rw_sem);
879 }
880
881 void pcc_super_fini(struct pcc_super *super)
882 {
883         pcc_remove_datasets(super);
884         put_cred(super->pccs_cred);
885 }
886
887 static bool pathname_is_valid(const char *pathname)
888 {
889         /* Needs to be absolute path */
890         if (pathname == NULL || strlen(pathname) == 0 ||
891             strlen(pathname) >= PATH_MAX || pathname[0] != '/')
892                 return false;
893         return true;
894 }
895
896 static struct pcc_cmd *
897 pcc_cmd_parse(char *buffer, unsigned long count)
898 {
899         static struct pcc_cmd *cmd;
900         char *token;
901         char *val;
902         int rc = 0;
903
904         OBD_ALLOC_PTR(cmd);
905         if (cmd == NULL)
906                 GOTO(out, rc = -ENOMEM);
907
908         /* clear all setting */
909         if (strncmp(buffer, "clear", 5) == 0) {
910                 cmd->pccc_cmd = PCC_CLEAR_ALL;
911                 GOTO(out, rc = 0);
912         }
913
914         val = buffer;
915         token = strsep(&val, " ");
916         if (val == NULL || strlen(val) == 0)
917                 GOTO(out_free_cmd, rc = -EINVAL);
918
919         /* Type of the command */
920         if (strcmp(token, "add") == 0)
921                 cmd->pccc_cmd = PCC_ADD_DATASET;
922         else if (strcmp(token, "del") == 0)
923                 cmd->pccc_cmd = PCC_DEL_DATASET;
924         else
925                 GOTO(out_free_cmd, rc = -EINVAL);
926
927         /* Pathname of the dataset */
928         token = strsep(&val, " ");
929         if ((val == NULL && cmd->pccc_cmd != PCC_DEL_DATASET) ||
930             !pathname_is_valid(token))
931                 GOTO(out_free_cmd, rc = -EINVAL);
932         cmd->pccc_pathname = token;
933
934         if (cmd->pccc_cmd == PCC_ADD_DATASET) {
935                 /* List of ID */
936                 LASSERT(val);
937                 token = val;
938                 val = strrchr(token, '}');
939                 if (!val)
940                         GOTO(out_free_cmd, rc = -EINVAL);
941
942                 /* Skip '}' */
943                 val++;
944                 if (*val == '\0') {
945                         val = NULL;
946                 } else if (*val == ' ') {
947                         *val = '\0';
948                         val++;
949                 } else {
950                         GOTO(out_free_cmd, rc = -EINVAL);
951                 }
952
953                 rc = pcc_id_parse(cmd, token);
954                 if (rc)
955                         GOTO(out_free_cmd, rc);
956
957                 rc = pcc_parse_value_pairs(cmd, val);
958                 if (rc)
959                         GOTO(out_cmd_fini, rc = -EINVAL);
960         }
961         goto out;
962 out_cmd_fini:
963         pcc_cmd_fini(cmd);
964 out_free_cmd:
965         OBD_FREE_PTR(cmd);
966 out:
967         if (rc)
968                 cmd = ERR_PTR(rc);
969         return cmd;
970 }
971
972 int pcc_cmd_handle(char *buffer, unsigned long count,
973                    struct pcc_super *super)
974 {
975         int rc = 0;
976         struct pcc_cmd *cmd;
977
978         cmd = pcc_cmd_parse(buffer, count);
979         if (IS_ERR(cmd))
980                 return PTR_ERR(cmd);
981
982         switch (cmd->pccc_cmd) {
983         case PCC_ADD_DATASET:
984                 rc = pcc_dataset_add(super, cmd);
985                 break;
986         case PCC_DEL_DATASET:
987                 rc = pcc_dataset_del(super, cmd->pccc_pathname);
988                 break;
989         case PCC_CLEAR_ALL:
990                 pcc_remove_datasets(super);
991                 break;
992         default:
993                 rc = -EINVAL;
994                 break;
995         }
996
997         pcc_cmd_fini(cmd);
998         OBD_FREE_PTR(cmd);
999         return rc;
1000 }
1001
1002 static inline void pcc_inode_lock(struct inode *inode)
1003 {
1004         mutex_lock(&ll_i2info(inode)->lli_pcc_lock);
1005 }
1006
1007 static inline void pcc_inode_unlock(struct inode *inode)
1008 {
1009         mutex_unlock(&ll_i2info(inode)->lli_pcc_lock);
1010 }
1011
1012 static void pcc_inode_init(struct pcc_inode *pcci, struct ll_inode_info *lli)
1013 {
1014         pcci->pcci_lli = lli;
1015         lli->lli_pcc_inode = pcci;
1016         atomic_set(&pcci->pcci_refcount, 0);
1017         pcci->pcci_type = LU_PCC_NONE;
1018         pcci->pcci_layout_gen = CL_LAYOUT_GEN_NONE;
1019         atomic_set(&pcci->pcci_active_ios, 0);
1020         init_waitqueue_head(&pcci->pcci_waitq);
1021 }
1022
1023 static void pcc_inode_fini(struct pcc_inode *pcci)
1024 {
1025         struct ll_inode_info *lli = pcci->pcci_lli;
1026
1027         path_put(&pcci->pcci_path);
1028         pcci->pcci_type = LU_PCC_NONE;
1029         OBD_SLAB_FREE_PTR(pcci, pcc_inode_slab);
1030         lli->lli_pcc_inode = NULL;
1031 }
1032
1033 static void pcc_inode_get(struct pcc_inode *pcci)
1034 {
1035         atomic_inc(&pcci->pcci_refcount);
1036 }
1037
1038 static void pcc_inode_put(struct pcc_inode *pcci)
1039 {
1040         if (atomic_dec_and_test(&pcci->pcci_refcount))
1041                 pcc_inode_fini(pcci);
1042 }
1043
1044 void pcc_inode_free(struct inode *inode)
1045 {
1046         struct pcc_inode *pcci = ll_i2pcci(inode);
1047
1048         if (pcci) {
1049                 WARN_ON(atomic_read(&pcci->pcci_refcount) > 1);
1050                 pcc_inode_put(pcci);
1051         }
1052 }
1053
1054 /*
1055  * TODO:
1056  * As Andreas suggested, we'd better use new layout to
1057  * reduce overhead:
1058  * (fid->f_oid >> 16 & oxFFFF)/FID
1059  */
1060 #define MAX_PCC_DATABASE_PATH (6 * 5 + FID_NOBRACE_LEN + 1)
1061 static int pcc_fid2dataset_path(char *buf, int sz, struct lu_fid *fid)
1062 {
1063         return scnprintf(buf, sz, "%04x/%04x/%04x/%04x/%04x/%04x/"
1064                          DFID_NOBRACE,
1065                          (fid)->f_oid       & 0xFFFF,
1066                          (fid)->f_oid >> 16 & 0xFFFF,
1067                          (unsigned int)((fid)->f_seq       & 0xFFFF),
1068                          (unsigned int)((fid)->f_seq >> 16 & 0xFFFF),
1069                          (unsigned int)((fid)->f_seq >> 32 & 0xFFFF),
1070                          (unsigned int)((fid)->f_seq >> 48 & 0xFFFF),
1071                          PFID(fid));
1072 }
1073
1074 static inline const struct cred *pcc_super_cred(struct super_block *sb)
1075 {
1076         return ll_s2sbi(sb)->ll_pcc_super.pccs_cred;
1077 }
1078
1079 void pcc_file_init(struct pcc_file *pccf)
1080 {
1081         pccf->pccf_file = NULL;
1082         pccf->pccf_type = LU_PCC_NONE;
1083 }
1084
1085 static inline bool pcc_auto_attach_enabled(enum pcc_dataset_flags flags,
1086                                            enum pcc_io_type iot)
1087 {
1088         if (iot == PIT_OPEN)
1089                 return flags & PCC_DATASET_OPEN_ATTACH;
1090         if (iot == PIT_GETATTR)
1091                 return flags & PCC_DATASET_STAT_ATTACH;
1092         else
1093                 return flags & PCC_DATASET_AUTO_ATTACH;
1094 }
1095
1096 static const char pcc_xattr_layout[] = XATTR_USER_PREFIX "PCC.layout";
1097
1098 static int pcc_layout_xattr_set(struct pcc_inode *pcci, __u32 gen)
1099 {
1100         struct dentry *pcc_dentry = pcci->pcci_path.dentry;
1101         struct ll_inode_info *lli = pcci->pcci_lli;
1102         int rc;
1103
1104         ENTRY;
1105
1106         if (!(lli->lli_pcc_dsflags & PCC_DATASET_AUTO_ATTACH))
1107                 RETURN(0);
1108
1109         rc = ll_vfs_setxattr(pcc_dentry, pcc_dentry->d_inode, pcc_xattr_layout,
1110                              &gen, sizeof(gen), 0);
1111
1112         RETURN(rc);
1113 }
1114
1115 static int pcc_get_layout_info(struct inode *inode, struct cl_layout *clt)
1116 {
1117         struct lu_env *env;
1118         struct ll_inode_info *lli = ll_i2info(inode);
1119         __u16 refcheck;
1120         int rc;
1121
1122         ENTRY;
1123
1124         if (!lli->lli_clob)
1125                 RETURN(-EINVAL);
1126
1127         env = cl_env_get(&refcheck);
1128         if (IS_ERR(env))
1129                 RETURN(PTR_ERR(env));
1130
1131         rc = cl_object_layout_get(env, lli->lli_clob, clt);
1132         if (rc < 0)
1133                 CDEBUG(D_INODE, "Cannot get layout for "DFID"\n",
1134                        PFID(ll_inode2fid(inode)));
1135
1136         cl_env_put(env, &refcheck);
1137         RETURN(rc < 0 ? rc : 0);
1138 }
1139
1140 static int pcc_fid2dataset_fullpath(char *buf, int sz, struct lu_fid *fid,
1141                                     struct pcc_dataset *dataset)
1142 {
1143         return scnprintf(buf, sz, "%s/%04x/%04x/%04x/%04x/%04x/%04x/"
1144                          DFID_NOBRACE,
1145                          dataset->pccd_pathname,
1146                          (fid)->f_oid       & 0xFFFF,
1147                          (fid)->f_oid >> 16 & 0xFFFF,
1148                          (unsigned int)((fid)->f_seq       & 0xFFFF),
1149                          (unsigned int)((fid)->f_seq >> 16 & 0xFFFF),
1150                          (unsigned int)((fid)->f_seq >> 32 & 0xFFFF),
1151                          (unsigned int)((fid)->f_seq >> 48 & 0xFFFF),
1152                          PFID(fid));
1153 }
1154
1155 /* Must be called with pcci->pcci_lock held */
1156 static void pcc_inode_attach_init(struct pcc_dataset *dataset,
1157                                   struct pcc_inode *pcci,
1158                                   struct dentry *dentry,
1159                                   enum lu_pcc_type type)
1160 {
1161         pcci->pcci_path.mnt = mntget(dataset->pccd_path.mnt);
1162         pcci->pcci_path.dentry = dentry;
1163         LASSERT(atomic_read(&pcci->pcci_refcount) == 0);
1164         atomic_set(&pcci->pcci_refcount, 1);
1165         pcci->pcci_type = type;
1166         pcci->pcci_attr_valid = false;
1167 }
1168
1169 static inline void pcc_inode_dsflags_set(struct ll_inode_info *lli,
1170                                          struct pcc_dataset *dataset)
1171 {
1172         lli->lli_pcc_generation = ll_info2pccs(lli)->pccs_generation;
1173         lli->lli_pcc_dsflags = dataset->pccd_flags;
1174 }
1175
1176 static void pcc_inode_attach_set(struct pcc_super *super,
1177                                  struct pcc_dataset *dataset,
1178                                  struct ll_inode_info *lli,
1179                                  struct pcc_inode *pcci,
1180                                  struct dentry *dentry,
1181                                  enum lu_pcc_type type)
1182 {
1183         pcc_inode_init(pcci, lli);
1184         pcc_inode_attach_init(dataset, pcci, dentry, type);
1185         down_read(&super->pccs_rw_sem);
1186         pcc_inode_dsflags_set(lli, dataset);
1187         up_read(&super->pccs_rw_sem);
1188 }
1189
1190 static inline void pcc_layout_gen_set(struct pcc_inode *pcci,
1191                                       __u32 gen)
1192 {
1193         pcci->pcci_layout_gen = gen;
1194 }
1195
1196 static inline bool pcc_inode_has_layout(struct pcc_inode *pcci)
1197 {
1198         return pcci->pcci_layout_gen != CL_LAYOUT_GEN_NONE;
1199 }
1200
1201 static int pcc_try_dataset_attach(struct inode *inode, __u32 gen,
1202                                   enum lu_pcc_type type,
1203                                   struct pcc_dataset *dataset,
1204                                   bool *cached)
1205 {
1206         struct ll_inode_info *lli = ll_i2info(inode);
1207         struct pcc_inode *pcci = lli->lli_pcc_inode;
1208         const struct cred *old_cred;
1209         struct dentry *pcc_dentry;
1210         struct path path;
1211         char *pathname;
1212         __u32 pcc_gen;
1213         int rc;
1214
1215         ENTRY;
1216
1217         if (type == LU_PCC_READWRITE &&
1218             !(dataset->pccd_flags & PCC_DATASET_RWPCC))
1219                 RETURN(0);
1220
1221         OBD_ALLOC(pathname, PATH_MAX);
1222         if (pathname == NULL)
1223                 RETURN(-ENOMEM);
1224
1225         pcc_fid2dataset_fullpath(pathname, PATH_MAX, &lli->lli_fid, dataset);
1226
1227         old_cred = override_creds(pcc_super_cred(inode->i_sb));
1228         rc = kern_path(pathname, LOOKUP_FOLLOW, &path);
1229         if (rc)
1230                 /* ignore this error */
1231                 GOTO(out, rc = 0);
1232
1233         pcc_dentry = path.dentry;
1234         rc = ll_vfs_getxattr(pcc_dentry, pcc_dentry->d_inode, pcc_xattr_layout,
1235                              &pcc_gen, sizeof(pcc_gen));
1236         if (rc < 0)
1237                 /* ignore this error */
1238                 GOTO(out_put_path, rc = 0);
1239
1240         rc = 0;
1241         /* The file is still valid cached in PCC, attach it immediately. */
1242         if (pcc_gen == gen) {
1243                 CDEBUG(D_CACHE, DFID" L.Gen (%d) consistent, auto attached.\n",
1244                        PFID(&lli->lli_fid), gen);
1245                 if (!pcci) {
1246                         OBD_SLAB_ALLOC_PTR_GFP(pcci, pcc_inode_slab, GFP_NOFS);
1247                         if (pcci == NULL)
1248                                 GOTO(out_put_path, rc = -ENOMEM);
1249
1250                         pcc_inode_init(pcci, lli);
1251                         dget(pcc_dentry);
1252                         pcc_inode_attach_init(dataset, pcci, pcc_dentry, type);
1253                 } else {
1254                         /*
1255                          * This happened when a file was once attached into
1256                          * PCC, and some processes keep this file opened
1257                          * (pcci->refcount > 1) and corresponding PCC file
1258                          * without any I/O activity, and then this file was
1259                          * detached by the manual detach command or the
1260                          * revocation of the layout lock (i.e. cached LRU lock
1261                          * shrinking).
1262                          */
1263                         pcc_inode_get(pcci);
1264                         pcci->pcci_type = type;
1265                 }
1266                 pcc_inode_dsflags_set(lli, dataset);
1267                 pcc_layout_gen_set(pcci, gen);
1268                 *cached = true;
1269         }
1270 out_put_path:
1271         path_put(&path);
1272 out:
1273         revert_creds(old_cred);
1274         OBD_FREE(pathname, PATH_MAX);
1275         RETURN(rc);
1276 }
1277
1278 static int pcc_try_datasets_attach(struct inode *inode, enum pcc_io_type iot,
1279                                    __u32 gen, enum lu_pcc_type type,
1280                                    bool *cached)
1281 {
1282         struct pcc_super *super = &ll_i2sbi(inode)->ll_pcc_super;
1283         struct ll_inode_info *lli = ll_i2info(inode);
1284         struct pcc_dataset *dataset = NULL, *tmp;
1285         int rc = 0;
1286
1287         ENTRY;
1288
1289         down_read(&super->pccs_rw_sem);
1290         list_for_each_entry_safe(dataset, tmp,
1291                                  &super->pccs_datasets, pccd_linkage) {
1292                 if (!pcc_auto_attach_enabled(dataset->pccd_flags, iot))
1293                         break;
1294
1295                 rc = pcc_try_dataset_attach(inode, gen, type, dataset, cached);
1296                 if (rc < 0 || (!rc && *cached))
1297                         break;
1298         }
1299
1300         /*
1301          * Update the saved dataset flags for the inode accordingly if failed.
1302          */
1303         if (!rc && !*cached) {
1304                 /*
1305                  * Currently auto attach strategy for a PCC backend is
1306                  * unchangeable once once it was added into the PCC datasets on
1307                  * a client as the support to change auto attach strategy is
1308                  * not implemented yet.
1309                  */
1310                 /*
1311                  * If tried to attach from one PCC backend:
1312                  * @lli_pcc_generation > 0:
1313                  * 1) The file was once attached into PCC, but now the
1314                  * corresponding PCC backend should be removed from the client;
1315                  * 2) The layout generation was changed, the data has been
1316                  * restored;
1317                  * 3) The corresponding PCC copy is not existed on PCC
1318                  * @lli_pcc_generation == 0:
1319                  * The file is never attached into PCC but in a HSM released
1320                  * state, or once attached into PCC but the inode was evicted
1321                  * from icache later.
1322                  * Set the saved dataset flags with PCC_DATASET_NONE. Then this
1323                  * file will skip from the candidates to try auto attach until
1324                  * the file is attached into PCC again.
1325                  *
1326                  * If the file was never attached into PCC, or once attached but
1327                  * its inode was evicted from icache (lli_pcc_generation == 0),
1328                  * or the corresponding dataset was removed from the client,
1329                  * set the saved dataset flags with PCC_DATASET_NONE.
1330                  *
1331                  * TODO: If the file was once attached into PCC but not try to
1332                  * auto attach due to the change of the configuration parameters
1333                  * for this dataset (i.e. change from auto attach enabled to
1334                  * auto attach disabled for this dataset), update the saved
1335                  * dataset flags with the found one.
1336                  */
1337                 lli->lli_pcc_dsflags = PCC_DATASET_NONE;
1338         }
1339         up_read(&super->pccs_rw_sem);
1340
1341         RETURN(rc);
1342 }
1343
1344 /*
1345  * TODO: For RW-PCC, it is desirable to store HSM info as a layout (LU-10606).
1346  * Thus the client can get archive ID from the layout directly. When try to
1347  * attach the file automatically which is in HSM released state (according to
1348  * LOV_PATTERN_F_RELEASED in the layout), it can determine whether the file is
1349  * valid cached on PCC more precisely according to the @rwid (archive ID) in
1350  * the PCC dataset and the archive ID in HSM attrs.
1351  */
1352 static int pcc_try_auto_attach(struct inode *inode, bool *cached,
1353                                enum pcc_io_type iot)
1354 {
1355         struct pcc_super *super = &ll_i2sbi(inode)->ll_pcc_super;
1356         struct cl_layout clt = {
1357                 .cl_layout_gen = 0,
1358                 .cl_is_released = false,
1359         };
1360         struct ll_inode_info *lli = ll_i2info(inode);
1361         __u32 gen;
1362         int rc;
1363
1364         ENTRY;
1365
1366         /*
1367          * Quick check whether there is PCC device.
1368          */
1369         if (list_empty(&super->pccs_datasets))
1370                 RETURN(0);
1371
1372         /*
1373          * The file layout lock was cancelled. And this open does not
1374          * obtain valid layout lock from MDT (i.e. the file is being
1375          * HSM restoring).
1376          */
1377         if (iot == PIT_OPEN) {
1378                 if (ll_layout_version_get(lli) == CL_LAYOUT_GEN_NONE)
1379                         RETURN(0);
1380         } else {
1381                 rc = ll_layout_refresh(inode, &gen);
1382                 if (rc)
1383                         RETURN(rc);
1384         }
1385
1386         rc = pcc_get_layout_info(inode, &clt);
1387         if (rc)
1388                 RETURN(rc);
1389
1390         if (iot != PIT_OPEN && gen != clt.cl_layout_gen) {
1391                 CDEBUG(D_CACHE, DFID" layout changed from %d to %d.\n",
1392                        PFID(ll_inode2fid(inode)), gen, clt.cl_layout_gen);
1393                 RETURN(-EINVAL);
1394         }
1395
1396         if (clt.cl_is_released)
1397                 rc = pcc_try_datasets_attach(inode, iot, clt.cl_layout_gen,
1398                                              LU_PCC_READWRITE, cached);
1399
1400         RETURN(rc);
1401 }
1402
1403 static inline bool pcc_may_auto_attach(struct inode *inode,
1404                                        enum pcc_io_type iot)
1405 {
1406         struct ll_inode_info *lli = ll_i2info(inode);
1407         struct pcc_super *super = ll_i2pccs(inode);
1408
1409         /* Known the file was not in any PCC backend. */
1410         if (lli->lli_pcc_dsflags & PCC_DATASET_NONE)
1411                 return false;
1412
1413         /*
1414          * lli_pcc_generation == 0 means that the file was never attached into
1415          * PCC, or may be once attached into PCC but detached as the inode is
1416          * evicted from icache (i.e. "echo 3 > /proc/sys/vm/drop_caches" or
1417          * icache shrinking due to the memory pressure), which will cause the
1418          * file detach from PCC when releasing the inode from icache.
1419          * In either case, we still try to attach.
1420          */
1421         /* lli_pcc_generation == 0, or the PCC setting was changed,
1422          * or there is no PCC setup on the client and the try will return
1423          * immediately in pcc_try_auto_attach().
1424          */
1425         if (super->pccs_generation != lli->lli_pcc_generation)
1426                 return true;
1427
1428         /* The cached setting @lli_pcc_dsflags is valid */
1429         if (iot == PIT_OPEN)
1430                 return lli->lli_pcc_dsflags & PCC_DATASET_OPEN_ATTACH;
1431
1432         if (iot == PIT_GETATTR)
1433                 return lli->lli_pcc_dsflags & PCC_DATASET_STAT_ATTACH;
1434
1435         return lli->lli_pcc_dsflags & PCC_DATASET_IO_ATTACH;
1436 }
1437
1438 int pcc_file_open(struct inode *inode, struct file *file)
1439 {
1440         struct pcc_inode *pcci;
1441         struct ll_inode_info *lli = ll_i2info(inode);
1442         struct ll_file_data *fd = file->private_data;
1443         struct pcc_file *pccf = &fd->fd_pcc_file;
1444         struct file *pcc_file;
1445         struct path *path;
1446         bool cached = false;
1447         int rc = 0;
1448
1449         ENTRY;
1450
1451         if (!S_ISREG(inode->i_mode))
1452                 RETURN(0);
1453
1454         pcc_inode_lock(inode);
1455         pcci = ll_i2pcci(inode);
1456
1457         if (lli->lli_pcc_state & PCC_STATE_FL_ATTACHING)
1458                 GOTO(out_unlock, rc = 0);
1459
1460         if (!pcci || !pcc_inode_has_layout(pcci)) {
1461                 if (pcc_may_auto_attach(inode, PIT_OPEN))
1462                         rc = pcc_try_auto_attach(inode, &cached, PIT_OPEN);
1463
1464                 if (rc < 0 || !cached)
1465                         GOTO(out_unlock, rc);
1466
1467                 if (!pcci)
1468                         pcci = ll_i2pcci(inode);
1469         }
1470
1471         pcc_inode_get(pcci);
1472         WARN_ON(pccf->pccf_file);
1473
1474         path = &pcci->pcci_path;
1475         CDEBUG(D_CACHE, "opening pcc file '%pd'\n", path->dentry);
1476
1477         pcc_file = dentry_open(path, file->f_flags,
1478                                pcc_super_cred(inode->i_sb));
1479         if (IS_ERR_OR_NULL(pcc_file)) {
1480                 rc = pcc_file == NULL ? -EINVAL : PTR_ERR(pcc_file);
1481                 pcc_inode_put(pcci);
1482         } else {
1483                 pccf->pccf_file = pcc_file;
1484                 pccf->pccf_type = pcci->pcci_type;
1485         }
1486
1487 out_unlock:
1488         pcc_inode_unlock(inode);
1489         RETURN(rc);
1490 }
1491
1492 void pcc_file_release(struct inode *inode, struct file *file)
1493 {
1494         struct pcc_inode *pcci;
1495         struct ll_file_data *fd = file->private_data;
1496         struct pcc_file *pccf;
1497         struct path *path;
1498
1499         ENTRY;
1500
1501         if (!S_ISREG(inode->i_mode) || fd == NULL)
1502                 RETURN_EXIT;
1503
1504         pccf = &fd->fd_pcc_file;
1505         pcc_inode_lock(inode);
1506         if (pccf->pccf_file == NULL)
1507                 goto out;
1508
1509         pcci = ll_i2pcci(inode);
1510         LASSERT(pcci);
1511         path = &pcci->pcci_path;
1512         CDEBUG(D_CACHE, "releasing pcc file \"%pd\"\n", path->dentry);
1513         pcc_inode_put(pcci);
1514         fput(pccf->pccf_file);
1515         pccf->pccf_file = NULL;
1516 out:
1517         pcc_inode_unlock(inode);
1518         RETURN_EXIT;
1519 }
1520
1521 static void pcc_io_init(struct inode *inode, enum pcc_io_type iot, bool *cached)
1522 {
1523         struct pcc_inode *pcci;
1524
1525         pcc_inode_lock(inode);
1526         pcci = ll_i2pcci(inode);
1527         if (pcci && pcc_inode_has_layout(pcci)) {
1528                 LASSERT(atomic_read(&pcci->pcci_refcount) > 0);
1529                 atomic_inc(&pcci->pcci_active_ios);
1530                 *cached = true;
1531         } else {
1532                 *cached = false;
1533                 if (pcc_may_auto_attach(inode, iot)) {
1534                         (void) pcc_try_auto_attach(inode, cached, iot);
1535                         if (*cached) {
1536                                 pcci = ll_i2pcci(inode);
1537                                 LASSERT(atomic_read(&pcci->pcci_refcount) > 0);
1538                                 atomic_inc(&pcci->pcci_active_ios);
1539                         }
1540                 }
1541         }
1542         pcc_inode_unlock(inode);
1543 }
1544
1545 static void pcc_io_fini(struct inode *inode)
1546 {
1547         struct pcc_inode *pcci = ll_i2pcci(inode);
1548
1549         LASSERT(pcci && atomic_read(&pcci->pcci_active_ios) > 0);
1550         if (atomic_dec_and_test(&pcci->pcci_active_ios))
1551                 wake_up(&pcci->pcci_waitq);
1552 }
1553
1554
1555 static ssize_t
1556 __pcc_file_read_iter(struct kiocb *iocb, struct iov_iter *iter)
1557 {
1558         struct file *file = iocb->ki_filp;
1559
1560 #ifdef HAVE_FILE_OPERATIONS_READ_WRITE_ITER
1561         return file->f_op->read_iter(iocb, iter);
1562 #else
1563         struct iovec iov;
1564         struct iov_iter i;
1565         ssize_t bytes = 0;
1566
1567         iov_for_each(iov, i, *iter) {
1568                 ssize_t res;
1569
1570                 res = file->f_op->aio_read(iocb, &iov, 1, iocb->ki_pos);
1571                 if (-EIOCBQUEUED == res)
1572                         res = wait_on_sync_kiocb(iocb);
1573                 if (res <= 0) {
1574                         if (bytes == 0)
1575                                 bytes = res;
1576                         break;
1577                 }
1578
1579                 bytes += res;
1580                 if (res < iov.iov_len)
1581                         break;
1582         }
1583
1584         if (bytes > 0)
1585                 iov_iter_advance(iter, bytes);
1586         return bytes;
1587 #endif
1588 }
1589
1590 ssize_t pcc_file_read_iter(struct kiocb *iocb,
1591                            struct iov_iter *iter, bool *cached)
1592 {
1593         struct file *file = iocb->ki_filp;
1594         struct ll_file_data *fd = file->private_data;
1595         struct pcc_file *pccf = &fd->fd_pcc_file;
1596         struct inode *inode = file_inode(file);
1597         ssize_t result;
1598
1599         ENTRY;
1600
1601         if (pccf->pccf_file == NULL) {
1602                 *cached = false;
1603                 RETURN(0);
1604         }
1605
1606         pcc_io_init(inode, PIT_READ, cached);
1607         if (!*cached)
1608                 RETURN(0);
1609
1610         iocb->ki_filp = pccf->pccf_file;
1611         /* generic_file_aio_read does not support ext4-dax,
1612          * __pcc_file_read_iter uses ->aio_read hook directly
1613          * to add support for ext4-dax.
1614          */
1615         result = __pcc_file_read_iter(iocb, iter);
1616         iocb->ki_filp = file;
1617
1618         pcc_io_fini(inode);
1619         RETURN(result);
1620 }
1621
1622 static ssize_t
1623 __pcc_file_write_iter(struct kiocb *iocb, struct iov_iter *iter)
1624 {
1625         struct file *file = iocb->ki_filp;
1626
1627 #ifdef HAVE_FILE_OPERATIONS_READ_WRITE_ITER
1628         return file->f_op->write_iter(iocb, iter);
1629 #else
1630         struct iovec iov;
1631         struct iov_iter i;
1632         ssize_t bytes = 0;
1633
1634         iov_for_each(iov, i, *iter) {
1635                 ssize_t res;
1636
1637                 res = file->f_op->aio_write(iocb, &iov, 1, iocb->ki_pos);
1638                 if (-EIOCBQUEUED == res)
1639                         res = wait_on_sync_kiocb(iocb);
1640                 if (res <= 0) {
1641                         if (bytes == 0)
1642                                 bytes = res;
1643                         break;
1644                 }
1645
1646                 bytes += res;
1647                 if (res < iov.iov_len)
1648                         break;
1649         }
1650
1651         if (bytes > 0)
1652                 iov_iter_advance(iter, bytes);
1653         return bytes;
1654 #endif
1655 }
1656
1657 ssize_t pcc_file_write_iter(struct kiocb *iocb,
1658                             struct iov_iter *iter, bool *cached)
1659 {
1660         struct file *file = iocb->ki_filp;
1661         struct ll_file_data *fd = file->private_data;
1662         struct pcc_file *pccf = &fd->fd_pcc_file;
1663         struct inode *inode = file_inode(file);
1664         ssize_t result;
1665
1666         ENTRY;
1667
1668         if (pccf->pccf_file == NULL) {
1669                 *cached = false;
1670                 RETURN(0);
1671         }
1672
1673         if (pccf->pccf_type != LU_PCC_READWRITE) {
1674                 *cached = false;
1675                 RETURN(-EAGAIN);
1676         }
1677
1678         pcc_io_init(inode, PIT_WRITE, cached);
1679         if (!*cached)
1680                 RETURN(0);
1681
1682         if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_PCC_FAKE_ERROR))
1683                 GOTO(out, result = -ENOSPC);
1684
1685         iocb->ki_filp = pccf->pccf_file;
1686
1687         /* Since __pcc_file_write_iter makes write calls via
1688          * the normal vfs interface to the local PCC file system,
1689          * the inode lock is not needed.
1690          */
1691         result = __pcc_file_write_iter(iocb, iter);
1692         iocb->ki_filp = file;
1693 out:
1694         pcc_io_fini(inode);
1695         RETURN(result);
1696 }
1697
1698 int pcc_inode_setattr(struct inode *inode, struct iattr *attr,
1699                       bool *cached)
1700 {
1701         int rc;
1702         const struct cred *old_cred;
1703         struct iattr attr2 = *attr;
1704         struct dentry *pcc_dentry;
1705         struct pcc_inode *pcci;
1706
1707         ENTRY;
1708
1709         if (!S_ISREG(inode->i_mode)) {
1710                 *cached = false;
1711                 RETURN(0);
1712         }
1713
1714         pcc_io_init(inode, PIT_SETATTR, cached);
1715         if (!*cached)
1716                 RETURN(0);
1717
1718         attr2.ia_valid = attr->ia_valid & (ATTR_SIZE | ATTR_ATIME |
1719                          ATTR_ATIME_SET | ATTR_MTIME | ATTR_MTIME_SET |
1720                          ATTR_CTIME | ATTR_UID | ATTR_GID);
1721         pcci = ll_i2pcci(inode);
1722         pcc_dentry = pcci->pcci_path.dentry;
1723         inode_lock(pcc_dentry->d_inode);
1724         old_cred = override_creds(pcc_super_cred(inode->i_sb));
1725         rc = pcc_dentry->d_inode->i_op->setattr(pcc_dentry, &attr2);
1726         revert_creds(old_cred);
1727         inode_unlock(pcc_dentry->d_inode);
1728
1729         pcc_io_fini(inode);
1730         RETURN(rc);
1731 }
1732
1733 int pcc_inode_getattr(struct inode *inode, u32 request_mask,
1734                       unsigned int flags, bool *cached)
1735 {
1736         struct ll_inode_info *lli = ll_i2info(inode);
1737         const struct cred *old_cred;
1738         struct kstat stat;
1739         s64 atime;
1740         s64 mtime;
1741         s64 ctime;
1742         int rc;
1743
1744         ENTRY;
1745
1746         if (!S_ISREG(inode->i_mode)) {
1747                 *cached = false;
1748                 RETURN(0);
1749         }
1750
1751         pcc_io_init(inode, PIT_GETATTR, cached);
1752         if (!*cached)
1753                 RETURN(0);
1754
1755         old_cred = override_creds(pcc_super_cred(inode->i_sb));
1756         rc = ll_vfs_getattr(&ll_i2pcci(inode)->pcci_path, &stat, request_mask,
1757                             flags);
1758         revert_creds(old_cred);
1759         if (rc)
1760                 GOTO(out, rc);
1761
1762         ll_inode_size_lock(inode);
1763         if (test_and_clear_bit(LLIF_UPDATE_ATIME, &lli->lli_flags) ||
1764             inode->i_atime.tv_sec < lli->lli_atime)
1765                 inode->i_atime.tv_sec = lli->lli_atime;
1766
1767         inode->i_mtime.tv_sec = lli->lli_mtime;
1768         inode->i_ctime.tv_sec = lli->lli_ctime;
1769
1770         atime = inode->i_atime.tv_sec;
1771         mtime = inode->i_mtime.tv_sec;
1772         ctime = inode->i_ctime.tv_sec;
1773
1774         if (atime < stat.atime.tv_sec)
1775                 atime = stat.atime.tv_sec;
1776
1777         if (ctime < stat.ctime.tv_sec)
1778                 ctime = stat.ctime.tv_sec;
1779
1780         if (mtime < stat.mtime.tv_sec)
1781                 mtime = stat.mtime.tv_sec;
1782
1783         i_size_write(inode, stat.size);
1784         inode->i_blocks = stat.blocks;
1785
1786         inode->i_atime.tv_sec = atime;
1787         inode->i_mtime.tv_sec = mtime;
1788         inode->i_ctime.tv_sec = ctime;
1789
1790         ll_inode_size_unlock(inode);
1791 out:
1792         pcc_io_fini(inode);
1793         RETURN(rc);
1794 }
1795
1796 #ifdef HAVE_DEFAULT_FILE_SPLICE_READ_EXPORT
1797 ssize_t pcc_file_splice_read(struct file *in_file, loff_t *ppos,
1798                              struct pipe_inode_info *pipe,
1799                              size_t count, unsigned int flags)
1800 {
1801         struct inode *inode = file_inode(in_file);
1802         struct ll_file_data *fd = in_file->private_data;
1803         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1804         bool cached = false;
1805         ssize_t result;
1806
1807         ENTRY;
1808
1809         if (!pcc_file)
1810                 RETURN(default_file_splice_read(in_file, ppos, pipe,
1811                                                 count, flags));
1812
1813         pcc_io_init(inode, PIT_SPLICE_READ, &cached);
1814         if (!cached)
1815                 RETURN(default_file_splice_read(in_file, ppos, pipe,
1816                                                 count, flags));
1817
1818         result = default_file_splice_read(pcc_file, ppos, pipe, count, flags);
1819
1820         pcc_io_fini(inode);
1821         RETURN(result);
1822 }
1823 #endif /* HAVE_DEFAULT_FILE_SPLICE_READ_EXPORT */
1824
1825 int pcc_fsync(struct file *file, loff_t start, loff_t end,
1826               int datasync, bool *cached)
1827 {
1828         struct inode *inode = file_inode(file);
1829         struct ll_file_data *fd = file->private_data;
1830         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1831         int rc;
1832
1833         ENTRY;
1834
1835         if (!pcc_file) {
1836                 *cached = false;
1837                 RETURN(0);
1838         }
1839
1840         pcc_io_init(inode, PIT_FSYNC, cached);
1841         if (!*cached)
1842                 RETURN(0);
1843
1844         rc = file_inode(pcc_file)->i_fop->fsync(pcc_file,
1845                                                 start, end, datasync);
1846
1847         pcc_io_fini(inode);
1848         RETURN(rc);
1849 }
1850
1851 int pcc_file_mmap(struct file *file, struct vm_area_struct *vma,
1852                   bool *cached)
1853 {
1854         struct inode *inode = file_inode(file);
1855         struct ll_file_data *fd = file->private_data;
1856         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1857         struct pcc_inode *pcci;
1858         int rc = 0;
1859
1860         ENTRY;
1861
1862         if (!pcc_file || !file_inode(pcc_file)->i_fop->mmap) {
1863                 *cached = false;
1864                 RETURN(0);
1865         }
1866
1867         pcc_inode_lock(inode);
1868         pcci = ll_i2pcci(inode);
1869         if (pcci && pcc_inode_has_layout(pcci)) {
1870                 LASSERT(atomic_read(&pcci->pcci_refcount) > 1);
1871                 *cached = true;
1872                 vma->vm_file = pcc_file;
1873                 rc = file_inode(pcc_file)->i_fop->mmap(pcc_file, vma);
1874                 vma->vm_file = file;
1875                 /* Save the vm ops of backend PCC */
1876                 vma->vm_private_data = (void *)vma->vm_ops;
1877         } else {
1878                 *cached = false;
1879         }
1880         pcc_inode_unlock(inode);
1881
1882         RETURN(rc);
1883 }
1884
1885 void pcc_vm_open(struct vm_area_struct *vma)
1886 {
1887         struct pcc_inode *pcci;
1888         struct file *file = vma->vm_file;
1889         struct inode *inode = file_inode(file);
1890         struct ll_file_data *fd = file->private_data;
1891         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1892         struct vm_operations_struct *pcc_vm_ops = vma->vm_private_data;
1893
1894         ENTRY;
1895
1896         if (!pcc_file || !pcc_vm_ops || !pcc_vm_ops->open)
1897                 RETURN_EXIT;
1898
1899         pcc_inode_lock(inode);
1900         pcci = ll_i2pcci(inode);
1901         if (pcci && pcc_inode_has_layout(pcci)) {
1902                 vma->vm_file = pcc_file;
1903                 pcc_vm_ops->open(vma);
1904                 vma->vm_file = file;
1905         }
1906         pcc_inode_unlock(inode);
1907         EXIT;
1908 }
1909
1910 void pcc_vm_close(struct vm_area_struct *vma)
1911 {
1912         struct file *file = vma->vm_file;
1913         struct inode *inode = file_inode(file);
1914         struct ll_file_data *fd = file->private_data;
1915         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1916         struct vm_operations_struct *pcc_vm_ops = vma->vm_private_data;
1917
1918         ENTRY;
1919
1920         if (!pcc_file || !pcc_vm_ops || !pcc_vm_ops->close)
1921                 RETURN_EXIT;
1922
1923         pcc_inode_lock(inode);
1924         /* Layout lock maybe revoked here */
1925         vma->vm_file = pcc_file;
1926         pcc_vm_ops->close(vma);
1927         vma->vm_file = file;
1928         pcc_inode_unlock(inode);
1929         EXIT;
1930 }
1931
1932 int pcc_page_mkwrite(struct vm_area_struct *vma, struct vm_fault *vmf,
1933                      bool *cached)
1934 {
1935         struct page *page = vmf->page;
1936         struct mm_struct *mm = vma->vm_mm;
1937         struct file *file = vma->vm_file;
1938         struct inode *inode = file_inode(file);
1939         struct ll_file_data *fd = file->private_data;
1940         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
1941         struct vm_operations_struct *pcc_vm_ops = vma->vm_private_data;
1942         int rc;
1943
1944         ENTRY;
1945
1946         if (!pcc_file || !pcc_vm_ops) {
1947                 *cached = false;
1948                 RETURN(0);
1949         }
1950
1951         if (!pcc_vm_ops->page_mkwrite &&
1952             page->mapping == pcc_file->f_mapping) {
1953                 CDEBUG(D_MMAP,
1954                        "%s: PCC backend fs not support ->page_mkwrite()\n",
1955                        ll_i2sbi(inode)->ll_fsname);
1956                 pcc_ioctl_detach(inode, PCC_DETACH_OPT_UNCACHE);
1957                 mmap_read_unlock(mm);
1958                 *cached = true;
1959                 RETURN(VM_FAULT_RETRY | VM_FAULT_NOPAGE);
1960         }
1961         /* Pause to allow for a race with concurrent detach */
1962         OBD_FAIL_TIMEOUT(OBD_FAIL_LLITE_PCC_MKWRITE_PAUSE, cfs_fail_val);
1963
1964         pcc_io_init(inode, PIT_PAGE_MKWRITE, cached);
1965         if (!*cached) {
1966                 /* This happens when the file is detached from PCC after got
1967                  * the fault page via ->fault() on the inode of the PCC copy.
1968                  * Here it can not simply fall back to normal Lustre I/O path.
1969                  * The reason is that the address space of fault page used by
1970                  * ->page_mkwrite() is still the one of PCC inode. In the
1971                  * normal Lustre ->page_mkwrite() I/O path, it will be wrongly
1972                  * handled as the address space of the fault page is not
1973                  * consistent with the one of the Lustre inode (though the
1974                  * fault page was truncated).
1975                  * As the file is detached from PCC, the fault page must
1976                  * be released frist, and retry the mmap write (->fault() and
1977                  * ->page_mkwrite).
1978                  * We use an ugly and tricky method by returning
1979                  * VM_FAULT_NOPAGE | VM_FAULT_RETRY to the caller
1980                  * __do_page_fault and retry the memory fault handling.
1981                  */
1982                 if (page->mapping == pcc_file->f_mapping) {
1983                         *cached = true;
1984                         mmap_read_unlock(mm);
1985                         RETURN(VM_FAULT_RETRY | VM_FAULT_NOPAGE);
1986                 }
1987
1988                 RETURN(0);
1989         }
1990
1991         /*
1992          * This fault injection can also be used to simulate -ENOSPC and
1993          * -EDQUOT failure of underlying PCC backend fs.
1994          */
1995         if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_PCC_DETACH_MKWRITE)) {
1996                 pcc_io_fini(inode);
1997                 pcc_ioctl_detach(inode, PCC_DETACH_OPT_UNCACHE);
1998                 mmap_read_unlock(mm);
1999                 RETURN(VM_FAULT_RETRY | VM_FAULT_NOPAGE);
2000         }
2001
2002         vma->vm_file = pcc_file;
2003 #ifdef HAVE_VM_OPS_USE_VM_FAULT_ONLY
2004         rc = pcc_vm_ops->page_mkwrite(vmf);
2005 #else
2006         rc = pcc_vm_ops->page_mkwrite(vma, vmf);
2007 #endif
2008         vma->vm_file = file;
2009
2010         pcc_io_fini(inode);
2011         RETURN(rc);
2012 }
2013
2014 int pcc_fault(struct vm_area_struct *vma, struct vm_fault *vmf,
2015               bool *cached)
2016 {
2017         struct file *file = vma->vm_file;
2018         struct inode *inode = file_inode(file);
2019         struct ll_file_data *fd = file->private_data;
2020         struct file *pcc_file = fd->fd_pcc_file.pccf_file;
2021         struct vm_operations_struct *pcc_vm_ops = vma->vm_private_data;
2022         int rc;
2023
2024         ENTRY;
2025
2026         if (!pcc_file || !pcc_vm_ops || !pcc_vm_ops->fault) {
2027                 *cached = false;
2028                 RETURN(0);
2029         }
2030
2031         pcc_io_init(inode, PIT_FAULT, cached);
2032         if (!*cached)
2033                 RETURN(0);
2034
2035         vma->vm_file = pcc_file;
2036 #ifdef HAVE_VM_OPS_USE_VM_FAULT_ONLY
2037         rc = pcc_vm_ops->fault(vmf);
2038 #else
2039         rc = pcc_vm_ops->fault(vma, vmf);
2040 #endif
2041         vma->vm_file = file;
2042
2043         pcc_io_fini(inode);
2044         RETURN(rc);
2045 }
2046
2047 static void __pcc_layout_invalidate(struct pcc_inode *pcci)
2048 {
2049         pcci->pcci_type = LU_PCC_NONE;
2050         pcc_layout_gen_set(pcci, CL_LAYOUT_GEN_NONE);
2051         if (atomic_read(&pcci->pcci_active_ios) == 0)
2052                 return;
2053
2054         CDEBUG(D_CACHE, "Waiting for IO completion: %d\n",
2055                        atomic_read(&pcci->pcci_active_ios));
2056         wait_event_idle(pcci->pcci_waitq,
2057                         atomic_read(&pcci->pcci_active_ios) == 0);
2058 }
2059
2060 void pcc_layout_invalidate(struct inode *inode)
2061 {
2062         struct pcc_inode *pcci;
2063
2064         ENTRY;
2065
2066         pcc_inode_lock(inode);
2067         pcci = ll_i2pcci(inode);
2068         if (pcci && pcc_inode_has_layout(pcci)) {
2069                 LASSERT(atomic_read(&pcci->pcci_refcount) > 0);
2070                 __pcc_layout_invalidate(pcci);
2071
2072                 CDEBUG(D_CACHE, "Invalidate "DFID" layout gen %d\n",
2073                        PFID(&ll_i2info(inode)->lli_fid), pcci->pcci_layout_gen);
2074
2075                 pcc_inode_put(pcci);
2076         }
2077         pcc_inode_unlock(inode);
2078
2079         EXIT;
2080 }
2081
2082 static int pcc_inode_remove(struct inode *inode, struct dentry *pcc_dentry)
2083 {
2084         int rc;
2085
2086         rc = ll_vfs_unlink(pcc_dentry->d_parent->d_inode, pcc_dentry);
2087         if (rc)
2088                 CWARN("%s: failed to unlink PCC file %pd, rc = %d\n",
2089                       ll_i2sbi(inode)->ll_fsname, pcc_dentry, rc);
2090
2091         return rc;
2092 }
2093
2094 /* Create directory under base if directory does not exist */
2095 static struct dentry *
2096 pcc_mkdir(struct dentry *base, const char *name, umode_t mode)
2097 {
2098         int rc;
2099         struct dentry *dentry;
2100         struct inode *dir = base->d_inode;
2101
2102         inode_lock(dir);
2103         dentry = lookup_one_len(name, base, strlen(name));
2104         if (IS_ERR(dentry))
2105                 goto out;
2106
2107         if (d_is_positive(dentry))
2108                 goto out;
2109
2110         rc = vfs_mkdir(dir, dentry, mode);
2111         if (rc) {
2112                 dput(dentry);
2113                 dentry = ERR_PTR(rc);
2114                 goto out;
2115         }
2116 out:
2117         inode_unlock(dir);
2118         return dentry;
2119 }
2120
2121 static struct dentry *
2122 pcc_mkdir_p(struct dentry *root, char *path, umode_t mode)
2123 {
2124         char *ptr, *entry_name;
2125         struct dentry *parent;
2126         struct dentry *child = ERR_PTR(-EINVAL);
2127
2128         ptr = path;
2129         while (*ptr == '/')
2130                 ptr++;
2131
2132         entry_name = ptr;
2133         parent = dget(root);
2134         while ((ptr = strchr(ptr, '/')) != NULL) {
2135                 *ptr = '\0';
2136                 child = pcc_mkdir(parent, entry_name, mode);
2137                 *ptr = '/';
2138                 dput(parent);
2139                 if (IS_ERR(child))
2140                         break;
2141
2142                 parent = child;
2143                 ptr++;
2144                 entry_name = ptr;
2145         }
2146
2147         return child;
2148 }
2149
2150 /* Create file under base. If file already exist, return failure */
2151 static struct dentry *
2152 pcc_create(struct dentry *base, const char *name, umode_t mode)
2153 {
2154         int rc;
2155         struct dentry *dentry;
2156         struct inode *dir = base->d_inode;
2157
2158         inode_lock(dir);
2159         dentry = lookup_one_len(name, base, strlen(name));
2160         if (IS_ERR(dentry))
2161                 goto out;
2162
2163         if (d_is_positive(dentry))
2164                 goto out;
2165
2166         rc = vfs_create(dir, dentry, mode, false);
2167         if (rc) {
2168                 dput(dentry);
2169                 dentry = ERR_PTR(rc);
2170                 goto out;
2171         }
2172 out:
2173         inode_unlock(dir);
2174         return dentry;
2175 }
2176
2177 static int __pcc_inode_create(struct pcc_dataset *dataset,
2178                               struct lu_fid *fid,
2179                               struct dentry **dentry)
2180 {
2181         char *path;
2182         struct dentry *base;
2183         struct dentry *child;
2184         int rc = 0;
2185
2186         OBD_ALLOC(path, MAX_PCC_DATABASE_PATH);
2187         if (path == NULL)
2188                 return -ENOMEM;
2189
2190         pcc_fid2dataset_path(path, MAX_PCC_DATABASE_PATH, fid);
2191
2192         base = pcc_mkdir_p(dataset->pccd_path.dentry, path, 0);
2193         if (IS_ERR(base)) {
2194                 rc = PTR_ERR(base);
2195                 GOTO(out, rc);
2196         }
2197
2198         snprintf(path, MAX_PCC_DATABASE_PATH, DFID_NOBRACE, PFID(fid));
2199         child = pcc_create(base, path, 0);
2200         if (IS_ERR(child)) {
2201                 rc = PTR_ERR(child);
2202                 GOTO(out_base, rc);
2203         }
2204         *dentry = child;
2205
2206 out_base:
2207         dput(base);
2208 out:
2209         OBD_FREE(path, MAX_PCC_DATABASE_PATH);
2210         return rc;
2211 }
2212
2213 /*
2214  * Reset uid, gid or size for the PCC copy masked by @valid.
2215  * TODO: Set the project ID for PCC copy.
2216  */
2217 int pcc_inode_reset_iattr(struct dentry *dentry, unsigned int valid,
2218                           kuid_t uid, kgid_t gid, loff_t size)
2219 {
2220         struct inode *inode = dentry->d_inode;
2221         struct iattr attr;
2222         int rc;
2223
2224         ENTRY;
2225
2226         attr.ia_valid = valid;
2227         attr.ia_uid = uid;
2228         attr.ia_gid = gid;
2229         attr.ia_size = size;
2230
2231         inode_lock(inode);
2232         rc = notify_change(dentry, &attr, NULL);
2233         inode_unlock(inode);
2234
2235         RETURN(rc);
2236 }
2237
2238 int pcc_inode_create(struct super_block *sb, struct pcc_dataset *dataset,
2239                      struct lu_fid *fid, struct dentry **pcc_dentry)
2240 {
2241         const struct cred *old_cred;
2242         int rc;
2243
2244         old_cred = override_creds(pcc_super_cred(sb));
2245         rc = __pcc_inode_create(dataset, fid, pcc_dentry);
2246         revert_creds(old_cred);
2247         return rc;
2248 }
2249
2250 int pcc_inode_create_fini(struct inode *inode, struct pcc_create_attach *pca)
2251 {
2252         struct dentry *pcc_dentry = pca->pca_dentry;
2253         struct pcc_super *super = ll_i2pccs(inode);
2254         const struct cred *old_cred;
2255         struct pcc_inode *pcci;
2256         int rc;
2257
2258         ENTRY;
2259
2260         if (!pca->pca_dataset)
2261                 RETURN(0);
2262
2263         if (!inode)
2264                 GOTO(out_dataset_put, rc = 0);
2265
2266         LASSERT(pcc_dentry);
2267
2268         old_cred = override_creds(super->pccs_cred);
2269         pcc_inode_lock(inode);
2270         LASSERT(ll_i2pcci(inode) == NULL);
2271         OBD_SLAB_ALLOC_PTR_GFP(pcci, pcc_inode_slab, GFP_NOFS);
2272         if (pcci == NULL)
2273                 GOTO(out_put, rc = -ENOMEM);
2274
2275         rc = pcc_inode_reset_iattr(pcc_dentry, ATTR_UID | ATTR_GID,
2276                                    old_cred->suid, old_cred->sgid, 0);
2277         if (rc)
2278                 GOTO(out_put, rc);
2279
2280         pcc_inode_attach_set(super, pca->pca_dataset, ll_i2info(inode),
2281                              pcci, pcc_dentry, LU_PCC_READWRITE);
2282
2283         rc = pcc_layout_xattr_set(pcci, 0);
2284         if (rc) {
2285                 (void) pcc_inode_remove(inode, pcci->pcci_path.dentry);
2286                 pcc_inode_put(pcci);
2287                 GOTO(out_unlock, rc);
2288         }
2289
2290         /* Set the layout generation of newly created file with 0 */
2291         pcc_layout_gen_set(pcci, 0);
2292
2293 out_put:
2294         if (rc) {
2295                 (void) pcc_inode_remove(inode, pcc_dentry);
2296                 dput(pcc_dentry);
2297
2298                 if (pcci)
2299                         OBD_SLAB_FREE_PTR(pcci, pcc_inode_slab);
2300         }
2301 out_unlock:
2302         pcc_inode_unlock(inode);
2303         revert_creds(old_cred);
2304 out_dataset_put:
2305         pcc_dataset_put(pca->pca_dataset);
2306         RETURN(rc);
2307 }
2308
2309 void pcc_create_attach_cleanup(struct super_block *sb,
2310                                struct pcc_create_attach *pca)
2311 {
2312         if (!pca->pca_dataset)
2313                 return;
2314
2315         if (pca->pca_dentry) {
2316                 const struct cred *old_cred;
2317                 int rc;
2318
2319                 old_cred = override_creds(pcc_super_cred(sb));
2320                 rc = ll_vfs_unlink(pca->pca_dentry->d_parent->d_inode,
2321                                    pca->pca_dentry);
2322                 if (rc)
2323                         CWARN("%s: failed to unlink PCC file %pd: rc = %d\n",
2324                               ll_s2sbi(sb)->ll_fsname, pca->pca_dentry, rc);
2325                 /* ignore the unlink failure */
2326                 revert_creds(old_cred);
2327                 dput(pca->pca_dentry);
2328         }
2329
2330         pcc_dataset_put(pca->pca_dataset);
2331 }
2332
2333 static int pcc_filp_write(struct file *filp, const void *buf, ssize_t count,
2334                           loff_t *offset)
2335 {
2336         while (count > 0) {
2337                 ssize_t size;
2338
2339                 size = cfs_kernel_write(filp, buf, count, offset);
2340                 if (size < 0)
2341                         return size;
2342                 count -= size;
2343                 buf += size;
2344         }
2345         return 0;
2346 }
2347
2348 static ssize_t pcc_copy_data(struct file *src, struct file *dst)
2349 {
2350         ssize_t rc = 0;
2351         ssize_t rc2;
2352         loff_t pos, offset = 0;
2353         size_t buf_len = 1048576;
2354         void *buf;
2355
2356         ENTRY;
2357
2358         OBD_ALLOC_LARGE(buf, buf_len);
2359         if (buf == NULL)
2360                 RETURN(-ENOMEM);
2361
2362         while (1) {
2363                 if (signal_pending(current))
2364                         GOTO(out_free, rc = -EINTR);
2365
2366                 pos = offset;
2367                 rc2 = cfs_kernel_read(src, buf, buf_len, &pos);
2368                 if (rc2 < 0)
2369                         GOTO(out_free, rc = rc2);
2370                 else if (rc2 == 0)
2371                         break;
2372
2373                 pos = offset;
2374                 rc = pcc_filp_write(dst, buf, rc2, &pos);
2375                 if (rc < 0)
2376                         GOTO(out_free, rc);
2377                 offset += rc2;
2378         }
2379
2380         rc = offset;
2381 out_free:
2382         OBD_FREE_LARGE(buf, buf_len);
2383         RETURN(rc);
2384 }
2385
2386 static int pcc_attach_allowed_check(struct inode *inode)
2387 {
2388         struct ll_inode_info *lli = ll_i2info(inode);
2389         struct pcc_inode *pcci;
2390         int rc = 0;
2391
2392         ENTRY;
2393
2394         pcc_inode_lock(inode);
2395         if (lli->lli_pcc_state & PCC_STATE_FL_ATTACHING)
2396                 GOTO(out_unlock, rc = -EBUSY);
2397
2398         pcci = ll_i2pcci(inode);
2399         if (pcci && pcc_inode_has_layout(pcci))
2400                 GOTO(out_unlock, rc = -EEXIST);
2401
2402         lli->lli_pcc_state |= PCC_STATE_FL_ATTACHING;
2403 out_unlock:
2404         pcc_inode_unlock(inode);
2405         RETURN(rc);
2406 }
2407
2408 int pcc_readwrite_attach(struct file *file, struct inode *inode,
2409                          __u32 archive_id)
2410 {
2411         struct pcc_dataset *dataset;
2412         struct ll_inode_info *lli = ll_i2info(inode);
2413         struct pcc_super *super = ll_i2pccs(inode);
2414         struct pcc_inode *pcci;
2415         const struct cred *old_cred;
2416         struct dentry *dentry;
2417         struct file *pcc_filp;
2418         struct path path;
2419         ssize_t ret;
2420         int rc;
2421
2422         ENTRY;
2423
2424         rc = pcc_attach_allowed_check(inode);
2425         if (rc)
2426                 RETURN(rc);
2427
2428         dataset = pcc_dataset_get(&ll_i2sbi(inode)->ll_pcc_super,
2429                                   LU_PCC_READWRITE, archive_id);
2430         if (dataset == NULL)
2431                 RETURN(-ENOENT);
2432
2433         old_cred = override_creds(super->pccs_cred);
2434         rc = __pcc_inode_create(dataset, &lli->lli_fid, &dentry);
2435         if (rc)
2436                 GOTO(out_dataset_put, rc);
2437
2438         path.mnt = dataset->pccd_path.mnt;
2439         path.dentry = dentry;
2440         pcc_filp = dentry_open(&path, O_WRONLY | O_LARGEFILE, current_cred());
2441         if (IS_ERR_OR_NULL(pcc_filp)) {
2442                 rc = pcc_filp == NULL ? -EINVAL : PTR_ERR(pcc_filp);
2443                 GOTO(out_dentry, rc);
2444         }
2445
2446         rc = pcc_inode_reset_iattr(dentry, ATTR_UID | ATTR_GID,
2447                                    old_cred->uid, old_cred->gid, 0);
2448         if (rc)
2449                 GOTO(out_fput, rc);
2450
2451         ret = pcc_copy_data(file, pcc_filp);
2452         if (ret < 0)
2453                 GOTO(out_fput, rc = ret);
2454
2455         /*
2456          * It must to truncate the PCC copy to the same size of the Lustre
2457          * copy after copy data. Otherwise, it may get wrong file size after
2458          * re-attach a file. See LU-13023 for details.
2459          */
2460         rc = pcc_inode_reset_iattr(dentry, ATTR_SIZE, KUIDT_INIT(0),
2461                                    KGIDT_INIT(0), ret);
2462         if (rc)
2463                 GOTO(out_fput, rc);
2464
2465         /* Pause to allow for a race with concurrent HSM remove */
2466         OBD_FAIL_TIMEOUT(OBD_FAIL_LLITE_PCC_ATTACH_PAUSE, cfs_fail_val);
2467
2468         pcc_inode_lock(inode);
2469         pcci = ll_i2pcci(inode);
2470         LASSERT(!pcci);
2471         OBD_SLAB_ALLOC_PTR_GFP(pcci, pcc_inode_slab, GFP_NOFS);
2472         if (pcci == NULL)
2473                 GOTO(out_unlock, rc = -ENOMEM);
2474
2475         pcc_inode_attach_set(super, dataset, lli, pcci,
2476                              dentry, LU_PCC_READWRITE);
2477 out_unlock:
2478         pcc_inode_unlock(inode);
2479 out_fput:
2480         fput(pcc_filp);
2481 out_dentry:
2482         if (rc) {
2483                 (void) pcc_inode_remove(inode, dentry);
2484                 dput(dentry);
2485         }
2486 out_dataset_put:
2487         pcc_dataset_put(dataset);
2488         revert_creds(old_cred);
2489
2490         RETURN(rc);
2491 }
2492
2493 int pcc_readwrite_attach_fini(struct file *file, struct inode *inode,
2494                               __u32 gen, bool lease_broken, int rc,
2495                               bool attached)
2496 {
2497         struct ll_inode_info *lli = ll_i2info(inode);
2498         const struct cred *old_cred;
2499         struct pcc_inode *pcci;
2500         __u32 gen2;
2501
2502         ENTRY;
2503
2504         old_cred = override_creds(pcc_super_cred(inode->i_sb));
2505         pcc_inode_lock(inode);
2506         pcci = ll_i2pcci(inode);
2507         if (rc || lease_broken) {
2508                 if (attached && pcci)
2509                         pcc_inode_put(pcci);
2510
2511                 GOTO(out_unlock, rc);
2512         }
2513
2514         /* PCC inode may be released due to layout lock revocatioin */
2515         if (!pcci)
2516                 GOTO(out_unlock, rc = -ESTALE);
2517
2518         LASSERT(attached);
2519         rc = pcc_layout_xattr_set(pcci, gen);
2520         if (rc)
2521                 GOTO(out_put, rc);
2522
2523         LASSERT(lli->lli_pcc_state & PCC_STATE_FL_ATTACHING);
2524         rc = ll_layout_refresh(inode, &gen2);
2525         if (!rc) {
2526                 if (gen2 == gen) {
2527                         pcc_layout_gen_set(pcci, gen);
2528                 } else {
2529                         CDEBUG(D_CACHE,
2530                                DFID" layout changed from %d to %d.\n",
2531                                PFID(ll_inode2fid(inode)), gen, gen2);
2532                         GOTO(out_put, rc = -ESTALE);
2533                 }
2534         }
2535
2536 out_put:
2537         if (rc) {
2538                 (void) pcc_inode_remove(inode, pcci->pcci_path.dentry);
2539                 pcc_inode_put(pcci);
2540         }
2541 out_unlock:
2542         lli->lli_pcc_state &= ~PCC_STATE_FL_ATTACHING;
2543         pcc_inode_unlock(inode);
2544         revert_creds(old_cred);
2545         RETURN(rc);
2546 }
2547
2548 static int pcc_hsm_remove(struct inode *inode)
2549 {
2550         struct hsm_user_request *hur;
2551         __u32 gen;
2552         int len;
2553         int rc;
2554
2555         ENTRY;
2556
2557         rc = ll_layout_restore(inode, 0, OBD_OBJECT_EOF);
2558         if (rc) {
2559                 CDEBUG(D_CACHE, DFID" RESTORE failure: %d\n",
2560                        PFID(&ll_i2info(inode)->lli_fid), rc);
2561                 RETURN(rc);
2562         }
2563
2564         ll_layout_refresh(inode, &gen);
2565
2566         len = sizeof(struct hsm_user_request) +
2567               sizeof(struct hsm_user_item);
2568         OBD_ALLOC(hur, len);
2569         if (hur == NULL)
2570                 RETURN(-ENOMEM);
2571
2572         hur->hur_request.hr_action = HUA_REMOVE;
2573         hur->hur_request.hr_archive_id = 0;
2574         hur->hur_request.hr_flags = 0;
2575         memcpy(&hur->hur_user_item[0].hui_fid, &ll_i2info(inode)->lli_fid,
2576                sizeof(hur->hur_user_item[0].hui_fid));
2577         hur->hur_user_item[0].hui_extent.offset = 0;
2578         hur->hur_user_item[0].hui_extent.length = OBD_OBJECT_EOF;
2579         hur->hur_request.hr_itemcount = 1;
2580         rc = obd_iocontrol(LL_IOC_HSM_REQUEST, ll_i2sbi(inode)->ll_md_exp,
2581                            len, hur, NULL);
2582         if (rc)
2583                 CDEBUG(D_CACHE, DFID" HSM REMOVE failure: %d\n",
2584                        PFID(&ll_i2info(inode)->lli_fid), rc);
2585
2586         OBD_FREE(hur, len);
2587         RETURN(rc);
2588 }
2589
2590 int pcc_ioctl_detach(struct inode *inode, __u32 opt)
2591 {
2592         struct ll_inode_info *lli = ll_i2info(inode);
2593         struct pcc_inode *pcci;
2594         bool hsm_remove = false;
2595         int rc = 0;
2596
2597         ENTRY;
2598
2599         pcc_inode_lock(inode);
2600         pcci = lli->lli_pcc_inode;
2601         if (!pcci || lli->lli_pcc_state & PCC_STATE_FL_ATTACHING ||
2602             !pcc_inode_has_layout(pcci))
2603                 GOTO(out_unlock, rc = 0);
2604
2605         LASSERT(atomic_read(&pcci->pcci_refcount) > 0);
2606
2607         if (pcci->pcci_type == LU_PCC_READWRITE) {
2608                 if (opt == PCC_DETACH_OPT_UNCACHE) {
2609                         hsm_remove = true;
2610                         /*
2611                          * The file will be removed from PCC, set the flags
2612                          * with PCC_DATASET_NONE even the later removal of the
2613                          * PCC copy fails.
2614                          */
2615                         lli->lli_pcc_dsflags = PCC_DATASET_NONE;
2616                 }
2617
2618                 __pcc_layout_invalidate(pcci);
2619                 pcc_inode_put(pcci);
2620         }
2621
2622 out_unlock:
2623         pcc_inode_unlock(inode);
2624         if (hsm_remove) {
2625                 const struct cred *old_cred;
2626
2627                 old_cred = override_creds(pcc_super_cred(inode->i_sb));
2628                 rc = pcc_hsm_remove(inode);
2629                 revert_creds(old_cred);
2630         }
2631
2632         RETURN(rc);
2633 }
2634
2635 int pcc_ioctl_state(struct file *file, struct inode *inode,
2636                     struct lu_pcc_state *state)
2637 {
2638         int rc = 0;
2639         int count;
2640         char *buf;
2641         char *path;
2642         int buf_len = sizeof(state->pccs_path);
2643         struct ll_file_data *fd = file->private_data;
2644         struct pcc_file *pccf = &fd->fd_pcc_file;
2645         struct pcc_inode *pcci;
2646
2647         ENTRY;
2648
2649         if (buf_len <= 0)
2650                 RETURN(-EINVAL);
2651
2652         OBD_ALLOC(buf, buf_len);
2653         if (buf == NULL)
2654                 RETURN(-ENOMEM);
2655
2656         pcc_inode_lock(inode);
2657         pcci = ll_i2pcci(inode);
2658         if (pcci == NULL) {
2659                 state->pccs_type = LU_PCC_NONE;
2660                 GOTO(out_unlock, rc = 0);
2661         }
2662
2663         count = atomic_read(&pcci->pcci_refcount);
2664         if (count == 0) {
2665                 state->pccs_type = LU_PCC_NONE;
2666                 state->pccs_open_count = 0;
2667                 GOTO(out_unlock, rc = 0);
2668         }
2669
2670         if (pcc_inode_has_layout(pcci))
2671                 count--;
2672         if (pccf->pccf_file != NULL)
2673                 count--;
2674         state->pccs_type = pcci->pcci_type;
2675         state->pccs_open_count = count;
2676         state->pccs_flags = ll_i2info(inode)->lli_pcc_state;
2677         path = dentry_path_raw(pcci->pcci_path.dentry, buf, buf_len);
2678         if (IS_ERR(path))
2679                 GOTO(out_unlock, rc = PTR_ERR(path));
2680
2681         if (strlcpy(state->pccs_path, path, buf_len) >= buf_len)
2682                 GOTO(out_unlock, rc = -ENAMETOOLONG);
2683
2684 out_unlock:
2685         pcc_inode_unlock(inode);
2686         OBD_FREE(buf, buf_len);
2687         RETURN(rc);
2688 }