glusterfs

1
/*
2
   Copyright (c) 2013-2014 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
#include "glusterd-op-sm.h"
11
#include "glusterd-store.h"
12
#include "glusterd-utils.h"
13
#include "glusterd-volgen.h"
14
#include "glusterd-locks.h"
15
#include "glusterd-errno.h"
16
#include <glusterfs/run.h>
17
#include <glusterfs/syscall.h>
18
#include "glusterd-messages.h"
19

20
#include <signal.h>
21

22
#define GF_MAX_LOCKING_ENTITIES 3
23

24
/* Valid entities that the mgmt_v3 lock can hold locks upon    *
25
 * To add newer entities to be locked, we can just add more    *
26
 * entries to this table along with the type and default value */
27
glusterd_valid_entities valid_types[] = {
28
    {"vol", _gf_true},
29
    {"snap", _gf_false},
30
    {"global", _gf_false},
31
    {NULL},
32
};
33

34
/* Checks if the lock request is for a valid entity */
35
static gf_boolean_t
36
glusterd_mgmt_v3_is_type_valid(char *type)
37
{
38
    int i = 0;
39

40
    GF_ASSERT(type);
41

42
    for (i = 0; valid_types[i].type; i++) {
43
        if (!strcmp(type, valid_types[i].type)) {
44
            return _gf_true;
45
        }
46
    }
47

48
    return _gf_false;
49
}
50

51
/* Initialize the global mgmt_v3 lock list(dict) when
52
 * glusterd is spawned */
53
int32_t
54
glusterd_mgmt_v3_lock_init(void)
55
{
56
    int32_t ret = -1;
57
    glusterd_conf_t *priv = NULL;
58

59
    priv = THIS->private;
60
    GF_ASSERT(priv);
61

62
    priv->mgmt_v3_lock = dict_new();
63
    if (!priv->mgmt_v3_lock)
64
        goto out;
65

66
    ret = 0;
67
out:
68
    return ret;
69
}
70

71
/* Destroy the global mgmt_v3 lock list(dict) when
72
 * glusterd cleanup is performed */
73
void
74
glusterd_mgmt_v3_lock_fini(void)
75
{
76
    glusterd_conf_t *priv = NULL;
77

78
    priv = THIS->private;
79
    GF_ASSERT(priv);
80

81
    if (priv->mgmt_v3_lock)
82
        dict_unref(priv->mgmt_v3_lock);
83
}
84

85
/* Initialize the global mgmt_v3_timer lock list(dict) when
86
 * glusterd is spawned */
87
int32_t
88
glusterd_mgmt_v3_lock_timer_init(void)
89
{
90
    int32_t ret = -1;
91
    xlator_t *this = THIS;
92
    glusterd_conf_t *priv = NULL;
93

94
    priv = this->private;
95
    GF_VALIDATE_OR_GOTO(this->name, priv, out);
96

97
    priv->mgmt_v3_lock_timer = dict_new();
98
    if (!priv->mgmt_v3_lock_timer)
99
        goto out;
100

101
    ret = 0;
102
out:
103
    return ret;
104
}
105

106
/* Destroy the global mgmt_v3_timer lock list(dict) when
107
 * glusterd cleanup is performed */
108
void
109
glusterd_mgmt_v3_lock_timer_fini(void)
110
{
111
    xlator_t *this = THIS;
112
    glusterd_conf_t *priv = NULL;
113

114
    priv = this->private;
115
    GF_VALIDATE_OR_GOTO(this->name, priv, out);
116

117
    if (priv->mgmt_v3_lock_timer)
118
        dict_unref(priv->mgmt_v3_lock_timer);
119
out:
120
    return;
121
}
122

123
static int32_t
124
glusterd_get_mgmt_v3_lock_owner(char *key, uuid_t *uuid)
125
{
126
    int32_t ret = -1;
127
    glusterd_mgmt_v3_lock_obj *lock_obj = NULL;
128
    glusterd_conf_t *priv = NULL;
129
    xlator_t *this = THIS;
130

131
    priv = this->private;
132
    GF_ASSERT(priv);
133

134
    if (!key || !uuid) {
135
        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
136
               "key or uuid is null.");
137
        ret = -1;
138
        goto out;
139
    }
140

