glusterfs

1
/*
2
  Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
3
  This file is part of GlusterFS.
4

5
  This file is licensed to you under your choice of the GNU Lesser
6
  General Public License, version 3 or any later version (LGPLv3 or
7
  later), or the GNU General Public License, version 2 (GPLv2), in all
8
  cases as published by the Free Software Foundation.
9
*/
10

11
/* Perform fop on all subvolumes represented by list[] array and wait
12
   for all callbacks to return */
13

14
/* NOTE: Cluster-syncop, like syncop blocks the executing thread until the
15
 * responses are gathered if it is not executed as part of synctask. So it
16
 * shouldn't be invoked in epoll worker thread */
17
#include "glusterfs/cluster-syncop.h"
18

19
#define FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fop,       \
20
                   args...)                                                    \
21
    do {                                                                       \
22
        int __i = 0;                                                           \
23
        int __count = 0;                                                       \
24
        cluster_local_t __local = {                                            \
25
            0,                                                                 \
26
        };                                                                     \
27
        void *__old_local = frame->local;                                      \
28
                                                                               \
29
        __local.replies = replies;                                             \
30
        memset(output, 0, numsubvols);                                         \
31
        cluster_replies_wipe(replies, numsubvols);                             \
32
        for (__i = 0; __i < numsubvols; __i++)                                 \
33
            INIT_LIST_HEAD(&replies[__i].entries.list);                        \
34
        if (syncbarrier_init(&__local.barrier))                                \
35
            break;                                                             \
36
        frame->local = &__local;                                               \
37
        for (__i = 0; __i < numsubvols; __i++) {                               \
38
            if (on[__i]) {                                                     \
39
                __count++;                                                     \
40
            }                                                                  \
41
        }                                                                      \
42
        __local.barrier.waitfor = __count;                                     \
43
        for (__i = 0; __i < numsubvols; __i++) {                               \
44
            if (!on[__i])                                                      \
45
                continue;                                                      \
46
            STACK_WIND_COOKIE(frame, cluster_##fop##_cbk, (void *)(long)__i,   \
47
                              subvols[__i], subvols[__i]->fops->fop, args);    \
48
        }                                                                      \
49
        syncbarrier_wait(&__local.barrier, __count);                           \
50
        syncbarrier_destroy(&__local.barrier);                                 \
51
        frame->local = __old_local;                                            \
52
        STACK_RESET(frame->root);                                              \
53
    } while (0)
54

55
#define FOP_SEQ(subvols, on, numsubvols, replies, output, frame, fop, args...) \
56
    do {                                                                       \
57
        int __i = 0;                                                           \
58
                                                                               \
59
        cluster_local_t __local = {                                            \
60
            0,                                                                 \
61
        };                                                                     \
62
        void *__old_local = frame->local;                                      \
63
        __local.replies = replies;                                             \
64
        memset(output, 0, numsubvols);                                         \
65
        cluster_replies_wipe(replies, numsubvols);                             \
66
        for (__i = 0; __i < numsubvols; __i++)                                 \
67
            INIT_LIST_HEAD(&replies[__i].entries.list);                        \
68
        if (syncbarrier_init(&__local.barrier))                                \
69
            break;                                                             \
70
        frame->local = &__local;                                               \
71
        for (__i = 0; __i < numsubvols; __i++) {                               \
72
            if (!on[__i])                                                      \
73
                continue;                                                      \
74
            STACK_WIND_COOKIE(frame, cluster_##fop##_cbk, (void *)(long)__i,   \
75
                              subvols[__i], subvols[__i]->fops->fop, args);    \
76
            syncbarrier_wait(&__local.barrier, 1);                             \
77
        }                                                                      \
78
        syncbarrier_destroy(&__local.barrier);                                 \
79
        frame->local = __old_local;                                            \
80
        STACK_RESET(frame->root);                                              \
81
    } while (0)
82

83
#define FOP_CBK(fop, frame, cookie, args...)                                   \
84
    do {                                                                       \
85
        cluster_local_t *__local = frame->local;                               \
86
        int __i = (long)cookie;                                                \
87
        args_##fop##_cbk_store(&__local->replies[__i], args);                  \
88
        __local->replies[__i].valid = 1;                                       \
89
        syncbarrier_wake(&__local->barrier);                                   \
90
    } while (0)
91

92
int32_t
93
cluster_fop_success_fill(default_args_cbk_t *replies, int numsubvols,
94
                         unsigned char *success)
95
{
96
    int i = 0;
97
    int count = 0;
98

99
    for (i = 0; i < numsubvols; i++) {
100
        if (replies[i].valid && replies[i].op_ret >= 0) {
101
            success[i] = 1;
102
            count++;
103
        } else {
104
            success[i] = 0;
105
        }
106
    }
107

108
    return count;
109
}
110

111
void
112
cluster_replies_wipe(default_args_cbk_t *replies, int numsubvols)
113
{
114
    int i = 0;
115

116
    if (!replies)
117
        return;
118

119
    for (i = 0; i < numsubvols; i++)
120
        args_cbk_wipe(&replies[i]);
121
    memset(replies, 0, numsubvols * sizeof(*replies));
122
}
123

124
int32_t
125
cluster_lookup_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
126
                   int32_t op_ret, int32_t op_errno, inode_t *inode,
127
                   struct iatt *buf, dict_t *xdata, struct iatt *postparent)
128
{
129
    FOP_CBK(lookup, frame, cookie, op_ret, op_errno, inode, buf, xdata,
130
            postparent);
131
    return 0;
132
}
133

134
int32_t
135
cluster_stat_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
136
                 int32_t op_ret, int32_t op_errno, struct iatt *buf,
137
                 dict_t *xdata)
138
{
139
    FOP_CBK(stat, frame, cookie, op_ret, op_errno, buf, xdata);
140
    return 0;
141
}
142

143
int32_t
144
cluster_truncate_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
145
                     int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
146
                     struct iatt *postbuf, dict_t *xdata)
147
{
148
    FOP_CBK(truncate, frame, cookie, op_ret, op_errno, prebuf, postbuf, xdata);
149
    return 0;
150
}
151

