1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
3 *
4 * Author: Peter J. Braam <braam@clusterfs.com>
5 * Copyright (C) 1998 Stelias Computing Inc
6 * Copyright (C) 1999 Red Hat Inc.
7 *
8 * This file is part of InterMezzo, http://www.inter-mezzo.org.
9 *
10 * InterMezzo is free software; you can redistribute it and/or
11 * modify it under the terms of version 2 of the GNU General Public
12 * License as published by the Free Software Foundation.
13 *
14 * InterMezzo is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 * GNU General Public License for more details.
18 *
19 * You should have received a copy of the GNU General Public License
20 * along with InterMezzo; if not, write to the Free Software
21 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
22 *
23 * This file implements basic routines supporting the semantics
24 */
25 #include <linux/types.h>
26 #include <linux/kernel.h>
27 #include <linux/sched.h>
28 #include <linux/fs.h>
29 #include <linux/stat.h>
30 #include <linux/errno.h>
31 #include <linux/vmalloc.h>
32 #include <linux/slab.h>
33 #include <linux/locks.h>
34 #include <asm/segment.h>
35 #include <asm/uaccess.h>
36 #include <linux/string.h>
37 #include <linux/smp_lock.h>
38
39 #include <linux/intermezzo_fs.h>
40 #include <linux/intermezzo_psdev.h>
41
presto_walk(const char * name,struct nameidata * nd)42 int presto_walk(const char *name, struct nameidata *nd)
43 {
44 int err;
45 /* we do not follow symlinks to support symlink operations
46 correctly. The vfs should always hand us resolved dentries
47 so we should not be required to use LOOKUP_FOLLOW. At the
48 reintegrating end, lento again should be working with the
49 resolved pathname and not the symlink. SHP
50 XXX: This code implies that direct symlinks do not work. SHP
51 */
52 unsigned int flags = LOOKUP_POSITIVE;
53
54 ENTRY;
55 err = 0;
56 if (path_init(name, flags, nd))
57 err = path_walk(name, nd);
58 return err;
59 }
60
61
62 /* find the presto minor device for this inode */
presto_i2m(struct inode * inode)63 int presto_i2m(struct inode *inode)
64 {
65 struct presto_cache *cache;
66 ENTRY;
67 cache = presto_get_cache(inode);
68 CDEBUG(D_PSDEV, "\n");
69 if ( !cache ) {
70 CERROR("PRESTO: BAD: cannot find cache for dev %d, ino %ld\n",
71 inode->i_dev, inode->i_ino);
72 EXIT;
73 return -1;
74 }
75 EXIT;
76 return cache->cache_psdev->uc_minor;
77 }
78
presto_f2m(struct presto_file_set * fset)79 inline int presto_f2m(struct presto_file_set *fset)
80 {
81 return fset->fset_cache->cache_psdev->uc_minor;
82
83 }
84
presto_c2m(struct presto_cache * cache)85 inline int presto_c2m(struct presto_cache *cache)
86 {
87 return cache->cache_psdev->uc_minor;
88
89 }
90
91 /* XXX check this out */
presto_path2fileset(const char * name)92 struct presto_file_set *presto_path2fileset(const char *name)
93 {
94 struct nameidata nd;
95 struct presto_file_set *fileset;
96 int error;
97 ENTRY;
98
99 error = presto_walk(name, &nd);
100 if (!error) {
101 #if 0
102 error = do_revalidate(nd.dentry);
103 #endif
104 if (!error)
105 fileset = presto_fset(nd.dentry);
106 path_release(&nd);
107 EXIT;
108 } else
109 fileset = ERR_PTR(error);
110
111 EXIT;
112 return fileset;
113 }
114
115 /* check a flag on this dentry or fset root. Semantics:
116 - most flags: test if it is set
117 - PRESTO_ATTR, PRESTO_DATA return 1 if PRESTO_FSETINSYNC is set
118 */
presto_chk(struct dentry * dentry,int flag)119 int presto_chk(struct dentry *dentry, int flag)
120 {
121 int minor;
122 struct presto_file_set *fset = presto_fset(dentry);
123
124 ENTRY;
125 minor = presto_i2m(dentry->d_inode);
126 if ( izo_channels[minor].uc_no_filter ) {
127 EXIT;
128 return ~0;
129 }
130
131 /* if the fileset is in sync DATA and ATTR are OK */
132 if ( fset &&
133 (flag == PRESTO_ATTR || flag == PRESTO_DATA) &&
134 (fset->fset_flags & FSET_INSYNC) ) {
135 CDEBUG(D_INODE, "fset in sync (ino %ld)!\n",
136 fset->fset_dentry->d_inode->i_ino);
137 EXIT;
138 return 1;
139 }
140
141 EXIT;
142 return (presto_d2d(dentry)->dd_flags & flag);
143 }
144
145 /* set a bit in the dentry flags */
presto_set(struct dentry * dentry,int flag)146 void presto_set(struct dentry *dentry, int flag)
147 {
148 ENTRY;
149 if ( dentry->d_inode ) {
150 CDEBUG(D_INODE, "SET ino %ld, flag %x\n",
151 dentry->d_inode->i_ino, flag);
152 }
153 if ( presto_d2d(dentry) == NULL) {
154 CERROR("dentry without d_fsdata in presto_set: %p: %*s", dentry,
155 dentry->d_name.len, dentry->d_name.name);
156 BUG();
157 }
158 presto_d2d(dentry)->dd_flags |= flag;
159 EXIT;
160 }
161
162 /* given a path: complete the closes on the fset */
lento_complete_closes(char * path)163 int lento_complete_closes(char *path)
164 {
165 struct nameidata nd;
166 struct dentry *dentry;
167 int error;
168 struct presto_file_set *fset;
169 ENTRY;
170
171 error = presto_walk(path, &nd);
172 if (error) {
173 EXIT;
174 return error;
175 }
176
177 dentry = nd.dentry;
178
179 error = -ENXIO;
180 if ( !presto_ispresto(dentry->d_inode) ) {
181 EXIT;
182 goto out_complete;
183 }
184
185 fset = presto_fset(dentry);
186 error = -EINVAL;
187 if ( !fset ) {
188 CERROR("No fileset!\n");
189 EXIT;
190 goto out_complete;
191 }
192
193 /* transactions and locking are internal to this function */
194 error = presto_complete_lml(fset);
195
196 EXIT;
197 out_complete:
198 path_release(&nd);
199 return error;
200 }
201
202 #if 0
203 /* given a path: write a close record and cancel an LML record, finally
204 call truncate LML. Lento is doing this so it goes in with uid/gid's
205 root.
206 */
207 int lento_cancel_lml(char *path,
208 __u64 lml_offset,
209 __u64 remote_ino,
210 __u32 remote_generation,
211 __u32 remote_version,
212 struct lento_vfs_context *info)
213 {
214 struct nameidata nd;
215 struct rec_info rec;
216 struct dentry *dentry;
217 int error;
218 struct presto_file_set *fset;
219 void *handle;
220 struct presto_version new_ver;
221 ENTRY;
222
223
224 error = presto_walk(path, &nd);
225 if (error) {
226 EXIT;
227 return error;
228 }
229 dentry = nd.dentry;
230
231 error = -ENXIO;
232 if ( !presto_ispresto(dentry->d_inode) ) {
233 EXIT;
234 goto out_cancel_lml;
235 }
236
237 fset = presto_fset(dentry);
238
239 error=-EINVAL;
240 if (fset==NULL) {
241 CERROR("No fileset!\n");
242 EXIT;
243 goto out_cancel_lml;
244 }
245
246 /* this only requires a transaction below which is automatic */
247 handle = presto_trans_start(fset, dentry->d_inode, PRESTO_OP_RELEASE);
248 if ( IS_ERR(handle) ) {
249 error = -ENOMEM;
250 EXIT;
251 goto out_cancel_lml;
252 }
253
254 if (info->flags & LENTO_FL_CANCEL_LML) {
255 error = presto_clear_lml_close(fset, lml_offset);
256 if ( error ) {
257 presto_trans_commit(fset, handle);
258 EXIT;
259 goto out_cancel_lml;
260 }
261 }
262
263
264 if (info->flags & LENTO_FL_WRITE_KML) {
265 struct file file;
266 file.private_data = NULL;
267 file.f_dentry = dentry;
268 presto_getversion(&new_ver, dentry->d_inode);
269 error = presto_journal_close(&rec, fset, &file, dentry,
270 &new_ver);
271 if ( error ) {
272 EXIT;
273 presto_trans_commit(fset, handle);
274 goto out_cancel_lml;
275 }
276 }
277
278 if (info->flags & LENTO_FL_WRITE_EXPECT) {
279 error = presto_write_last_rcvd(&rec, fset, info);
280 if ( error < 0 ) {
281 EXIT;
282 presto_trans_commit(fset, handle);
283 goto out_cancel_lml;
284 }
285 }
286
287 presto_trans_commit(fset, handle);
288
289 if (info->flags & LENTO_FL_CANCEL_LML) {
290 presto_truncate_lml(fset);
291 }
292
293
294 out_cancel_lml:
295 EXIT;
296 path_release(&nd);
297 return error;
298 }
299 #endif
300
301 /* given a dentry, operate on the flags in its dentry. Used by downcalls */
izo_mark_dentry(struct dentry * dentry,int and_flag,int or_flag,int * res)302 int izo_mark_dentry(struct dentry *dentry, int and_flag, int or_flag,
303 int *res)
304 {
305 int error = 0;
306
307 if (presto_d2d(dentry) == NULL) {
308 CERROR("InterMezzo: no ddata for inode %ld in %s\n",
309 dentry->d_inode->i_ino, __FUNCTION__);
310 return -EINVAL;
311 }
312
313 CDEBUG(D_INODE, "inode: %ld, and flag %x, or flag %x, dd_flags %x\n",
314 dentry->d_inode->i_ino, and_flag, or_flag,
315 presto_d2d(dentry)->dd_flags);
316
317 presto_d2d(dentry)->dd_flags &= and_flag;
318 presto_d2d(dentry)->dd_flags |= or_flag;
319 if (res)
320 *res = presto_d2d(dentry)->dd_flags;
321
322 return error;
323 }
324
325 /* given a path, operate on the flags in its cache. Used by mark_ioctl */
izo_mark_cache(struct dentry * dentry,int and_flag,int or_flag,int * res)326 int izo_mark_cache(struct dentry *dentry, int and_flag, int or_flag,
327 int *res)
328 {
329 struct presto_cache *cache;
330
331 if (presto_d2d(dentry) == NULL) {
332 CERROR("InterMezzo: no ddata for inode %ld in %s\n",
333 dentry->d_inode->i_ino, __FUNCTION__);
334 return -EINVAL;
335 }
336
337 CDEBUG(D_INODE, "inode: %ld, and flag %x, or flag %x, dd_flags %x\n",
338 dentry->d_inode->i_ino, and_flag, or_flag,
339 presto_d2d(dentry)->dd_flags);
340
341 cache = presto_get_cache(dentry->d_inode);
342 if ( !cache ) {
343 CERROR("PRESTO: BAD: cannot find cache in izo_mark_cache\n");
344 return -EBADF;
345 }
346
347 cache->cache_flags &= and_flag;
348 cache->cache_flags |= or_flag;
349 if (res)
350 *res = (int)cache->cache_flags;
351
352 return 0;
353 }
354
presto_set_max_kml_size(const char * path,unsigned long max_size)355 int presto_set_max_kml_size(const char *path, unsigned long max_size)
356 {
357 struct presto_file_set *fset;
358
359 ENTRY;
360
361 fset = presto_path2fileset(path);
362 if (IS_ERR(fset)) {
363 EXIT;
364 return PTR_ERR(fset);
365 }
366
367 fset->kml_truncate_size = max_size;
368 CDEBUG(D_CACHE, "KML truncate size set to %lu bytes for fset %s.\n",
369 max_size, path);
370
371 EXIT;
372 return 0;
373 }
374
izo_mark_fset(struct dentry * dentry,int and_flag,int or_flag,int * res)375 int izo_mark_fset(struct dentry *dentry, int and_flag, int or_flag,
376 int * res)
377 {
378 struct presto_file_set *fset;
379
380 fset = presto_fset(dentry);
381 if ( !fset ) {
382 CERROR("PRESTO: BAD: cannot find cache in izo_mark_cache\n");
383 make_bad_inode(dentry->d_inode);
384 return -EBADF;
385 }
386 fset->fset_flags &= and_flag;
387 fset->fset_flags |= or_flag;
388 if (res)
389 *res = (int)fset->fset_flags;
390
391 return 0;
392 }
393
394 /* talk to Lento about the permit */
presto_permit_upcall(struct dentry * dentry)395 static int presto_permit_upcall(struct dentry *dentry)
396 {
397 int rc;
398 char *path, *buffer;
399 int pathlen;
400 int minor;
401 int fsetnamelen;
402 struct presto_file_set *fset = NULL;
403
404 ENTRY;
405
406 if ( (minor = presto_i2m(dentry->d_inode)) < 0) {
407 EXIT;
408 return -EINVAL;
409 }
410
411 fset = presto_fset(dentry);
412 if (!fset) {
413 EXIT;
414 return -ENOTCONN;
415 }
416
417 if ( !presto_lento_up(minor) ) {
418 if ( fset->fset_flags & FSET_STEAL_PERMIT ) {
419 EXIT;
420 return 0;
421 } else {
422 EXIT;
423 return -ENOTCONN;
424 }
425 }
426
427 PRESTO_ALLOC(buffer, PAGE_SIZE);
428 if ( !buffer ) {
429 CERROR("PRESTO: out of memory!\n");
430 EXIT;
431 return -ENOMEM;
432 }
433 path = presto_path(dentry, fset->fset_dentry, buffer, PAGE_SIZE);
434 pathlen = MYPATHLEN(buffer, path);
435 fsetnamelen = strlen(fset->fset_name);
436 rc = izo_upc_permit(minor, dentry, pathlen, path, fset->fset_name);
437 PRESTO_FREE(buffer, PAGE_SIZE);
438 EXIT;
439 return rc;
440 }
441
442 /* get a write permit for the fileset of this inode
443 * - if this returns a negative value there was an error
444 * - if 0 is returned the permit was already in the kernel -- or --
445 * Lento gave us the permit without reintegration
446 * - lento returns the number of records it reintegrated
447 *
448 * Note that if this fileset has branches, a permit will -never- to a normal
449 * process for writing in the data area (ie, outside of .intermezzo)
450 */
presto_get_permit(struct inode * inode)451 int presto_get_permit(struct inode * inode)
452 {
453 struct dentry *de;
454 struct presto_file_set *fset;
455 int minor = presto_i2m(inode);
456 int rc = 0;
457
458 ENTRY;
459 if (minor < 0) {
460 EXIT;
461 return -1;
462 }
463
464 if ( ISLENTO(minor) ) {
465 EXIT;
466 return 0;
467 }
468
469 if (list_empty(&inode->i_dentry)) {
470 CERROR("No alias for inode %d\n", (int) inode->i_ino);
471 EXIT;
472 return -EINVAL;
473 }
474
475 de = list_entry(inode->i_dentry.next, struct dentry, d_alias);
476
477 if (presto_chk(de, PRESTO_DONT_JOURNAL)) {
478 EXIT;
479 return 0;
480 }
481
482 fset = presto_fset(de);
483 if ( !fset ) {
484 CERROR("Presto: no fileset in presto_get_permit!\n");
485 EXIT;
486 return -EINVAL;
487 }
488
489 if (fset->fset_flags & FSET_HAS_BRANCHES) {
490 EXIT;
491 return -EROFS;
492 }
493
494 spin_lock(&fset->fset_permit_lock);
495 if (fset->fset_flags & FSET_HASPERMIT) {
496 fset->fset_permit_count++;
497 CDEBUG(D_INODE, "permit count now %d, inode %lx\n",
498 fset->fset_permit_count, inode->i_ino);
499 spin_unlock(&fset->fset_permit_lock);
500 EXIT;
501 return 0;
502 }
503
504 /* Allow reintegration to proceed without locks -SHP */
505 fset->fset_permit_upcall_count++;
506 if (fset->fset_permit_upcall_count == 1) {
507 spin_unlock(&fset->fset_permit_lock);
508 rc = presto_permit_upcall(fset->fset_dentry);
509 spin_lock(&fset->fset_permit_lock);
510 fset->fset_permit_upcall_count--;
511 if (rc == 0) {
512 izo_mark_fset(fset->fset_dentry, ~0, FSET_HASPERMIT,
513 NULL);
514 fset->fset_permit_count++;
515 } else if (rc == ENOTCONN) {
516 CERROR("InterMezzo: disconnected operation. stealing permit.\n");
517 izo_mark_fset(fset->fset_dentry, ~0, FSET_HASPERMIT,
518 NULL);
519 fset->fset_permit_count++;
520 /* set a disconnected flag here to stop upcalls */
521 rc = 0;
522 } else {
523 CERROR("InterMezzo: presto_permit_upcall failed: %d\n", rc);
524 rc = -EROFS;
525 /* go to sleep here and try again? */
526 }
527 wake_up_interruptible(&fset->fset_permit_queue);
528 } else {
529 /* Someone is already doing an upcall; go to sleep. */
530 DECLARE_WAITQUEUE(wait, current);
531
532 spin_unlock(&fset->fset_permit_lock);
533 add_wait_queue(&fset->fset_permit_queue, &wait);
534 while (1) {
535 set_current_state(TASK_INTERRUPTIBLE);
536
537 spin_lock(&fset->fset_permit_lock);
538 if (fset->fset_permit_upcall_count == 0)
539 break;
540 spin_unlock(&fset->fset_permit_lock);
541
542 if (signal_pending(current)) {
543 remove_wait_queue(&fset->fset_permit_queue,
544 &wait);
545 return -ERESTARTSYS;
546 }
547 schedule();
548 }
549 remove_wait_queue(&fset->fset_permit_queue, &wait);
550 /* We've been woken up: do we have the permit? */
551 if (fset->fset_flags & FSET_HASPERMIT)
552 /* FIXME: Is this the right thing? */
553 rc = -EAGAIN;
554 }
555
556 CDEBUG(D_INODE, "permit count now %d, ino %ld (likely 1), "
557 "rc %d\n", fset->fset_permit_count, inode->i_ino, rc);
558 spin_unlock(&fset->fset_permit_lock);
559 EXIT;
560 return rc;
561 }
562
presto_put_permit(struct inode * inode)563 int presto_put_permit(struct inode * inode)
564 {
565 struct dentry *de;
566 struct presto_file_set *fset;
567 int minor = presto_i2m(inode);
568
569 ENTRY;
570 if (minor < 0) {
571 EXIT;
572 return -1;
573 }
574
575 if ( ISLENTO(minor) ) {
576 EXIT;
577 return 0;
578 }
579
580 if (list_empty(&inode->i_dentry)) {
581 CERROR("No alias for inode %d\n", (int) inode->i_ino);
582 EXIT;
583 return -1;
584 }
585
586 de = list_entry(inode->i_dentry.next, struct dentry, d_alias);
587
588 fset = presto_fset(de);
589 if ( !fset ) {
590 CERROR("InterMezzo: no fileset in %s!\n", __FUNCTION__);
591 EXIT;
592 return -1;
593 }
594
595 if (presto_chk(de, PRESTO_DONT_JOURNAL)) {
596 EXIT;
597 return 0;
598 }
599
600 spin_lock(&fset->fset_permit_lock);
601 if (fset->fset_flags & FSET_HASPERMIT) {
602 if (fset->fset_permit_count > 0)
603 fset->fset_permit_count--;
604 else
605 CERROR("Put permit while permit count is 0, "
606 "inode %ld!\n", inode->i_ino);
607 } else {
608 fset->fset_permit_count = 0;
609 CERROR("InterMezzo: put permit while no permit, inode %ld, "
610 "flags %x!\n", inode->i_ino, fset->fset_flags);
611 }
612
613 CDEBUG(D_INODE, "permit count now %d, inode %ld\n",
614 fset->fset_permit_count, inode->i_ino);
615
616 if (fset->fset_flags & FSET_PERMIT_WAITING &&
617 fset->fset_permit_count == 0) {
618 CDEBUG(D_INODE, "permit count now 0, ino %ld, wake sleepers\n",
619 inode->i_ino);
620 wake_up_interruptible(&fset->fset_permit_queue);
621 }
622 spin_unlock(&fset->fset_permit_lock);
623
624 EXIT;
625 return 0;
626 }
627
presto_getversion(struct presto_version * presto_version,struct inode * inode)628 void presto_getversion(struct presto_version * presto_version,
629 struct inode * inode)
630 {
631 presto_version->pv_mtime = (__u64)inode->i_mtime;
632 presto_version->pv_ctime = (__u64)inode->i_ctime;
633 presto_version->pv_size = (__u64)inode->i_size;
634 }
635
636
637 /* If uuid is non-null, it is the uuid of the peer that's making the revocation
638 * request. If it is null, this request was made locally, without external
639 * pressure to give up the permit. This most often occurs when a client
640 * starts up.
641 *
642 * FIXME: this function needs to be refactored slightly once we start handling
643 * multiple clients.
644 */
izo_revoke_permit(struct dentry * dentry,__u8 uuid[16])645 int izo_revoke_permit(struct dentry *dentry, __u8 uuid[16])
646 {
647 struct presto_file_set *fset;
648 DECLARE_WAITQUEUE(wait, current);
649 int minor, rc;
650
651 ENTRY;
652
653 minor = presto_i2m(dentry->d_inode);
654 if (minor < 0) {
655 EXIT;
656 return -ENODEV;
657 }
658
659 fset = presto_fset(dentry);
660 if (fset == NULL) {
661 EXIT;
662 return -ENODEV;
663 }
664
665 spin_lock(&fset->fset_permit_lock);
666 if (fset->fset_flags & FSET_PERMIT_WAITING) {
667 CERROR("InterMezzo: Two processes are waiting on the same permit--this not yet supported! Aborting this particular permit request...\n");
668 EXIT;
669 spin_unlock(&fset->fset_permit_lock);
670 return -EINVAL;
671 }
672
673 if (fset->fset_permit_count == 0)
674 goto got_permit;
675
676 /* Something is still using this permit. Mark that we're waiting for it
677 * and go to sleep. */
678 rc = izo_mark_fset(dentry, ~0, FSET_PERMIT_WAITING, NULL);
679 spin_unlock(&fset->fset_permit_lock);
680 if (rc < 0) {
681 EXIT;
682 return rc;
683 }
684
685 add_wait_queue(&fset->fset_permit_queue, &wait);
686 while (1) {
687 set_current_state(TASK_INTERRUPTIBLE);
688
689 spin_lock(&fset->fset_permit_lock);
690 if (fset->fset_permit_count == 0)
691 break;
692 spin_unlock(&fset->fset_permit_lock);
693
694 if (signal_pending(current)) {
695 /* FIXME: there must be a better thing to return... */
696 remove_wait_queue(&fset->fset_permit_queue, &wait);
697 EXIT;
698 return -ERESTARTSYS;
699 }
700
701 /* FIXME: maybe there should be a timeout here. */
702
703 schedule();
704 }
705
706 remove_wait_queue(&fset->fset_permit_queue, &wait);
707 got_permit:
708 /* By this point fset->fset_permit_count is zero and we're holding the
709 * lock. */
710 CDEBUG(D_CACHE, "InterMezzo: releasing permit inode %ld\n",
711 dentry->d_inode->i_ino);
712
713 if (uuid != NULL) {
714 rc = izo_upc_revoke_permit(minor, fset->fset_name, uuid);
715 if (rc < 0) {
716 spin_unlock(&fset->fset_permit_lock);
717 EXIT;
718 return rc;
719 }
720 }
721
722 izo_mark_fset(fset->fset_dentry, ~FSET_PERMIT_WAITING, 0, NULL);
723 izo_mark_fset(fset->fset_dentry, ~FSET_HASPERMIT, 0, NULL);
724 spin_unlock(&fset->fset_permit_lock);
725 EXIT;
726 return 0;
727 }
728
presto_is_read_only(struct presto_file_set * fset)729 inline int presto_is_read_only(struct presto_file_set * fset)
730 {
731 int minor, mask;
732 struct presto_cache *cache = fset->fset_cache;
733
734 minor= cache->cache_psdev->uc_minor;
735 mask= (ISLENTO(minor)? FSET_LENTO_RO : FSET_CLIENT_RO);
736 if ( fset->fset_flags & mask )
737 return 1;
738 mask= (ISLENTO(minor)? CACHE_LENTO_RO : CACHE_CLIENT_RO);
739 return ((cache->cache_flags & mask)? 1 : 0);
740 }
741