141
    ret = dict_get_bin(priv->mgmt_v3_lock, key, (void **)&lock_obj);
142
    if (!ret)
143
        gf_uuid_copy(*uuid, lock_obj->lock_owner);
144

145
    ret = 0;
146
out:
147
    gf_msg_trace(this->name, 0, "Returning %d", ret);
148
    return ret;
149
}
150

151
/* This function is called with the locked_count and type, to   *
152
 * release all the acquired locks. */
153
static int32_t
154
glusterd_release_multiple_locks_per_entity(dict_t *dict, uuid_t uuid,
155
                                           int32_t locked_count, char *type)
156
{
157
    char name_buf[PATH_MAX] = "";
158
    char *name = NULL;
159
    int32_t i = -1;
160
    int32_t op_ret = 0;
161
    int32_t ret = -1;
162
    xlator_t *this = THIS;
163

164
    GF_ASSERT(dict);
165
    GF_ASSERT(type);
166

167
    if (locked_count == 0) {
168
        gf_msg_debug(this->name, 0, "No %s locked as part of this transaction",
169
                     type);
170
        goto out;
171
    }
172

173
    /* Release all the locks held */
174
    for (i = 0; i < locked_count; i++) {
175
        ret = snprintf(name_buf, sizeof(name_buf), "%sname%d", type, i + 1);
176

177
        /* Looking for volname1, volname2 or snapname1, *
178
         * as key in the dict snapname2 */
179
        ret = dict_get_strn(dict, name_buf, ret, &name);
180
        if (ret) {
181
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
182
                   "Unable to get %s locked_count = %d", name_buf,
183
                   locked_count);
184
            op_ret = ret;
185
            continue;
186
        }
187

188
        ret = glusterd_mgmt_v3_unlock(name, uuid, type);
189
        if (ret) {
190
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_MGMTV3_UNLOCK_FAIL,
191
                   "Failed to release lock for %s.", name);
192
            op_ret = ret;
193
        }
194
    }
195

196
out:
197
    gf_msg_trace(this->name, 0, "Returning %d", op_ret);
198
    return op_ret;
199
}
200

201
/* Given the count and type of the entity this function acquires     *
202
 * locks on multiple elements of the same entity. For example:       *
203
 * If type is "vol" this function tries to acquire locks on multiple *
204
 * volumes */
205
static int32_t
206
glusterd_acquire_multiple_locks_per_entity(dict_t *dict, uuid_t uuid,
207
                                           uint32_t *op_errno, int32_t count,
208
                                           char *type)
209
{
210
    char name_buf[PATH_MAX] = "";
211
    char *name = NULL;
212
    int32_t i = -1;
213
    int32_t ret = -1;
214
    int32_t locked_count = 0;
215
    xlator_t *this = THIS;
216

217
    GF_ASSERT(dict);
218
    GF_ASSERT(type);
219

220
    /* Locking one element after other */
221
    for (i = 0; i < count; i++) {
222
        ret = snprintf(name_buf, sizeof(name_buf), "%sname%d", type, i + 1);
223

224
        /* Looking for volname1, volname2 or snapname1, *
225
         * as key in the dict snapname2 */
226
        ret = dict_get_strn(dict, name_buf, ret, &name);
227
        if (ret) {
228
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
229
                   "Unable to get %s count = %d", name_buf, count);
230
            break;
231
        }
232

233
        ret = glusterd_mgmt_v3_lock(name, uuid, op_errno, type);
234
        if (ret) {
235
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_MGMTV3_LOCK_GET_FAIL,
236
                   "Failed to acquire lock for %s %s "
237
                   "on behalf of %s. Reversing "
238
                   "this transaction",
239
                   type, name, uuid_utoa(uuid));
240
            break;
241
        }
242
        locked_count++;
243
    }
244

245
    if (count == locked_count) {
246
        /* If all locking ops went successfully, return as success */
247
        ret = 0;
248
        goto out;
249
    }
250

251
    /* If we failed to lock one element, unlock others and return failure */
252
    ret = glusterd_release_multiple_locks_per_entity(dict, uuid, locked_count,
253
                                                     type);
254
    if (ret) {
255
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_MULTIPLE_LOCK_RELEASE_FAIL,
256
               "Failed to release multiple %s locks", type);
257
    }
258
    ret = -1;