152
int32_t
153
cluster_ftruncate_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
154
                      int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
155
                      struct iatt *postbuf, dict_t *xdata)
156
{
157
    FOP_CBK(ftruncate, frame, cookie, op_ret, op_errno, prebuf, postbuf, xdata);
158
    return 0;
159
}
160

161
int32_t
162
cluster_access_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
163
                   int32_t op_ret, int32_t op_errno, dict_t *xdata)
164
{
165
    FOP_CBK(access, frame, cookie, op_ret, op_errno, xdata);
166
    return 0;
167
}
168

169
int32_t
170
cluster_readlink_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
171
                     int32_t op_ret, int32_t op_errno, const char *path,
172
                     struct iatt *buf, dict_t *xdata)
173
{
174
    FOP_CBK(readlink, frame, cookie, op_ret, op_errno, path, buf, xdata);
175
    return 0;
176
}
177

178
int32_t
179
cluster_mknod_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
180
                  int32_t op_ret, int32_t op_errno, inode_t *inode,
181
                  struct iatt *buf, struct iatt *preparent,
182
                  struct iatt *postparent, dict_t *xdata)
183
{
184
    FOP_CBK(mknod, frame, cookie, op_ret, op_errno, inode, buf, preparent,
185
            postparent, xdata);
186
    return 0;
187
}
188

189
int32_t
190
cluster_mkdir_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
191
                  int32_t op_ret, int32_t op_errno, inode_t *inode,
192
                  struct iatt *buf, struct iatt *preparent,
193
                  struct iatt *postparent, dict_t *xdata)
194
{
195
    FOP_CBK(mkdir, frame, cookie, op_ret, op_errno, inode, buf, preparent,
196
            postparent, xdata);
197
    return 0;
198
}
199

200
int32_t
201
cluster_unlink_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
202
                   int32_t op_ret, int32_t op_errno, struct iatt *preparent,
203
                   struct iatt *postparent, dict_t *xdata)
204
{
205
    FOP_CBK(unlink, frame, cookie, op_ret, op_errno, preparent, postparent,
206
            xdata);
207
    return 0;
208
}
209

210
int32_t
211
cluster_rmdir_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
212
                  int32_t op_ret, int32_t op_errno, struct iatt *preparent,
213
                  struct iatt *postparent, dict_t *xdata)
214
{
215
    FOP_CBK(rmdir, frame, cookie, op_ret, op_errno, preparent, postparent,
216
            xdata);
217
    return 0;
218
}
219

220
int32_t
221
cluster_symlink_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
222
                    int32_t op_ret, int32_t op_errno, inode_t *inode,
223
                    struct iatt *buf, struct iatt *preparent,
224
                    struct iatt *postparent, dict_t *xdata)
225
{
226
    FOP_CBK(symlink, frame, cookie, op_ret, op_errno, inode, buf, preparent,
227
            postparent, xdata);
228
    return 0;
229
}
230

231
int32_t
232
cluster_rename_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
233
                   int32_t op_ret, int32_t op_errno, struct iatt *buf,
234
                   struct iatt *preoldparent, struct iatt *postoldparent,
235
                   struct iatt *prenewparent, struct iatt *postnewparent,
236
                   dict_t *xdata)
237
{
238
    FOP_CBK(rename, frame, cookie, op_ret, op_errno, buf, preoldparent,
239
            postoldparent, prenewparent, postnewparent, xdata);
240
    return 0;
241
}
242

243
int32_t
244
cluster_link_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
245
                 int32_t op_ret, int32_t op_errno, inode_t *inode,
246
                 struct iatt *buf, struct iatt *preparent,
247
                 struct iatt *postparent, dict_t *xdata)
248
{
249
    FOP_CBK(link, frame, cookie, op_ret, op_errno, inode, buf, preparent,
250
            postparent, xdata);
251
    return 0;
252
}
253

254
int32_t
255
cluster_create_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
256
                   int32_t op_ret, int32_t op_errno, fd_t *fd, inode_t *inode,
257
                   struct iatt *buf, struct iatt *preparent,
258
                   struct iatt *postparent, dict_t *xdata)
259
{
260
    FOP_CBK(create, frame, cookie, op_ret, op_errno, fd, inode, buf, preparent,
261
            postparent, xdata);
262
    return 0;
263
}
264

265
int32_t
266
cluster_open_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
267
                 int32_t op_ret, int32_t op_errno, fd_t *fd, dict_t *xdata)
268
{
269
    FOP_CBK(open, frame, cookie, op_ret, op_errno, fd, xdata);
270
    return 0;
271
}
272

273
int32_t
274
cluster_readv_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
275
                  int32_t op_ret, int32_t op_errno, struct iovec *vector,
276
                  int32_t count, struct iatt *stbuf, struct iobref *iobref,
277
                  dict_t *xdata)
278
{
279
    FOP_CBK(readv, frame, cookie, op_ret, op_errno, vector, count, stbuf,
280
            iobref, xdata);
281
    return 0;
282
}
283

284
int32_t
285
cluster_writev_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
286
                   int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
287
                   struct iatt *postbuf, dict_t *xdata)
288
{
289
    FOP_CBK(writev, frame, cookie, op_ret, op_errno, prebuf, postbuf, xdata);
290
    return 0;
291
}
292

293
int32_t
294
cluster_put_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
295
                int32_t op_ret, int32_t op_errno, inode_t *inode,
296
                struct iatt *buf, struct iatt *preparent,
297
                struct iatt *postparent, dict_t *xdata)
298
{
299
    FOP_CBK(put, frame, cookie, op_ret, op_errno, inode, buf, preparent,
300
            postparent, xdata);
301
    return 0;
302
}
303

304
int32_t
305
cluster_flush_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
306
                  int32_t op_ret, int32_t op_errno, dict_t *xdata)
307
{
308
    FOP_CBK(flush, frame, cookie, op_ret, op_errno, xdata);
309
    return 0;
310
}
311

312
int32_t
313
cluster_fsync_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
314
                  int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