259
out:
260
    gf_msg_trace(this->name, 0, "Returning %d", ret);
261
    return ret;
262
}
263

264
/* Given the type of entity, this function figures out if it should unlock a   *
265
 * single element of multiple elements of the said entity. For example:        *
266
 * if the type is "vol", this function will accordingly unlock a single volume *
267
 * or multiple volumes */
268
static int32_t
269
glusterd_mgmt_v3_unlock_entity(dict_t *dict, uuid_t uuid, char *type,
270
                               gf_boolean_t default_value)
271
{
272
    char name_buf[PATH_MAX] = "";
273
    char *name = NULL;
274
    int32_t count = -1;
275
    int32_t ret = -1;
276
    gf_boolean_t hold_locks = _gf_false;
277
    xlator_t *this = THIS;
278

279
    GF_ASSERT(dict);
280
    GF_ASSERT(type);
281

282
    snprintf(name_buf, sizeof(name_buf), "hold_%s_locks", type);
283
    hold_locks = dict_get_str_boolean(dict, name_buf, default_value);
284

285
    if (hold_locks == _gf_false) {
286
        /* Locks were not held for this particular entity *
287
         * Hence nothing to release */
288
        ret = 0;
289
        goto out;
290
    }
291

292
    /* Looking for volcount or snapcount in the dict */
293
    ret = snprintf(name_buf, sizeof(name_buf), "%scount", type);
294
    ret = dict_get_int32n(dict, name_buf, ret, &count);
295
    if (ret) {
296
        /* count is not present. Only one *
297
         * element name needs to be unlocked */
298
        ret = snprintf(name_buf, sizeof(name_buf), "%sname", type);
299
        ret = dict_get_strn(dict, name_buf, ret, &name);
300
        if (ret) {
301
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
302
                   "Unable to fetch %sname", type);
303
            goto out;
304
        }
305

306
        ret = glusterd_mgmt_v3_unlock(name, uuid, type);
307
        if (ret) {
308
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_MGMTV3_UNLOCK_FAIL,
309
                   "Failed to release lock for %s %s "
310
                   "on behalf of %s.",
311
                   type, name, uuid_utoa(uuid));
312
            goto out;
313
        }
314
    } else {
315
        /* Unlocking one element name after another */
316
        ret = glusterd_release_multiple_locks_per_entity(dict, uuid, count,
317
                                                         type);
318
        if (ret) {
319
            gf_msg(this->name, GF_LOG_ERROR, 0,
320
                   GD_MSG_MULTIPLE_LOCK_RELEASE_FAIL,
321
                   "Failed to release all %s locks", type);
322
            goto out;
323
        }
324
    }
325

326
    ret = 0;
327
out:
328
    gf_msg_trace(this->name, 0, "Returning %d", ret);
329
    return ret;
330
}
331

332
/* Given the type of entity, this function figures out if it should lock a   *
333
 * single element or multiple elements of the said entity. For example:      *
334
 * if the type is "vol", this function will accordingly lock a single volume *
335
 * or multiple volumes */
336
static int32_t
337
glusterd_mgmt_v3_lock_entity(dict_t *dict, uuid_t uuid, uint32_t *op_errno,
338
                             char *type, gf_boolean_t default_value)
339
{
340
    char name_buf[PATH_MAX] = "";
341
    char *name = NULL;
342
    int32_t count = -1;
343
    int32_t ret = -1;
344
    gf_boolean_t hold_locks = _gf_false;
345
    xlator_t *this = THIS;
346

347
    GF_ASSERT(dict);
348
    GF_ASSERT(type);
349

350
    snprintf(name_buf, sizeof(name_buf), "hold_%s_locks", type);
351
    hold_locks = dict_get_str_boolean(dict, name_buf, default_value);
352

353
    if (hold_locks == _gf_false) {
354
        /* Not holding locks for this particular entity */
355
        ret = 0;
356
        goto out;
357
    }
358

359
    /* Looking for volcount or snapcount in the dict */
360
    ret = snprintf(name_buf, sizeof(name_buf), "%scount", type);
361
    ret = dict_get_int32n(dict, name_buf, ret, &count);
362
    if (ret) {
363
        /* count is not present. Only one *
364
         * element name needs to be locked */
365
        ret = snprintf(name_buf, sizeof(name_buf), "%sname", type);
366
        ret = dict_get_strn(dict, name_buf, ret, &name);
367
        if (ret) {
368
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
369
                   "Unable to fetch %sname", type);
370
            goto out;
371
        }
372

373
        ret = glusterd_mgmt_v3_lock(name, uuid, op_errno, type);
374
        if (ret) {
375
            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_MGMTV3_LOCK_GET_FAIL,
376
                   "Failed to acquire lock for %s %s "
377
                   "on behalf of %s.",
378
                   type, name, uuid_utoa(uuid));
379
            goto out;
380
        }