315
                  struct iatt *postbuf, dict_t *xdata)
316
{
317
    FOP_CBK(fsync, frame, cookie, op_ret, op_errno, prebuf, postbuf, xdata);
318
    return 0;
319
}
320

321
int32_t
322
cluster_fstat_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
323
                  int32_t op_ret, int32_t op_errno, struct iatt *buf,
324
                  dict_t *xdata)
325
{
326
    FOP_CBK(fstat, frame, cookie, op_ret, op_errno, buf, xdata);
327
    return 0;
328
}
329

330
int32_t
331
cluster_opendir_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
332
                    int32_t op_ret, int32_t op_errno, fd_t *fd, dict_t *xdata)
333
{
334
    FOP_CBK(opendir, frame, cookie, op_ret, op_errno, fd, xdata);
335
    return 0;
336
}
337

338
int32_t
339
cluster_fsyncdir_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
340
                     int32_t op_ret, int32_t op_errno, dict_t *xdata)
341
{
342
    FOP_CBK(fsyncdir, frame, cookie, op_ret, op_errno, xdata);
343
    return 0;
344
}
345

346
int32_t
347
cluster_statfs_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
348
                   int32_t op_ret, int32_t op_errno, struct statvfs *buf,
349
                   dict_t *xdata)
350
{
351
    FOP_CBK(statfs, frame, cookie, op_ret, op_errno, buf, xdata);
352
    return 0;
353
}
354

355
int32_t
356
cluster_setxattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
357
                     int32_t op_ret, int32_t op_errno, dict_t *xdata)
358
{
359
    FOP_CBK(setxattr, frame, cookie, op_ret, op_errno, xdata);
360
    return 0;
361
}
362

363
int32_t
364
cluster_fsetxattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
365
                      int32_t op_ret, int32_t op_errno, dict_t *xdata)
366
{
367
    FOP_CBK(fsetxattr, frame, cookie, op_ret, op_errno, xdata);
368
    return 0;
369
}
370

371
int32_t
372
cluster_fgetxattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
373
                      int32_t op_ret, int32_t op_errno, dict_t *dict,
374
                      dict_t *xdata)
375
{
376
    FOP_CBK(fgetxattr, frame, cookie, op_ret, op_errno, dict, xdata);
377
    return 0;
378
}
379

380
int32_t
381
cluster_getxattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
382
                     int32_t op_ret, int32_t op_errno, dict_t *dict,
383
                     dict_t *xdata)
384
{
385
    FOP_CBK(getxattr, frame, cookie, op_ret, op_errno, dict, xdata);
386
    return 0;
387
}
388

389
int32_t
390
cluster_xattrop_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
391
                    int32_t op_ret, int32_t op_errno, dict_t *dict,
392
                    dict_t *xdata)
393
{
394
    FOP_CBK(xattrop, frame, cookie, op_ret, op_errno, dict, xdata);
395
    return 0;
396
}
397

398
int32_t
399
cluster_fxattrop_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
400
                     int32_t op_ret, int32_t op_errno, dict_t *dict,
401
                     dict_t *xdata)
402
{
403
    FOP_CBK(fxattrop, frame, cookie, op_ret, op_errno, dict, xdata);
404
    return 0;
405
}
406

407
int32_t
408
cluster_removexattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
409
                        int32_t op_ret, int32_t op_errno, dict_t *xdata)
410
{
411
    FOP_CBK(removexattr, frame, cookie, op_ret, op_errno, xdata);
412
    return 0;
413
}
414

415
int32_t
416
cluster_fremovexattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
417
                         int32_t op_ret, int32_t op_errno, dict_t *xdata)
418
{
419
    FOP_CBK(fremovexattr, frame, cookie, op_ret, op_errno, xdata);
420
    return 0;
421
}
422

423
int32_t
424
cluster_lk_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
425
               int32_t op_ret, int32_t op_errno, struct gf_flock *lock,
426
               dict_t *xdata)
427
{
428
    FOP_CBK(lk, frame, cookie, op_ret, op_errno, lock, xdata);
429
    return 0;
430
}
431

432
int32_t
433
cluster_inodelk_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
434
                    int32_t op_ret, int32_t op_errno, dict_t *xdata)
435
{
436
    FOP_CBK(inodelk, frame, cookie, op_ret, op_errno, xdata);
437
    return 0;
438
}
439

440
int32_t
441
cluster_finodelk_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
442
                     int32_t op_ret, int32_t op_errno, dict_t *xdata)
443
{
444
    FOP_CBK(finodelk, frame, cookie, op_ret, op_errno, xdata);
445
    return 0;
446
}
447

448
int32_t
449
cluster_entrylk_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
450
                    int32_t op_ret, int32_t op_errno, dict_t *xdata)
451
{
452
    FOP_CBK(entrylk, frame, cookie, op_ret, op_errno, xdata);
453
    return 0;
454
}
455

456
int32_t
457
cluster_fentrylk_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
458
                     int32_t op_ret, int32_t op_errno, dict_t *xdata)
459
{
460
    FOP_CBK(fentrylk, frame, cookie, op_ret, op_errno, xdata);
461
    return 0;
462
}
463

464
int32_t
465
cluster_rchecksum_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
466
                      int32_t op_ret, int32_t op_errno, uint32_t weak_checksum,
467
                      uint8_t *strong_checksum, dict_t *xdata)
468
{
469
    FOP_CBK(rchecksum, frame, cookie, op_ret, op_errno, weak_checksum,
470
            strong_checksum, xdata);
471
    return 0;
472
}
473

474
int32_t
475
cluster_readdir_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
476
                    int32_t op_ret, int32_t op_errno, gf_dirent_t *entries,
477
                    dict_t *xdata)
478
{
479
    FOP_CBK(readdir, frame, cookie, op_ret, op_errno, entries, xdata);
480
    return 0;
481
}
482

483
int32_t
484
cluster_readdirp_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
485
                     int32_t op_ret, int32_t op_errno, gf_dirent_t *entries,
486
                     dict_t *xdata)
487
{
488
    FOP_CBK(readdirp, frame, cookie, op_ret, op_errno, entries, xdata);
489
    return 0;
490
}
491

492
int32_t
493
cluster_setattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
494
                    int32_t op_ret, int32_t op_errno, struct iatt *statpre,
495
                    struct iatt *statpost, dict_t *xdata)
496
{
497
    FOP_CBK(setattr, frame, cookie, op_ret, op_errno, statpre, statpost, xdata);
498
    return 0;
499
}
500

501
int32_t
502
cluster_fsetattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
503
                     int32_t op_ret, int32_t op_errno, struct iatt *statpre,
504
                     struct iatt *statpost, dict_t *xdata)
505
{
506
    FOP_CBK(fsetattr, frame, cookie, op_ret, op_errno, statpre, statpost,
507
            xdata);
508
    return 0;
509
}
510

511
int32_t
512
cluster_fallocate_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
513
                      int32_t op_ret, int32_t op_errno, struct iatt *pre,
514
                      struct iatt *post, dict_t *xdata)
515
{
516
    FOP_CBK(fallocate, frame, cookie, op_ret, op_errno, pre, post, xdata);
517
    return 0;
518
}
519

520
int32_t
521
cluster_discard_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
522
                    int32_t op_ret, int32_t op_errno, struct iatt *pre,
523
                    struct iatt *post, dict_t *xdata)
524
{
525
    FOP_CBK(discard, frame, cookie, op_ret, op_errno, pre, post, xdata);
526
    return 0;
527
}
528

529
int32_t
530
cluster_zerofill_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
531
                     int32_t op_ret, int32_t op_errno, struct iatt *pre,
532
                     struct iatt *post, dict_t *xdata)
533
{
534
    FOP_CBK(zerofill, frame, cookie, op_ret, op_errno, pre, post, xdata);
535
    return 0;
536
}
537

538
int32_t
539
cluster_ipc_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
540
                int32_t op_ret, int32_t op_errno, dict_t *xdata)
541
{
542
    FOP_CBK(ipc, frame, cookie, op_ret, op_errno, xdata);
543
    return 0;
544
}
545

546
int32_t
547
cluster_fgetxattr(xlator_t **subvols, unsigned char *on, int numsubvols,
548
                  default_args_cbk_t *replies, unsigned char *output,
549
                  call_frame_t *frame, xlator_t *this, fd_t *fd,
550
                  const char *name, dict_t *xdata)
551
{
552
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fgetxattr, fd,
553
               name, xdata);
554
    return cluster_fop_success_fill(replies, numsubvols, output);
555
}
556

557
int32_t
558
cluster_fsetxattr(xlator_t **subvols, unsigned char *on, int numsubvols,
559
                  default_args_cbk_t *replies, unsigned char *output,
560
                  call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
561
                  int32_t flags, dict_t *xdata)
562
{
563
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fsetxattr, fd,
564
               dict, flags, xdata);
565
    return cluster_fop_success_fill(replies, numsubvols, output);
566
}
567

568
int32_t
569
cluster_setxattr(xlator_t **subvols, unsigned char *on, int numsubvols,
570
                 default_args_cbk_t *replies, unsigned char *output,
571
                 call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
572
                 int32_t flags, dict_t *xdata)
573
{
574
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, setxattr, loc,
575
               dict, flags, xdata);
576
    return cluster_fop_success_fill(replies, numsubvols, output);
577
}
578

579
int32_t
580
cluster_statfs(xlator_t **subvols, unsigned char *on, int numsubvols,
581
               default_args_cbk_t *replies, unsigned char *output,
582
               call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
583
{
584
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, statfs, loc,
585
               xdata);
586
    return cluster_fop_success_fill(replies, numsubvols, output);
587
}
588

589
int32_t
590
cluster_fsyncdir(xlator_t **subvols, unsigned char *on, int numsubvols,
591
                 default_args_cbk_t *replies, unsigned char *output,
592
                 call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
593
                 dict_t *xdata)
594
{
595
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fsyncdir, fd,
596
               flags, xdata);
597
    return cluster_fop_success_fill(replies, numsubvols, output);
598
}
599

600
int32_t
601
cluster_opendir(xlator_t **subvols, unsigned char *on, int numsubvols,
602
                default_args_cbk_t *replies, unsigned char *output,
603
                call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd,
604
                dict_t *xdata)
605
{
606
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, opendir, loc,
607
               fd, xdata);
608
    return cluster_fop_success_fill(replies, numsubvols, output);
609
}
610

611
int32_t
612
cluster_fstat(xlator_t **subvols, unsigned char *on, int numsubvols,
613
              default_args_cbk_t *replies, unsigned char *output,
614
              call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
615
{
616
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fstat, fd,
617
               xdata);
618
    return cluster_fop_success_fill(replies, numsubvols, output);
619
}
620

621
int32_t
622
cluster_fsync(xlator_t **subvols, unsigned char *on, int numsubvols,
623
              default_args_cbk_t *replies, unsigned char *output,
624
              call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
625
              dict_t *xdata)
626
{
627
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fsync, fd,
628
               flags, xdata);
629
    return cluster_fop_success_fill(replies, numsubvols, output);
630
}
631

632
int32_t
633
cluster_flush(xlator_t **subvols, unsigned char *on, int numsubvols,
634
              default_args_cbk_t *replies, unsigned char *output,
635
              call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
636
{
637
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, flush, fd,
638
               xdata);
639
    return cluster_fop_success_fill(replies, numsubvols, output);
640
}
641

642
int32_t
643
cluster_writev(xlator_t **subvols, unsigned char *on, int numsubvols,
644
               default_args_cbk_t *replies, unsigned char *output,
645
               call_frame_t *frame, xlator_t *this, fd_t *fd,
646
               struct iovec *vector, int32_t count, off_t off, uint32_t flags,
647
               struct iobref *iobref, dict_t *xdata)
648
{
649
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, writev, fd,
650
               vector, count, off, flags, iobref, xdata);
651
    return cluster_fop_success_fill(replies, numsubvols, output);
652
}
653