381
    } else {
382
        /* Locking one element name after another */
383
        ret = glusterd_acquire_multiple_locks_per_entity(dict, uuid, op_errno,
384
                                                         count, type);
385
        if (ret) {
386
            gf_msg(this->name, GF_LOG_ERROR, 0,
387
                   GD_MSG_MULTIPLE_LOCK_ACQUIRE_FAIL,
388
                   "Failed to acquire all %s locks", type);
389
            goto out;
390
        }
391
    }
392

393
    ret = 0;
394
out:
395
    gf_msg_trace(this->name, 0, "Returning %d", ret);
396
    return ret;
397
}
398

399
/* Try to release locks of multiple entities like *
400
 * volume, snaps etc. */
401
int32_t
402
glusterd_multiple_mgmt_v3_unlock(dict_t *dict, uuid_t uuid)
403
{
404
    int32_t i = -1;
405
    int32_t ret = -1;
406
    int32_t op_ret = 0;
407
    xlator_t *this = THIS;
408

409
    if (!dict) {
410
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_EMPTY, "dict is null.");
411
        ret = -1;
412
        goto out;
413
    }
414

415
    for (i = 0; valid_types[i].type; i++) {
416
        ret = glusterd_mgmt_v3_unlock_entity(dict, uuid, valid_types[i].type,
417
                                             valid_types[i].default_value);
418
        if (ret) {
419
            gf_msg(this->name, GF_LOG_ERROR, 0,
420
                   GD_MSG_MULTIPLE_LOCK_RELEASE_FAIL, "Unable to unlock all %s",
421
                   valid_types[i].type);
422
            op_ret = ret;
423
        }
424
    }
425

426
    ret = op_ret;
427
out:
428
    gf_msg_debug(this->name, 0, "Returning %d", ret);
429
    return ret;
430
}
431

432
/* Try to acquire locks on multiple entities like *
433
 * volume, snaps etc. */
434
int32_t
435
glusterd_multiple_mgmt_v3_lock(dict_t *dict, uuid_t uuid, uint32_t *op_errno)
436
{
437
    int32_t i = -1;
438
    int32_t ret = -1;
439
    int32_t locked_count = 0;
440
    xlator_t *this = THIS;
441

442
    if (!dict) {
443
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_EMPTY, "dict is null.");
444
        ret = -1;
445
        goto out;
446
    }
447

448
    /* Locking one entity after other */
449
    for (i = 0; valid_types[i].type; i++) {
450
        ret = glusterd_mgmt_v3_lock_entity(dict, uuid, op_errno,
451
                                           valid_types[i].type,
452
                                           valid_types[i].default_value);
453
        if (ret) {
454
            gf_msg(this->name, GF_LOG_ERROR, 0,
455
                   GD_MSG_MULTIPLE_LOCK_ACQUIRE_FAIL, "Unable to lock all %s",
456
                   valid_types[i].type);
457
            break;
458
        }
459
        locked_count++;
460
    }
461

462
    if (locked_count == GF_MAX_LOCKING_ENTITIES) {
463
        /* If all locking ops went successfully, return as success */
464
        ret = 0;
465
        goto out;
466
    }
467

468
    /* If we failed to lock one entity, unlock others and return failure */
469
    for (i = 0; i < locked_count; i++) {
470
        ret = glusterd_mgmt_v3_unlock_entity(dict, uuid, valid_types[i].type,
471
                                             valid_types[i].default_value);
472
        if (ret) {
473
            gf_msg(this->name, GF_LOG_ERROR, 0,
474
                   GD_MSG_MULTIPLE_LOCK_RELEASE_FAIL, "Unable to unlock all %s",
475
                   valid_types[i].type);
476
        }
477
    }
478
    ret = -1;
479
out:
480
    gf_msg_debug(this->name, 0, "Returning %d", ret);
481
    return ret;
482
}
483

484
int32_t
485
glusterd_mgmt_v3_lock(const char *name, uuid_t uuid, uint32_t *op_errno,
486
                      char *type)
487
{
488
    char key[PATH_MAX] = "";
489
    int32_t ret = -1;
490
    glusterd_mgmt_v3_lock_obj *lock_obj = NULL;
491
    gf_timer_t *mgmt_lock_timer = NULL;
492
    glusterd_conf_t *priv = NULL;
493
    gf_boolean_t is_valid = _gf_true;
494
    uuid_t owner = {0};
495
    xlator_t *this = THIS;
496
    struct timespec delay = {0};
497
    char *key_dup = NULL;
498

499
    priv = this->private;
500
    GF_ASSERT(priv);
501

502
    if (!name || !type) {
503
        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
504
               "name or type is null.");
505
        ret = -1;
506
        goto out;
507
    }
508

509
    is_valid = glusterd_mgmt_v3_is_type_valid(type);
510
    if (is_valid != _gf_true) {
511
        gf_msg_callingfn(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
512
                         "Invalid entity. Cannot perform locking "
513
                         "operation on %s types",
514
                         type);
515
        ret = -1;
516
        goto out;
517
    }
518

519
    ret = snprintf(key, sizeof(key), "%s_%s", name, type);
520
    if (ret != strlen(name) + 1 + strlen(type)) {
521
        ret = -1;
522
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_CREATE_KEY_FAIL,
523
               "Unable to create key");
524
        goto out;
525
    }
526

527
    gf_msg_debug(this->name, 0, "Trying to acquire lock of %s for %s", key,
528
                 uuid_utoa(uuid));
529

530
    ret = glusterd_get_mgmt_v3_lock_owner(key, &owner);
531
    if (ret) {
532
        gf_msg_debug(this->name, 0, "Unable to get mgmt_v3 lock owner");
533
        goto out;
534
    }
535

536
    /* If the lock has already been held for the given volume
537
     * we fail */
538
    if (!gf_uuid_is_null(owner)) {
539
        gf_msg_callingfn(this->name, GF_LOG_WARNING, 0,
540
                         GD_MSG_LOCK_ALREADY_HELD, "Lock for %s held by %s",
541
                         name, uuid_utoa(owner));
542
        ret = -1;
543
        *op_errno = EG_ANOTRANS;
544
        goto out;
545
    }
546

547
    lock_obj = GF_MALLOC(sizeof(glusterd_mgmt_v3_lock_obj),
548
                         gf_common_mt_mgmt_v3_lock_obj_t);
549
    if (!lock_obj) {
550
        ret = -1;
551
        goto out;
552
    }
553

554
    gf_uuid_copy(lock_obj->lock_owner, uuid);
555

556
    ret = dict_set_bin(priv->mgmt_v3_lock, key, lock_obj,
557
                       sizeof(glusterd_mgmt_v3_lock_obj));
558
    if (ret) {
559
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
560
               "Unable to set lock owner in mgmt_v3 lock");
561
        GF_FREE(lock_obj);
562
        goto out;
563
    }
564

565
    key_dup = gf_strdup(key);
566
    delay.tv_sec = priv->mgmt_v3_lock_timeout;
567
    delay.tv_nsec = 0;
568

569
    /* Changing to default timeout value. */
570
    priv->mgmt_v3_lock_timeout = GF_LOCK_TIMER;
571

572
    mgmt_lock_timer = gf_timer_call_after(this->ctx, delay,
573
                                          gd_mgmt_v3_unlock_timer_cbk, key_dup);
574

575
    /* Timer object is allocated dynamically but freed with
576
       gf_timer_call_cancel(). So it should not be managed by
577
       dict to avoid double free and treated as static here.
578
       Also note that just the pointer to timer is stored in
579
       the dict, not the timer object as a whole (the latter
580
       causes copying and most likely a crash in timer code). */
581
    /* coverity[SUSPICIOUS_SIZEOF] */
582
    ret = dict_set_static_bin(priv->mgmt_v3_lock_timer, key, mgmt_lock_timer,
583
                              sizeof(mgmt_lock_timer));