654
int32_t
655
cluster_put(xlator_t **subvols, unsigned char *on, int numsubvols,
656
            default_args_cbk_t *replies, unsigned char *output,
657
            call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
658
            mode_t umask, uint32_t flags, struct iovec *vector, int32_t count,
659
            off_t offset, struct iobref *iobref, dict_t *xattr, dict_t *xdata)
660
{
661
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, put, loc, mode,
662
               umask, flags, vector, count, offset, iobref, xattr, xdata);
663
    return cluster_fop_success_fill(replies, numsubvols, output);
664
}
665

666
int32_t
667
cluster_readv(xlator_t **subvols, unsigned char *on, int numsubvols,
668
              default_args_cbk_t *replies, unsigned char *output,
669
              call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
670
              off_t offset, uint32_t flags, dict_t *xdata)
671
{
672
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, readv, fd, size,
673
               offset, flags, xdata);
674
    return cluster_fop_success_fill(replies, numsubvols, output);
675
}
676

677
int32_t
678
cluster_open(xlator_t **subvols, unsigned char *on, int numsubvols,
679
             default_args_cbk_t *replies, unsigned char *output,
680
             call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
681
             fd_t *fd, dict_t *xdata)
682
{
683
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, open, loc,
684
               flags, fd, xdata);
685
    return cluster_fop_success_fill(replies, numsubvols, output);
686
}
687

688
int32_t
689
cluster_create(xlator_t **subvols, unsigned char *on, int numsubvols,
690
               default_args_cbk_t *replies, unsigned char *output,
691
               call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
692
               mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
693
{
694
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, create, loc,
695
               flags, mode, umask, fd, xdata);
696
    return cluster_fop_success_fill(replies, numsubvols, output);
697
}
698

699
int32_t
700
cluster_link(xlator_t **subvols, unsigned char *on, int numsubvols,
701
             default_args_cbk_t *replies, unsigned char *output,
702
             call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
703
             dict_t *xdata)
704
{
705
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, link, oldloc,
706
               newloc, xdata);
707
    return cluster_fop_success_fill(replies, numsubvols, output);
708
}
709

710
int32_t
711
cluster_rename(xlator_t **subvols, unsigned char *on, int numsubvols,
712
               default_args_cbk_t *replies, unsigned char *output,
713
               call_frame_t *frame, xlator_t *this, loc_t *oldloc,
714
               loc_t *newloc, dict_t *xdata)
715
{
716
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, rename, oldloc,
717
               newloc, xdata);
718
    return cluster_fop_success_fill(replies, numsubvols, output);
719
}
720

721
int
722
cluster_symlink(xlator_t **subvols, unsigned char *on, int numsubvols,
723
                default_args_cbk_t *replies, unsigned char *output,
724
                call_frame_t *frame, xlator_t *this, const char *linkpath,
725
                loc_t *loc, mode_t umask, dict_t *xdata)
726
{
727
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, symlink,
728
               linkpath, loc, umask, xdata);
729
    return cluster_fop_success_fill(replies, numsubvols, output);
730
}
731

732
int32_t
733
cluster_rmdir(xlator_t **subvols, unsigned char *on, int numsubvols,
734
              default_args_cbk_t *replies, unsigned char *output,
735
              call_frame_t *frame, xlator_t *this, loc_t *loc, int flags,
736
              dict_t *xdata)
737
{
738
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, rmdir, loc,
739
               flags, xdata);
740
    return cluster_fop_success_fill(replies, numsubvols, output);
741
}
742

743
int32_t
744
cluster_unlink(xlator_t **subvols, unsigned char *on, int numsubvols,
745
               default_args_cbk_t *replies, unsigned char *output,
746
               call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
747
               dict_t *xdata)
748
{
749
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, unlink, loc,
750
               xflag, xdata);
751
    return cluster_fop_success_fill(replies, numsubvols, output);
752
}
753

754
int
755
cluster_mkdir(xlator_t **subvols, unsigned char *on, int numsubvols,
756
              default_args_cbk_t *replies, unsigned char *output,
757
              call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
758
              mode_t umask, dict_t *xdata)
759
{
760
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, mkdir, loc,
761
               mode, umask, xdata);
762
    return cluster_fop_success_fill(replies, numsubvols, output);
763
}
764

765
int
766
cluster_mknod(xlator_t **subvols, unsigned char *on, int numsubvols,
767
              default_args_cbk_t *replies, unsigned char *output,
768
              call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
769
              dev_t rdev, mode_t umask, dict_t *xdata)
770
{
771
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, mknod, loc,
772
               mode, rdev, umask, xdata);
773
    return cluster_fop_success_fill(replies, numsubvols, output);
774
}
775

776
int32_t
777
cluster_readlink(xlator_t **subvols, unsigned char *on, int numsubvols,
778
                 default_args_cbk_t *replies, unsigned char *output,
779
                 call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
780
                 dict_t *xdata)
781
{
782
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, readlink, loc,
783
               size, xdata);
784
    return cluster_fop_success_fill(replies, numsubvols, output);
785
}
786

787
int32_t
788
cluster_access(xlator_t **subvols, unsigned char *on, int numsubvols,
789
               default_args_cbk_t *replies, unsigned char *output,
790
               call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t mask,
791
               dict_t *xdata)
792
{
793
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, access, loc,
794
               mask, xdata);
795
    return cluster_fop_success_fill(replies, numsubvols, output);
796
}
797

798
int32_t
799
cluster_ftruncate(xlator_t **subvols, unsigned char *on, int numsubvols,
800
                  default_args_cbk_t *replies, unsigned char *output,
801
                  call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
802
                  dict_t *xdata)
803
{
804
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, ftruncate, fd,
805
               offset, xdata);
806
    return cluster_fop_success_fill(replies, numsubvols, output);
807
}
808

809
int32_t
810
cluster_getxattr(xlator_t **subvols, unsigned char *on, int numsubvols,
811
                 default_args_cbk_t *replies, unsigned char *output,
812
                 call_frame_t *frame, xlator_t *this, loc_t *loc,
813
                 const char *name, dict_t *xdata)