584
    if (ret) {
585
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
586
               "Unable to set timer in mgmt_v3 lock");
587
        GF_FREE(key_dup);
588
        GF_FREE(mgmt_lock_timer);
589
        goto out;
590
    }
591

592
#ifdef DEBUG
593
    char *bt = NULL;
594

595
    /* Saving the backtrace into the pre-allocated buffer, ctx->btbuf*/
596
    if ((bt = gf_backtrace_save(NULL))) {
597
        snprintf(key, sizeof(key), "debug.last-success-bt-%s", key_dup);
598
        ret = dict_set_dynstr_with_alloc(priv->mgmt_v3_lock, key, bt);
599
        if (ret)
600
            gf_msg(this->name, GF_LOG_WARNING, 0, GD_MSG_DICT_SET_FAILED,
601
                   "Failed to save "
602
                   "the back trace for lock %s granted to %s",
603
                   key_dup, uuid_utoa(uuid));
604
        ret = 0;
605
    }
606

607
#endif
608

609
    gf_msg_debug(this->name, 0, "Lock for %s successfully held by %s", key_dup,
610
                 uuid_utoa(uuid));
611

612
    ret = 0;
613
out:
614
    gf_msg_trace(this->name, 0, "Returning %d", ret);
615
    return ret;
616
}
617

618
/*
619
 * This call back will ensure to unlock the lock_obj, in case we hit a situation
620
 * where unlocking failed and stale lock exist*/
621
void
622
gd_mgmt_v3_unlock_timer_cbk(void *data)
623
{
624
    xlator_t *this = THIS;
625
    glusterd_conf_t *conf = NULL;
626
    gf_timer_t *mgmt_lock_timer = NULL;
627
    char *key = NULL;
628
    int keylen;
629
    int32_t ret = -1;
630

631
    conf = this->private;
632
    GF_VALIDATE_OR_GOTO(this->name, conf, out);
633

634
    GF_ASSERT(NULL != data);
635
    key = (char *)data;
636

637
    keylen = strlen(key);
638
    dict_deln(conf->mgmt_v3_lock, key, keylen);
639

640
#ifdef DEBUG
641
    char bt_key[PATH_MAX] = "";
642
    int bt_key_len = 0;
643

644
    bt_key_len = snprintf(bt_key, PATH_MAX, "debug.last-success-bt-%s", key);
645
    if (bt_key_len != SLEN("debug.last-success-bt-") + keylen) {
646
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_CREATE_KEY_FAIL,
647
               "Unable to create backtrace "
648
               "key");
649
        goto out;
650
    }
651

652
    dict_deln(conf->mgmt_v3_lock, bt_key, bt_key_len);
653
#endif
654

655
    ret = dict_get_bin(conf->mgmt_v3_lock_timer, key,
656
                       (void **)&mgmt_lock_timer);
657
    if (ret) {
658
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
659
               "Unable to get lock owner in mgmt_v3 lock");
660
    }
661

662
out:
663
    if (mgmt_lock_timer) {
664
        GF_ASSERT(mgmt_lock_timer->xl && mgmt_lock_timer->xl->ctx);
665
        GF_FREE(mgmt_lock_timer->data);
666
        gf_timer_call_cancel(mgmt_lock_timer->xl->ctx, mgmt_lock_timer);
667
#ifdef DEBUG
668
        dict_deln(conf->mgmt_v3_lock_timer, bt_key, bt_key_len);
669
#endif
670
        gf_log(this->name, GF_LOG_INFO,
671
               "unlock timer is cancelled for volume_type"
672
               " %s",
673
               key);
674
    }
675
}
676

677
int32_t
678
glusterd_mgmt_v3_unlock(const char *name, uuid_t uuid, char *type)
679
{
680
    char key[PATH_MAX] = "";
681
    char key_dup[PATH_MAX] = "";
682
    int keylen;
683
    int32_t ret = -1;
684
    gf_boolean_t is_valid = _gf_true;
685
    glusterd_conf_t *priv = NULL;
686
    glusterd_volinfo_t *volinfo = NULL;
687
    gf_timer_t *mgmt_lock_timer = NULL;
688
    uuid_t owner = {0};
689
    xlator_t *this = THIS;
690

691
    priv = this->private;
692
    GF_ASSERT(priv);
693

694
    if (!name || !type) {
695
        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
696
               "name is null.");
697
        ret = -1;
698
        goto out;
699
    }
700

701
    is_valid = glusterd_mgmt_v3_is_type_valid(type);
702
    if (is_valid != _gf_true) {
703
        gf_msg_callingfn(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
704
                         "Invalid entity. Cannot perform unlocking "
705
                         "operation on %s types",
706
                         type);
707
        ret = -1;
708
        goto out;
709
    }
710

711
    keylen = snprintf(key, sizeof(key), "%s_%s", name, type);
712
    if (keylen != strlen(name) + 1 + strlen(type)) {
713
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_CREATE_KEY_FAIL,
714
               "Unable to create key");
715
        ret = -1;
716
        goto out;
717
    }
718

719
    gf_msg_debug(this->name, 0, "Trying to release lock of %s %s for %s as %s",
720
                 type, name, uuid_utoa(uuid), key);
721

722
    ret = glusterd_get_mgmt_v3_lock_owner(key, &owner);
723
    if (ret) {
724
        gf_msg_debug(this->name, 0, "Unable to get mgmt_v3 lock owner");
725
        goto out;
726
    }
727

728
    if (gf_uuid_is_null(owner)) {
729
        gf_msg_callingfn(this->name, GF_LOG_WARNING, 0, GD_MSG_LOCK_NOT_HELD,
730
                         "Lock for %s %s not held", type, name);
731
        ret = -1;
732
        goto out;
733
    }
734

735
    ret = gf_uuid_compare(uuid, owner);
736
    if (ret) {
737
        gf_msg_callingfn(this->name, GF_LOG_WARNING, 0,
738
                         GD_MSG_LOCK_OWNER_MISMATCH,
739
                         "Lock owner mismatch. "
740
                         "Lock for %s %s held by %s",
741
                         type, name, uuid_utoa(owner));
742
        goto out;
743
    }
744

745
    /* Removing the mgmt_v3 lock from the global list */
746
    dict_deln(priv->mgmt_v3_lock, key, keylen);
747

748
    ret = dict_get_bin(priv->mgmt_v3_lock_timer, key,
749
                       (void **)&mgmt_lock_timer);
750
    if (ret) {
751
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
752
               "Unable to get mgmt lock key in mgmt_v3 lock");
753
        goto out;
754
    }
755

756
    (void)snprintf(key_dup, sizeof(key_dup), "%s", key);
757

758
#ifdef DEBUG
759
    /* Remove the backtrace key as well */
760
    ret = snprintf(key, sizeof(key), "debug.last-success-bt-%s", key_dup);
761
    if (ret != SLEN("debug.last-success-bt-") + keylen) {
762
        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_CREATE_KEY_FAIL,
763
               "Unable to create backtrace "
764
               "key");
765
        ret = -1;
766
        goto out;
767
    }
768
    dict_deln(priv->mgmt_v3_lock, key, ret);
769
#endif
770

771
    gf_msg_debug(this->name, 0, "Lock for %s %s successfully released", type,
772
                 name);
773

774
    /* Release owner reference which was held during lock. */
775
    if (mgmt_lock_timer) {
776
        GF_ASSERT(mgmt_lock_timer->xl && mgmt_lock_timer->xl->ctx);
777
        GF_FREE(mgmt_lock_timer->data);
778
        gf_timer_call_cancel(mgmt_lock_timer->xl->ctx, mgmt_lock_timer);
779
        dict_deln(priv->mgmt_v3_lock_timer, key_dup, keylen);
780
    }
781
    ret = glusterd_volinfo_find(name, &volinfo);
782
    if (volinfo && volinfo->stage_deleted) {
783
        /* this indicates a volume still exists and the volume delete
784
         * operation has failed in some of the phases, need to ensure
785
         * stage_deleted flag is set back to false
786
         */
787
        volinfo->stage_deleted = _gf_false;
788
        gf_log(this->name, GF_LOG_INFO,
789
               "Volume %s still exist, setting "
790
               "stage deleted flag to false for the volume",
791
               volinfo->volname);
792
    }
793
    ret = 0;
794
out:
795

796
    gf_msg_trace(this->name, 0, "Returning %d", ret);
797
    return ret;
798
}
799