814
{
815
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, getxattr, loc,
816
               name, xdata);
817
    return cluster_fop_success_fill(replies, numsubvols, output);
818
}
819

820
int32_t
821
cluster_xattrop(xlator_t **subvols, unsigned char *on, int numsubvols,
822
                default_args_cbk_t *replies, unsigned char *output,
823
                call_frame_t *frame, xlator_t *this, loc_t *loc,
824
                gf_xattrop_flags_t flags, dict_t *dict, dict_t *xdata)
825
{
826
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, xattrop, loc,
827
               flags, dict, xdata);
828
    return cluster_fop_success_fill(replies, numsubvols, output);
829
}
830

831
int32_t
832
cluster_fxattrop(xlator_t **subvols, unsigned char *on, int numsubvols,
833
                 default_args_cbk_t *replies, unsigned char *output,
834
                 call_frame_t *frame, xlator_t *this, fd_t *fd,
835
                 gf_xattrop_flags_t flags, dict_t *dict, dict_t *xdata)
836
{
837
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fxattrop, fd,
838
               flags, dict, xdata);
839
    return cluster_fop_success_fill(replies, numsubvols, output);
840
}
841

842
int32_t
843
cluster_removexattr(xlator_t **subvols, unsigned char *on, int numsubvols,
844
                    default_args_cbk_t *replies, unsigned char *output,
845
                    call_frame_t *frame, xlator_t *this, loc_t *loc,
846
                    const char *name, dict_t *xdata)
847
{
848
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, removexattr,
849
               loc, name, xdata);
850
    return cluster_fop_success_fill(replies, numsubvols, output);
851
}
852

853
int32_t
854
cluster_fremovexattr(xlator_t **subvols, unsigned char *on, int numsubvols,
855
                     default_args_cbk_t *replies, unsigned char *output,
856
                     call_frame_t *frame, xlator_t *this, fd_t *fd,
857
                     const char *name, dict_t *xdata)
858
{
859
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fremovexattr,
860
               fd, name, xdata);
861
    return cluster_fop_success_fill(replies, numsubvols, output);
862
}
863

864
int32_t
865
cluster_lk(xlator_t **subvols, unsigned char *on, int numsubvols,
866
           default_args_cbk_t *replies, unsigned char *output,
867
           call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t cmd,
868
           struct gf_flock *lock, dict_t *xdata)
869
{
870
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, lk, fd, cmd,
871
               lock, xdata);
872
    return cluster_fop_success_fill(replies, numsubvols, output);
873
}
874

875
int32_t
876
cluster_rchecksum(xlator_t **subvols, unsigned char *on, int numsubvols,
877
                  default_args_cbk_t *replies, unsigned char *output,
878
                  call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
879
                  int32_t len, dict_t *xdata)
880
{
881
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, rchecksum, fd,
882
               offset, len, xdata);
883
    return cluster_fop_success_fill(replies, numsubvols, output);
884
}
885

886
int32_t
887
cluster_readdir(xlator_t **subvols, unsigned char *on, int numsubvols,
888
                default_args_cbk_t *replies, unsigned char *output,
889
                call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
890
                off_t off, dict_t *xdata)
891
{
892
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, readdir, fd,
893
               size, off, xdata);
894
    return cluster_fop_success_fill(replies, numsubvols, output);
895
}
896

897
int32_t
898
cluster_readdirp(xlator_t **subvols, unsigned char *on, int numsubvols,
899
                 default_args_cbk_t *replies, unsigned char *output,
900
                 call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
901
                 off_t off, dict_t *xdata)
902
{
903
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, readdirp, fd,
904
               size, off, xdata);
905
    return cluster_fop_success_fill(replies, numsubvols, output);
906
}
907

908
int32_t
909
cluster_setattr(xlator_t **subvols, unsigned char *on, int numsubvols,
910
                default_args_cbk_t *replies, unsigned char *output,
911
                call_frame_t *frame, xlator_t *this, loc_t *loc,
912
                struct iatt *stbuf, int32_t valid, dict_t *xdata)
913
{
914
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, setattr, loc,
915
               stbuf, valid, xdata);
916
    return cluster_fop_success_fill(replies, numsubvols, output);
917
}
918

919
int32_t
920
cluster_truncate(xlator_t **subvols, unsigned char *on, int numsubvols,
921
                 default_args_cbk_t *replies, unsigned char *output,
922
                 call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
923
                 dict_t *xdata)
924
{
925
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, truncate, loc,
926
               offset, xdata);
927
    return cluster_fop_success_fill(replies, numsubvols, output);
928
}
929

930
int32_t
931
cluster_stat(xlator_t **subvols, unsigned char *on, int numsubvols,
932
             default_args_cbk_t *replies, unsigned char *output,
933
             call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
934
{
935
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, stat, loc,
936
               xdata);
937
    return cluster_fop_success_fill(replies, numsubvols, output);
938
}
939

940
int32_t
941
cluster_lookup(xlator_t **subvols, unsigned char *on, int numsubvols,
942
               default_args_cbk_t *replies, unsigned char *output,
943
               call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
944
{
945
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, lookup, loc,
946
               xdata);
947
    return cluster_fop_success_fill(replies, numsubvols, output);
948
}
949

950
int32_t
951
cluster_fsetattr(xlator_t **subvols, unsigned char *on, int numsubvols,
952
                 default_args_cbk_t *replies, unsigned char *output,
953
                 call_frame_t *frame, xlator_t *this, fd_t *fd,
954
                 struct iatt *stbuf, int32_t valid, dict_t *xdata)
955
{
956
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fsetattr, fd,
957
               stbuf, valid, xdata);
958
    return cluster_fop_success_fill(replies, numsubvols, output);
959
}
960

961
int32_t
962
cluster_fallocate(xlator_t **subvols, unsigned char *on, int numsubvols,
963
                  default_args_cbk_t *replies, unsigned char *output,
964
                  call_frame_t *frame, xlator_t *this, fd_t *fd,
965
                  int32_t keep_size, off_t offset, size_t len, dict_t *xdata)
966
{
967
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, fallocate, fd,
968
               keep_size, offset, len, xdata);
969
    return cluster_fop_success_fill(replies, numsubvols, output);
970
}
971

972
int32_t
973
cluster_discard(xlator_t **subvols, unsigned char *on, int numsubvols,
974
                default_args_cbk_t *replies, unsigned char *output,
975
                call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
976
                size_t len, dict_t *xdata)
977
{
978
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, discard, fd,
979
               offset, len, xdata);
980
    return cluster_fop_success_fill(replies, numsubvols, output);
981
}
982

983
int32_t
984
cluster_zerofill(xlator_t **subvols, unsigned char *on, int numsubvols,
985
                 default_args_cbk_t *replies, unsigned char *output,
986
                 call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
987
                 off_t len, dict_t *xdata)
988
{
989
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, zerofill, fd,
990
               offset, len, xdata);
991
    return cluster_fop_success_fill(replies, numsubvols, output);
992
}
993

994
int32_t
995
cluster_ipc(xlator_t **subvols, unsigned char *on, int numsubvols,
996
            default_args_cbk_t *replies, unsigned char *output,
997
            call_frame_t *frame, xlator_t *this, int32_t op, dict_t *xdata)
998
{
999
    FOP_ONLIST(subvols, on, numsubvols, replies, output, frame, ipc, op, xdata);
1000
    return cluster_fop_success_fill(replies, numsubvols, output);
1001
}
1002

1003
int
1004
cluster_uninodelk(xlator_t **subvols, unsigned char *locked_on, int numsubvols,
1005
                  default_args_cbk_t *replies, unsigned char *output,
1006
                  call_frame_t *frame, xlator_t *this, char *dom,
1007
                  inode_t *inode, off_t off, size_t size)
1008
{
1009
    loc_t loc = {
1010
        0,
1011
    };
1012
    struct gf_flock flock = {
1013
        0,
1014
    };
1015

1016
    loc.inode = inode_ref(inode);
1017
    gf_uuid_copy(loc.gfid, inode->gfid);
1018

1019
    flock.l_type = F_UNLCK;
1020
    flock.l_start = off;
1021
    flock.l_len = size;
1022

1023
    FOP_ONLIST(subvols, locked_on, numsubvols, replies, output, frame, inodelk,
1024
               dom, &loc, F_SETLK, &flock, NULL);
1025

1026
    loc_wipe(&loc);
1027

1028
    return cluster_fop_success_fill(replies, numsubvols, output);
1029
}
1030

1031
int
1032
cluster_tryinodelk(xlator_t **subvols, unsigned char *on, int numsubvols,
1033
                   default_args_cbk_t *replies, unsigned char *locked_on,
1034
                   call_frame_t *frame, xlator_t *this, char *dom,
1035
                   inode_t *inode, off_t off, size_t size)
1036
{
1037
    struct gf_flock flock = {
1038
        0,
1039
    };
1040
    loc_t loc = {0};
1041

1042
    flock.l_type = F_WRLCK;
1043
    flock.l_start = off;
1044
    flock.l_len = size;
1045

1046
    loc.inode = inode_ref(inode);
1047
    gf_uuid_copy(loc.gfid, inode->gfid);
1048
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, inodelk, dom,
1049
               &loc, F_SETLK, &flock, NULL);
1050

1051
    loc_wipe(&loc);
1052
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1053
}
1054

1055
int
1056
cluster_inodelk(xlator_t **subvols, unsigned char *on, int numsubvols,
1057
                default_args_cbk_t *replies, unsigned char *locked_on,
1058
                call_frame_t *frame, xlator_t *this, char *dom, inode_t *inode,
1059
                off_t off, size_t size)
1060
{
1061
    struct gf_flock flock = {
1062
        0,
1063
    };
1064
    int i = 0;
1065
    loc_t loc = {0};
1066
    unsigned char *output = NULL;
1067

1068
    flock.l_type = F_WRLCK;
1069
    flock.l_start = off;
1070
    flock.l_len = size;
1071

1072
    output = alloca(numsubvols);
1073
    loc.inode = inode_ref(inode);
1074
    gf_uuid_copy(loc.gfid, inode->gfid);
1075
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, inodelk, dom,
1076
               &loc, F_SETLK, &flock, NULL);
1077

1078
    for (i = 0; i < numsubvols; i++) {
1079
        if (replies[i].op_ret == -1 && replies[i].op_errno == EAGAIN) {
1080
            cluster_fop_success_fill(replies, numsubvols, locked_on);
1081
            cluster_uninodelk(subvols, locked_on, numsubvols, replies, output,
1082
                              frame, this, dom, inode, off, size);
1083

1084
            FOP_SEQ(subvols, on, numsubvols, replies, locked_on, frame, inodelk,
1085
                    dom, &loc, F_SETLKW, &flock, NULL);
1086
            break;
1087
        }
1088
    }
1089

1090
    loc_wipe(&loc);
1091
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1092
}
1093

1094
int
1095
cluster_unentrylk(xlator_t **subvols, unsigned char *locked_on, int numsubvols,
1096
                  default_args_cbk_t *replies, unsigned char *output,
1097
                  call_frame_t *frame, xlator_t *this, char *dom,
1098
                  inode_t *inode, const char *name)
1099
{
1100
    loc_t loc = {
1101
        0,
1102
    };
1103

1104
    loc.inode = inode_ref(inode);
1105
    gf_uuid_copy(loc.gfid, inode->gfid);
1106

1107
    FOP_ONLIST(subvols, locked_on, numsubvols, replies, output, frame, entrylk,
1108
               dom, &loc, name, ENTRYLK_UNLOCK, ENTRYLK_WRLCK, NULL);
1109

1110
    loc_wipe(&loc);
1111

1112
    return cluster_fop_success_fill(replies, numsubvols, output);
1113
}
1114

1115
int
1116
cluster_tryentrylk(xlator_t **subvols, unsigned char *on, int numsubvols,
1117
                   default_args_cbk_t *replies, unsigned char *locked_on,
1118
                   call_frame_t *frame, xlator_t *this, char *dom,
1119
                   inode_t *inode, const char *name)
1120
{
1121
    loc_t loc = {0};
1122

1123
    loc.inode = inode_ref(inode);
1124
    gf_uuid_copy(loc.gfid, inode->gfid);
1125
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, entrylk, dom,
1126
               &loc, name, ENTRYLK_LOCK_NB, ENTRYLK_WRLCK, NULL);
1127

1128
    loc_wipe(&loc);
1129
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1130
}
1131

1132
int
1133
cluster_entrylk(xlator_t **subvols, unsigned char *on, int numsubvols,
1134
                default_args_cbk_t *replies, unsigned char *locked_on,
1135
                call_frame_t *frame, xlator_t *this, char *dom, inode_t *inode,
1136
                const char *name)
1137
{
1138
    int i = 0;
1139
    loc_t loc = {0};
1140
    unsigned char *output = NULL;
1141

1142
    output = alloca(numsubvols);
1143
    loc.inode = inode_ref(inode);
1144
    gf_uuid_copy(loc.gfid, inode->gfid);
1145
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, entrylk, dom,
1146
               &loc, name, ENTRYLK_LOCK_NB, ENTRYLK_WRLCK, NULL);
1147

1148
    for (i = 0; i < numsubvols; i++) {
1149
        if (replies[i].op_ret == -1 && replies[i].op_errno == EAGAIN) {
1150
            cluster_fop_success_fill(replies, numsubvols, locked_on);
1151
            cluster_unentrylk(subvols, locked_on, numsubvols, replies, output,
1152
                              frame, this, dom, inode, name);
1153
            FOP_SEQ(subvols, on, numsubvols, replies, locked_on, frame, entrylk,
1154
                    dom, &loc, name, ENTRYLK_LOCK, ENTRYLK_WRLCK, NULL);
1155
            break;
1156
        }
1157
    }
1158

1159
    loc_wipe(&loc);
1160
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1161
}
1162

1163
int
1164
cluster_tiebreaker_inodelk(xlator_t **subvols, unsigned char *on,
1165
                           int numsubvols, default_args_cbk_t *replies,
1166
                           unsigned char *locked_on, call_frame_t *frame,
1167
                           xlator_t *this, char *dom, inode_t *inode, off_t off,
1168
                           size_t size)
1169
{
1170
    struct gf_flock flock = {
1171
        0,
1172
    };
1173
    int i = 0;
1174
    int num_success = 0;
1175
    loc_t loc = {0};
1176
    unsigned char *output = NULL;
1177

1178
    flock.l_type = F_WRLCK;
1179
    flock.l_start = off;
1180
    flock.l_len = size;
1181

1182
    output = alloca(numsubvols);
1183
    loc.inode = inode_ref(inode);
1184
    gf_uuid_copy(loc.gfid, inode->gfid);
1185
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, inodelk, dom,
1186
               &loc, F_SETLK, &flock, NULL);
1187

1188
    for (i = 0; i < numsubvols; i++) {
1189
        if (replies[i].valid && replies[i].op_ret == 0) {
1190
            num_success++;
1191
            continue;
1192
        }
1193

1194
        /* TODO: If earlier subvols fail with an error other
1195
         * than EAGAIN, we could still have 2 clients competing
1196
         * for the lock*/
1197
        if (replies[i].op_ret == -1 && replies[i].op_errno == EAGAIN) {
1198
            cluster_fop_success_fill(replies, numsubvols, locked_on);
1199
            cluster_uninodelk(subvols, locked_on, numsubvols, replies, output,
1200
                              frame, this, dom, inode, off, size);
1201

1202
            if (num_success) {
1203
                FOP_SEQ(subvols, on, numsubvols, replies, locked_on, frame,
1204
                        inodelk, dom, &loc, F_SETLKW, &flock, NULL);
1205
            } else {
1206
                loc_wipe(&loc);
1207
                memset(locked_on, 0, numsubvols);
1208
                return 0;
1209
            }
1210
            break;
1211
        }
1212
    }
1213

1214
    loc_wipe(&loc);
1215
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1216
}
1217

1218
int
1219
cluster_tiebreaker_entrylk(xlator_t **subvols, unsigned char *on,
1220
                           int numsubvols, default_args_cbk_t *replies,
1221
                           unsigned char *locked_on, call_frame_t *frame,
1222
                           xlator_t *this, char *dom, inode_t *inode,
1223
                           const char *name)
1224
{
1225
    int i = 0;
1226
    loc_t loc = {0};
1227
    unsigned char *output = NULL;
1228
    int num_success = 0;
1229

1230
    output = alloca(numsubvols);
1231
    loc.inode = inode_ref(inode);
1232
    gf_uuid_copy(loc.gfid, inode->gfid);
1233
    FOP_ONLIST(subvols, on, numsubvols, replies, locked_on, frame, entrylk, dom,
1234
               &loc, name, ENTRYLK_LOCK_NB, ENTRYLK_WRLCK, NULL);
1235

1236
    for (i = 0; i < numsubvols; i++) {
1237
        if (replies[i].valid && replies[i].op_ret == 0) {
1238
            num_success++;
1239
            continue;
1240
        }
1241
        if (replies[i].op_ret == -1 && replies[i].op_errno == EAGAIN) {
1242
            cluster_fop_success_fill(replies, numsubvols, locked_on);
1243
            cluster_unentrylk(subvols, locked_on, numsubvols, replies, output,
1244
                              frame, this, dom, inode, name);
1245
            if (num_success) {
1246
                FOP_SEQ(subvols, on, numsubvols, replies, locked_on, frame,
1247
                        entrylk, dom, &loc, name, ENTRYLK_LOCK, ENTRYLK_WRLCK,
1248
                        NULL);
1249
            } else {
1250
                loc_wipe(&loc);
1251
                memset(locked_on, 0, numsubvols);
1252
                return 0;
1253
            }
1254
            break;
1255
        }
1256
    }
1257

1258
    loc_wipe(&loc);
1259
    return cluster_fop_success_fill(replies, numsubvols, locked_on);
1260
}
1261