source3/lib/g_lock.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    global locks based on dbwrap and messaging
   4    Copyright (C) 2009 by Volker Lendecke
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "replace.h"
  21 #include "system/filesys.h"
  22 #include "lib/util/server_id.h"
  23 #include "lib/util/debug.h"
  24 #include "lib/util/talloc_stack.h"
  25 #include "lib/util/samba_util.h"
  26 #include "lib/util_path.h"
  27 #include "dbwrap/dbwrap.h"
  28 #include "dbwrap/dbwrap_open.h"
  29 #include "dbwrap/dbwrap_watch.h"
  30 #include "g_lock.h"
  31 #include "util_tdb.h"
  32 #include "../lib/util/tevent_ntstatus.h"
  33 #include "messages.h"
  34 #include "serverid.h"
  35
  36 struct g_lock_ctx {
  37         struct db_context *db;
  38         struct messaging_context *msg;
  39         enum dbwrap_lock_order lock_order;
  40 };
  41
  42 struct g_lock {
  43         struct server_id exclusive;
  44         size_t num_shared;
  45         uint8_t *shared;
  46         uint64_t unique_data_epoch;
  47         size_t datalen;
  48         uint8_t *data;
  49 };
  50
  51 static bool g_lock_parse(uint8_t *buf, size_t buflen, struct g_lock *lck)
  52 {
  53         struct server_id exclusive;
  54         size_t num_shared, shared_len;
  55         uint64_t unique_data_epoch;
  56
  57         if (buflen < (SERVER_ID_BUF_LENGTH + /* exclusive */
  58                       sizeof(uint64_t) +     /* seqnum */
  59                       sizeof(uint32_t))) {   /* num_shared */
  60                 struct g_lock ret = {
  61                         .exclusive.pid = 0,
  62                         .unique_data_epoch = generate_unique_u64(0),
  63                 };
  64                 *lck = ret;
  65                 return true;
  66         }
  67
  68         server_id_get(&exclusive, buf);
  69         buf += SERVER_ID_BUF_LENGTH;
  70         buflen -= SERVER_ID_BUF_LENGTH;
  71
  72         unique_data_epoch = BVAL(buf, 0);
  73         buf += sizeof(uint64_t);
  74         buflen -= sizeof(uint64_t);
  75
  76         num_shared = IVAL(buf, 0);
  77         buf += sizeof(uint32_t);
  78         buflen -= sizeof(uint32_t);
  79
  80         if (num_shared > buflen/SERVER_ID_BUF_LENGTH) {
  81                 DBG_DEBUG("num_shared=%zu, buflen=%zu\n",
  82                           num_shared,
  83                           buflen);
  84                 return false;
  85         }
  86
  87         shared_len = num_shared * SERVER_ID_BUF_LENGTH;
  88
  89         *lck = (struct g_lock) {
  90                 .exclusive = exclusive,
  91                 .num_shared = num_shared,
  92                 .shared = buf,
  93                 .unique_data_epoch = unique_data_epoch,
  94                 .datalen = buflen-shared_len,
  95                 .data = buf+shared_len,
  96         };
  97
  98         return true;
  99 }
 100
 101 static void g_lock_get_shared(const struct g_lock *lck,
 102                               size_t i,
 103                               struct server_id *shared)
 104 {
 105         if (i >= lck->num_shared) {
 106                 abort();
 107         }
 108         server_id_get(shared, lck->shared + i*SERVER_ID_BUF_LENGTH);
 109 }
 110
 111 static void g_lock_del_shared(struct g_lock *lck, size_t i)
 112 {
 113         if (i >= lck->num_shared) {
 114                 abort();
 115         }
 116         lck->num_shared -= 1;
 117         if (i < lck->num_shared) {
 118                 memcpy(lck->shared + i*SERVER_ID_BUF_LENGTH,
 119                        lck->shared + lck->num_shared*SERVER_ID_BUF_LENGTH,
 120                        SERVER_ID_BUF_LENGTH);
 121         }
 122 }
 123
 124 static NTSTATUS g_lock_store(
 125         struct db_record *rec,
 126         struct g_lock *lck,
 127         struct server_id *new_shared,
 128         const TDB_DATA *new_dbufs,
 129         size_t num_new_dbufs)
 130 {
 131         uint8_t exclusive[SERVER_ID_BUF_LENGTH];
 132         uint8_t seqnum_buf[sizeof(uint64_t)];
 133         uint8_t sizebuf[sizeof(uint32_t)];
 134         uint8_t new_shared_buf[SERVER_ID_BUF_LENGTH];
 135
 136         struct TDB_DATA dbufs[6 + num_new_dbufs];
 137
 138         dbufs[0] = (TDB_DATA) {
 139                 .dptr = exclusive, .dsize = sizeof(exclusive),
 140         };
 141         dbufs[1] = (TDB_DATA) {
 142                 .dptr = seqnum_buf, .dsize = sizeof(seqnum_buf),
 143         };
 144         dbufs[2] = (TDB_DATA) {
 145                 .dptr = sizebuf, .dsize = sizeof(sizebuf),
 146         };
 147         dbufs[3] = (TDB_DATA) {
 148                 .dptr = lck->shared,
 149                 .dsize = lck->num_shared * SERVER_ID_BUF_LENGTH,
 150         };
 151         dbufs[4] = (TDB_DATA) { 0 };
 152         dbufs[5] = (TDB_DATA) {
 153                 .dptr = lck->data, .dsize = lck->datalen,
 154         };
 155
 156         if (num_new_dbufs != 0) {
 157                 memcpy(&dbufs[6],
 158                        new_dbufs,
 159                        num_new_dbufs * sizeof(TDB_DATA));
 160         }
 161
 162         server_id_put(exclusive, lck->exclusive);
 163         SBVAL(seqnum_buf, 0, lck->unique_data_epoch);
 164
 165         if (new_shared != NULL) {
 166                 if (lck->num_shared >= UINT32_MAX) {
 167                         return NT_STATUS_BUFFER_OVERFLOW;
 168                 }
 169
 170                 server_id_put(new_shared_buf, *new_shared);
 171
 172                 dbufs[4] = (TDB_DATA) {
 173                         .dptr = new_shared_buf,
 174                         .dsize = sizeof(new_shared_buf),
 175                 };
 176
 177                 lck->num_shared += 1;
 178         }
 179
 180         SIVAL(sizebuf, 0, lck->num_shared);
 181
 182         return dbwrap_record_storev(rec, dbufs, ARRAY_SIZE(dbufs), 0);
 183 }
 184
 185 struct g_lock_ctx *g_lock_ctx_init_backend(
 186         TALLOC_CTX *mem_ctx,
 187         struct messaging_context *msg,
 188         struct db_context **backend)
 189 {
 190         struct g_lock_ctx *result;
 191
 192         result = talloc(mem_ctx, struct g_lock_ctx);
 193         if (result == NULL) {
 194                 return NULL;
 195         }
 196         result->msg = msg;
 197         result->lock_order = DBWRAP_LOCK_ORDER_NONE;
 198
 199         result->db = db_open_watched(result, backend, msg);
 200         if (result->db == NULL) {
 201                 DBG_WARNING("db_open_watched failed\n");
 202                 TALLOC_FREE(result);
 203                 return NULL;
 204         }
 205         return result;
 206 }
 207
 208 void g_lock_set_lock_order(struct g_lock_ctx *ctx,
 209                            enum dbwrap_lock_order lock_order)
 210 {
 211         ctx->lock_order = lock_order;
 212 }
 213
 214 struct g_lock_ctx *g_lock_ctx_init(TALLOC_CTX *mem_ctx,
 215                                    struct messaging_context *msg)
 216 {
 217         char *db_path = NULL;
 218         struct db_context *backend = NULL;
 219         struct g_lock_ctx *ctx = NULL;
 220
 221         db_path = lock_path(mem_ctx, "g_lock.tdb");
 222         if (db_path == NULL) {
 223                 return NULL;
 224         }
 225
 226         backend = db_open(
 227                 mem_ctx,
 228                 db_path,
 229                 0,
 230                 TDB_CLEAR_IF_FIRST|TDB_INCOMPATIBLE_HASH,
 231                 O_RDWR|O_CREAT,
 232                 0600,
 233                 DBWRAP_LOCK_ORDER_3,
 234                 DBWRAP_FLAG_NONE);
 235         TALLOC_FREE(db_path);
 236         if (backend == NULL) {
 237                 DBG_WARNING("Could not open g_lock.tdb\n");
 238                 return NULL;
 239         }
 240
 241         ctx = g_lock_ctx_init_backend(mem_ctx, msg, &backend);
 242         return ctx;
 243 }
 244
 245 static NTSTATUS g_lock_cleanup_dead(
 246         struct db_record *rec,
 247         struct g_lock *lck,
 248         struct server_id *dead_blocker)
 249 {
 250         bool modified = false;
 251         bool exclusive_died;
 252         NTSTATUS status = NT_STATUS_OK;
 253         struct server_id_buf tmp;
 254
 255         if (dead_blocker == NULL) {
 256                 return NT_STATUS_OK;
 257         }
 258
 259         exclusive_died = server_id_equal(dead_blocker, &lck->exclusive);
 260
 261         if (exclusive_died) {
 262                 DBG_DEBUG("Exclusive holder %s died\n",
 263                           server_id_str_buf(lck->exclusive, &tmp));
 264                 lck->exclusive.pid = 0;
 265                 modified = true;
 266         }
 267
 268         if (lck->num_shared != 0) {
 269                 bool shared_died;
 270                 struct server_id shared;
 271
 272                 g_lock_get_shared(lck, 0, &shared);
 273                 shared_died = server_id_equal(dead_blocker, &shared);
 274
 275                 if (shared_died) {
 276                         DBG_DEBUG("Shared holder %s died\n",
 277                                   server_id_str_buf(shared, &tmp));
 278                         g_lock_del_shared(lck, 0);
 279                         modified = true;
 280                 }
 281         }
 282
 283         if (modified) {
 284                 status = g_lock_store(rec, lck, NULL, NULL, 0);
 285                 if (!NT_STATUS_IS_OK(status)) {
 286                         DBG_DEBUG("g_lock_store() failed: %s\n",
 287                                   nt_errstr(status));
 288                 }
 289         }
 290
 291         return status;
 292 }
 293
 294 static ssize_t g_lock_find_shared(
 295         struct g_lock *lck,
 296         const struct server_id *self)
 297 {
 298         size_t i;
 299
 300         for (i=0; i<lck->num_shared; i++) {
 301                 struct server_id shared;
 302                 bool same;
 303
 304                 g_lock_get_shared(lck, i, &shared);
 305
 306                 same = server_id_equal(self, &shared);
 307                 if (same) {
 308                         return i;
 309                 }
 310         }
 311
 312         return -1;
 313 }
 314
 315 static void g_lock_cleanup_shared(struct g_lock *lck)
 316 {
 317         size_t i;
 318         struct server_id check;
 319         bool exists;
 320
 321         if (lck->num_shared == 0) {
 322                 return;
 323         }
 324
 325         /*
 326          * Read locks can stay around forever if the process dies. Do
 327          * a heuristic check for process existence: Check one random
 328          * process for existence. Hopefully this will keep runaway
 329          * read locks under control.
 330          */
 331         i = generate_random() % lck->num_shared;
 332         g_lock_get_shared(lck, i, &check);
 333
 334         exists = serverid_exists(&check);
 335         if (!exists) {
 336                 struct server_id_buf tmp;
 337                 DBG_DEBUG("Shared locker %s died -- removing\n",
 338                           server_id_str_buf(check, &tmp));
 339                 g_lock_del_shared(lck, i);
 340         }
 341 }
 342
 343 struct g_lock_lock_state {
 344         struct tevent_context *ev;
 345         struct g_lock_ctx *ctx;
 346         TDB_DATA key;
 347         enum g_lock_type type;
 348         bool retry;
 349 };
 350
 351 struct g_lock_lock_fn_state {
 352         struct g_lock_lock_state *req_state;
 353         struct server_id *dead_blocker;
 354
 355         struct tevent_req *watch_req;
 356         NTSTATUS status;
 357 };
 358
 359 static int g_lock_lock_state_destructor(struct g_lock_lock_state *s);
 360
 361 static NTSTATUS g_lock_trylock(
 362         struct db_record *rec,
 363         struct g_lock_lock_fn_state *state,
 364         TDB_DATA data,
 365         struct server_id *blocker)
 366 {
 367         struct g_lock_lock_state *req_state = state->req_state;
 368         struct server_id self = messaging_server_id(req_state->ctx->msg);
 369         enum g_lock_type type = req_state->type;
 370         bool retry = req_state->retry;
 371         struct g_lock lck = { .exclusive.pid = 0 };
 372         struct server_id_buf tmp;
 373         NTSTATUS status;
 374         bool ok;
 375
 376         ok = g_lock_parse(data.dptr, data.dsize, &lck);
 377         if (!ok) {
 378                 DBG_DEBUG("g_lock_parse failed\n");
 379                 return NT_STATUS_INTERNAL_DB_CORRUPTION;
 380         }
 381
 382         status = g_lock_cleanup_dead(rec, &lck, state->dead_blocker);
 383         if (!NT_STATUS_IS_OK(status)) {
 384                 DBG_DEBUG("g_lock_cleanup_dead() failed: %s\n",
 385                           nt_errstr(status));
 386                 return status;
 387         }
 388
 389         if (lck.exclusive.pid != 0) {
 390                 bool self_exclusive = server_id_equal(&self, &lck.exclusive);
 391
 392                 if (!self_exclusive) {
 393                         bool exists = serverid_exists(&lck.exclusive);
 394                         if (!exists) {
 395                                 lck.exclusive = (struct server_id) { .pid=0 };
 396                                 goto noexclusive;
 397                         }
 398
 399                         DBG_DEBUG("%s has an exclusive lock\n",
 400                                   server_id_str_buf(lck.exclusive, &tmp));
 401
 402                         if (type == G_LOCK_DOWNGRADE) {
 403                                 struct server_id_buf tmp2;
 404                                 DBG_DEBUG("%s: Trying to downgrade %s\n",
 405                                           server_id_str_buf(self, &tmp),
 406                                           server_id_str_buf(
 407                                                   lck.exclusive, &tmp2));
 408                                 return NT_STATUS_NOT_LOCKED;
 409                         }
 410
 411                         if (type == G_LOCK_UPGRADE) {
 412                                 ssize_t shared_idx;
 413                                 shared_idx = g_lock_find_shared(&lck, &self);
 414
 415                                 if (shared_idx == -1) {
 416                                         DBG_DEBUG("Trying to upgrade %s "
 417                                                   "without "
 418                                                   "existing shared lock\n",
 419                                                   server_id_str_buf(
 420                                                           self, &tmp));
 421                                         return NT_STATUS_NOT_LOCKED;
 422                                 }
 423
 424                                 /*
 425                                  * We're trying to upgrade, and the
 426                                  * exlusive lock is taken by someone
 427                                  * else. This means that someone else
 428                                  * is waiting for us to give up our
 429                                  * shared lock. If we now also wait
 430                                  * for someone to give their shared
 431                                  * lock, we will deadlock.
 432                                  */
 433
 434                                 DBG_DEBUG("Trying to upgrade %s while "
 435                                           "someone else is also "
 436                                           "trying to upgrade\n",
 437                                           server_id_str_buf(self, &tmp));
 438                                 return NT_STATUS_POSSIBLE_DEADLOCK;
 439                         }
 440
 441                         DBG_DEBUG("Waiting for lck.exclusive=%s\n",
 442                                   server_id_str_buf(lck.exclusive, &tmp));
 443
 444                         *blocker = lck.exclusive;
 445                         return NT_STATUS_LOCK_NOT_GRANTED;
 446                 }
 447
 448                 if (type == G_LOCK_DOWNGRADE) {
 449                         DBG_DEBUG("Downgrading %s from WRITE to READ\n",
 450                                   server_id_str_buf(self, &tmp));
 451
 452                         lck.exclusive = (struct server_id) { .pid = 0 };
 453                         goto do_shared;
 454                 }
 455
 456                 if (!retry) {
 457                         DBG_DEBUG("%s already locked by self\n",
 458                                   server_id_str_buf(self, &tmp));
 459                         return NT_STATUS_WAS_LOCKED;
 460                 }
 461
 462                 if (lck.num_shared != 0) {
 463                         g_lock_get_shared(&lck, 0, blocker);
 464
 465                         DBG_DEBUG("Continue waiting for shared lock %s\n",
 466                                   server_id_str_buf(*blocker, &tmp));
 467
 468                         return NT_STATUS_LOCK_NOT_GRANTED;
 469                 }
 470
 471                 talloc_set_destructor(req_state, NULL);
 472
 473                 /*
 474                  * Retry after a conflicting lock was released
 475                  */
 476                 return NT_STATUS_OK;
 477         }
 478
 479 noexclusive:
 480
 481         if (type == G_LOCK_UPGRADE) {
 482                 ssize_t shared_idx = g_lock_find_shared(&lck, &self);
 483
 484                 if (shared_idx == -1) {
 485                         DBG_DEBUG("Trying to upgrade %s without "
 486                                   "existing shared lock\n",
 487                                   server_id_str_buf(self, &tmp));
 488                         return NT_STATUS_NOT_LOCKED;
 489                 }
 490
 491                 g_lock_del_shared(&lck, shared_idx);
 492                 type = G_LOCK_WRITE;
 493         }
 494
 495         if (type == G_LOCK_WRITE) {
 496                 ssize_t shared_idx = g_lock_find_shared(&lck, &self);
 497
 498                 if (shared_idx != -1) {
 499                         DBG_DEBUG("Trying to writelock existing shared %s\n",
 500                                   server_id_str_buf(self, &tmp));
 501                         return NT_STATUS_WAS_LOCKED;
 502                 }
 503
 504                 lck.exclusive = self;
 505
 506                 status = g_lock_store(rec, &lck, NULL, NULL, 0);
 507                 if (!NT_STATUS_IS_OK(status)) {
 508                         DBG_DEBUG("g_lock_store() failed: %s\n",
 509                                   nt_errstr(status));
 510                         return status;
 511                 }
 512
 513                 if (lck.num_shared != 0) {
 514                         talloc_set_destructor(
 515                                 req_state, g_lock_lock_state_destructor);
 516
 517                         g_lock_get_shared(&lck, 0, blocker);
 518
 519                         DBG_DEBUG("Waiting for %zu shared locks, "
 520                                   "picking blocker %s\n",
 521                                   lck.num_shared,
 522                                   server_id_str_buf(*blocker, &tmp));
 523
 524                         return NT_STATUS_LOCK_NOT_GRANTED;
 525                 }
 526
 527                 talloc_set_destructor(req_state, NULL);
 528
 529                 return NT_STATUS_OK;
 530         }
 531
 532 do_shared:
 533
 534         if (lck.num_shared == 0) {
 535                 status = g_lock_store(rec, &lck, &self, NULL, 0);
 536                 if (!NT_STATUS_IS_OK(status)) {
 537                         DBG_DEBUG("g_lock_store() failed: %s\n",
 538                                   nt_errstr(status));
 539                 }
 540
 541                 return status;
 542         }
 543
 544         g_lock_cleanup_shared(&lck);
 545
 546         status = g_lock_store(rec, &lck, &self, NULL, 0);
 547         if (!NT_STATUS_IS_OK(status)) {
 548                 DBG_DEBUG("g_lock_store() failed: %s\n",
 549                           nt_errstr(status));
 550                 return status;
 551         }
 552
 553         return NT_STATUS_OK;
 554 }
 555
 556 static void g_lock_lock_fn(
 557         struct db_record *rec,
 558         TDB_DATA value,
 559         void *private_data)
 560 {
 561         struct g_lock_lock_fn_state *state = private_data;
 562         struct server_id blocker = {0};
 563
 564         state->status = g_lock_trylock(rec, state, value, &blocker);
 565         if (!NT_STATUS_IS_OK(state->status)) {
 566                 DBG_DEBUG("g_lock_trylock returned %s\n",
 567                           nt_errstr(state->status));
 568         }
 569         if (!NT_STATUS_EQUAL(state->status, NT_STATUS_LOCK_NOT_GRANTED)) {
 570                 return;
 571         }
 572
 573         state->watch_req = dbwrap_watched_watch_send(
 574                 state->req_state, state->req_state->ev, rec, blocker);
 575         if (state->watch_req == NULL) {
 576                 state->status = NT_STATUS_NO_MEMORY;
 577         }
 578 }
 579
 580 static int g_lock_lock_state_destructor(struct g_lock_lock_state *s)
 581 {
 582         NTSTATUS status = g_lock_unlock(s->ctx, s->key);
 583         if (!NT_STATUS_IS_OK(status)) {
 584                 DBG_DEBUG("g_lock_unlock failed: %s\n", nt_errstr(status));
 585         }
 586         return 0;
 587 }
 588
 589 static void g_lock_lock_retry(struct tevent_req *subreq);
 590
 591 struct tevent_req *g_lock_lock_send(TALLOC_CTX *mem_ctx,
 592                                     struct tevent_context *ev,
 593                                     struct g_lock_ctx *ctx,
 594                                     TDB_DATA key,
 595                                     enum g_lock_type type)
 596 {
 597         struct tevent_req *req;
 598         struct g_lock_lock_state *state;
 599         struct g_lock_lock_fn_state fn_state;
 600         NTSTATUS status;
 601         bool ok;
 602
 603         req = tevent_req_create(mem_ctx, &state, struct g_lock_lock_state);
 604         if (req == NULL) {
 605                 return NULL;
 606         }
 607         state->ev = ev;
 608         state->ctx = ctx;
 609         state->key = key;
 610         state->type = type;
 611
 612         fn_state = (struct g_lock_lock_fn_state) {
 613                 .req_state = state,
 614         };
 615
 616         status = dbwrap_do_locked(ctx->db, key, g_lock_lock_fn, &fn_state);
 617         if (tevent_req_nterror(req, status)) {
 618                 DBG_DEBUG("dbwrap_do_locked failed: %s\n",
 619                           nt_errstr(status));
 620                 return tevent_req_post(req, ev);
 621         }
 622
 623         if (NT_STATUS_IS_OK(fn_state.status)) {
 624                 tevent_req_done(req);
 625                 return tevent_req_post(req, ev);
 626         }
 627         if (!NT_STATUS_EQUAL(fn_state.status, NT_STATUS_LOCK_NOT_GRANTED)) {
 628                 tevent_req_nterror(req, fn_state.status);
 629                 return tevent_req_post(req, ev);
 630         }
 631
 632         if (tevent_req_nomem(fn_state.watch_req, req)) {
 633                 return tevent_req_post(req, ev);
 634         }
 635
 636         ok = tevent_req_set_endtime(
 637                 fn_state.watch_req,
 638                 state->ev,
 639                 timeval_current_ofs(5 + generate_random() % 5, 0));
 640         if (!ok) {
 641                 tevent_req_oom(req);
 642                 return tevent_req_post(req, ev);
 643         }
 644         tevent_req_set_callback(fn_state.watch_req, g_lock_lock_retry, req);
 645
 646         return req;
 647 }
 648
 649 static void g_lock_lock_retry(struct tevent_req *subreq)
 650 {
 651         struct tevent_req *req = tevent_req_callback_data(
 652                 subreq, struct tevent_req);
 653         struct g_lock_lock_state *state = tevent_req_data(
 654                 req, struct g_lock_lock_state);
 655         struct g_lock_lock_fn_state fn_state;
 656         struct server_id blocker = { .pid = 0 };
 657         bool blockerdead = false;
 658         NTSTATUS status;
 659
 660         status = dbwrap_watched_watch_recv(subreq, &blockerdead, &blocker);
 661         DBG_DEBUG("watch_recv returned %s\n", nt_errstr(status));
 662         TALLOC_FREE(subreq);
 663
 664         if (!NT_STATUS_IS_OK(status) &&
 665             !NT_STATUS_EQUAL(status, NT_STATUS_IO_TIMEOUT)) {
 666                 tevent_req_nterror(req, status);
 667                 return;
 668         }
 669
 670         state->retry = true;
 671
 672         fn_state = (struct g_lock_lock_fn_state) {
 673                 .req_state = state,
 674                 .dead_blocker = blockerdead ? &blocker : NULL,
 675         };
 676
 677         status = dbwrap_do_locked(state->ctx->db, state->key,
 678                                   g_lock_lock_fn, &fn_state);
 679         if (tevent_req_nterror(req, status)) {
 680                 DBG_DEBUG("dbwrap_do_locked failed: %s\n",
 681                           nt_errstr(status));
 682                 return;
 683         }
 684
 685         if (NT_STATUS_IS_OK(fn_state.status)) {
 686                 tevent_req_done(req);
 687                 return;
 688         }
 689         if (!NT_STATUS_EQUAL(fn_state.status, NT_STATUS_LOCK_NOT_GRANTED)) {
 690                 tevent_req_nterror(req, fn_state.status);
 691                 return;
 692         }
 693
 694         if (tevent_req_nomem(fn_state.watch_req, req)) {
 695                 return;
 696         }
 697
 698         if (!tevent_req_set_endtime(
 699                     fn_state.watch_req, state->ev,
 700                     timeval_current_ofs(5 + generate_random() % 5, 0))) {
 701                 return;
 702         }
 703         tevent_req_set_callback(fn_state.watch_req, g_lock_lock_retry, req);
 704 }
 705
 706 NTSTATUS g_lock_lock_recv(struct tevent_req *req)
 707 {
 708         struct g_lock_lock_state *state = tevent_req_data(
 709                 req, struct g_lock_lock_state);
 710         struct g_lock_ctx *ctx = state->ctx;
 711         NTSTATUS status;
 712
 713         if (tevent_req_is_nterror(req, &status)) {
 714                 return status;
 715         }
 716
 717         if ((ctx->lock_order != DBWRAP_LOCK_ORDER_NONE) &&
 718             ((state->type == G_LOCK_READ) ||
 719              (state->type == G_LOCK_WRITE))) {
 720                 const char *name = dbwrap_name(ctx->db);
 721                 dbwrap_lock_order_lock(name, ctx->lock_order);
 722         }
 723
 724         return NT_STATUS_OK;
 725 }
 726
 727 struct g_lock_lock_simple_state {
 728         struct server_id me;
 729         enum g_lock_type type;
 730         NTSTATUS status;
 731 };
 732
 733 static void g_lock_lock_simple_fn(
 734         struct db_record *rec,
 735         TDB_DATA value,
 736         void *private_data)
 737 {
 738         struct g_lock_lock_simple_state *state = private_data;
 739         struct server_id_buf buf;
 740         struct g_lock lck = { .exclusive.pid = 0 };
 741         bool ok;
 742
 743         ok = g_lock_parse(value.dptr, value.dsize, &lck);
 744         if (!ok) {
 745                 DBG_DEBUG("g_lock_parse failed\n");
 746                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 747                 return;
 748         }
 749
 750         if (lck.exclusive.pid != 0) {
 751                 DBG_DEBUG("locked by %s\n",
 752                           server_id_str_buf(lck.exclusive, &buf));
 753                 goto not_granted;
 754         }
 755
 756         if (state->type == G_LOCK_WRITE) {
 757                 if (lck.num_shared != 0) {
 758                         DBG_DEBUG("num_shared=%zu\n", lck.num_shared);
 759                         goto not_granted;
 760                 }
 761                 lck.exclusive = state->me;
 762                 state->status = g_lock_store(rec, &lck, NULL, NULL, 0);
 763                 return;
 764         }
 765
 766         if (state->type == G_LOCK_READ) {
 767                 g_lock_cleanup_shared(&lck);
 768                 state->status = g_lock_store(rec, &lck, &state->me, NULL, 0);
 769                 return;
 770         }
 771
 772 not_granted:
 773         state->status = NT_STATUS_LOCK_NOT_GRANTED;
 774 }
 775
 776 NTSTATUS g_lock_lock(struct g_lock_ctx *ctx, TDB_DATA key,
 777                      enum g_lock_type type, struct timeval timeout)
 778 {
 779         TALLOC_CTX *frame;
 780         struct tevent_context *ev;
 781         struct tevent_req *req;
 782         struct timeval end;
 783         NTSTATUS status;
 784
 785         if ((type == G_LOCK_READ) || (type == G_LOCK_WRITE)) {
 786                 /*
 787                  * This is an abstraction violation: Normally we do
 788                  * the sync wrappers around async functions with full
 789                  * nested event contexts. However, this is used in
 790                  * very hot code paths, so avoid the event context
 791                  * creation for the good path where there's no lock
 792                  * contention. My benchmark gave a factor of 2
 793                  * improvement for lock/unlock.
 794                  */
 795                 struct g_lock_lock_simple_state state = {
 796                         .me = messaging_server_id(ctx->msg),
 797                         .type = type,
 798                 };
 799                 status = dbwrap_do_locked(
 800                         ctx->db, key, g_lock_lock_simple_fn, &state);
 801                 if (!NT_STATUS_IS_OK(status)) {
 802                         DBG_DEBUG("dbwrap_do_locked() failed: %s\n",
 803                                   nt_errstr(status));
 804                         return status;
 805                 }
 806
 807                 DBG_DEBUG("status=%s, state.status=%s\n",
 808                           nt_errstr(status),
 809                           nt_errstr(state.status));
 810
 811                 if (NT_STATUS_IS_OK(state.status)) {
 812                         if (ctx->lock_order != DBWRAP_LOCK_ORDER_NONE) {
 813                                 const char *name = dbwrap_name(ctx->db);
 814                                 dbwrap_lock_order_lock(name, ctx->lock_order);
 815                         }
 816                         return NT_STATUS_OK;
 817                 }
 818                 if (!NT_STATUS_EQUAL(
 819                             state.status, NT_STATUS_LOCK_NOT_GRANTED)) {
 820                         return state.status;
 821                 }
 822
 823                 /*
 824                  * Fall back to the full g_lock_trylock logic,
 825                  * g_lock_lock_simple_fn() called above only covers
 826                  * the uncontended path.
 827                  */
 828         }
 829
 830         frame = talloc_stackframe();
 831         status = NT_STATUS_NO_MEMORY;
 832
 833         ev = samba_tevent_context_init(frame);
 834         if (ev == NULL) {
 835                 goto fail;
 836         }
 837         req = g_lock_lock_send(frame, ev, ctx, key, type);
 838         if (req == NULL) {
 839                 goto fail;
 840         }
 841         end = timeval_current_ofs(timeout.tv_sec, timeout.tv_usec);
 842         if (!tevent_req_set_endtime(req, ev, end)) {
 843                 goto fail;
 844         }
 845         if (!tevent_req_poll_ntstatus(req, ev, &status)) {
 846                 goto fail;
 847         }
 848         status = g_lock_lock_recv(req);
 849  fail:
 850         TALLOC_FREE(frame);
 851         return status;
 852 }
 853
 854 struct g_lock_unlock_state {
 855         struct server_id self;
 856         NTSTATUS status;
 857 };
 858
 859 static void g_lock_unlock_fn(
 860         struct db_record *rec,
 861         TDB_DATA value,
 862         void *private_data)
 863 {
 864         struct g_lock_unlock_state *state = private_data;
 865         struct server_id_buf tmp1, tmp2;
 866         struct g_lock lck;
 867         size_t i;
 868         bool ok, exclusive;
 869
 870         ok = g_lock_parse(value.dptr, value.dsize, &lck);
 871         if (!ok) {
 872                 DBG_DEBUG("g_lock_parse() failed\n");
 873                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 874                 return;
 875         }
 876
 877         exclusive = server_id_equal(&state->self, &lck.exclusive);
 878
 879         for (i=0; i<lck.num_shared; i++) {
 880                 struct server_id shared;
 881                 g_lock_get_shared(&lck, i, &shared);
 882                 if (server_id_equal(&state->self, &shared)) {
 883                         break;
 884                 }
 885         }
 886
 887         if (i < lck.num_shared) {
 888                 if (exclusive) {
 889                         DBG_DEBUG("%s both exclusive and shared (%zu)\n",
 890                                   server_id_str_buf(state->self, &tmp1),
 891                                   i);
 892                         state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 893                         return;
 894                 }
 895                 g_lock_del_shared(&lck, i);
 896         } else {
 897                 if (!exclusive) {
 898                         DBG_DEBUG("Lock not found, self=%s, lck.exclusive=%s, "
 899                                   "num_shared=%zu\n",
 900                                   server_id_str_buf(state->self, &tmp1),
 901                                   server_id_str_buf(lck.exclusive, &tmp2),
 902                                   lck.num_shared);
 903                         state->status = NT_STATUS_NOT_FOUND;
 904                         return;
 905                 }
 906                 lck.exclusive = (struct server_id) { .pid = 0 };
 907         }
 908
 909         if ((lck.exclusive.pid == 0) &&
 910             (lck.num_shared == 0) &&
 911             (lck.datalen == 0)) {
 912                 state->status = dbwrap_record_delete(rec);
 913                 return;
 914         }
 915
 916         state->status = g_lock_store(rec, &lck, NULL, NULL, 0);
 917 }
 918
 919 NTSTATUS g_lock_unlock(struct g_lock_ctx *ctx, TDB_DATA key)
 920 {
 921         struct g_lock_unlock_state state = {
 922                 .self = messaging_server_id(ctx->msg),
 923         };
 924         NTSTATUS status;
 925
 926         status = dbwrap_do_locked(ctx->db, key, g_lock_unlock_fn, &state);
 927         if (!NT_STATUS_IS_OK(status)) {
 928                 DBG_WARNING("dbwrap_do_locked failed: %s\n",
 929                             nt_errstr(status));
 930                 return status;
 931         }
 932         if (!NT_STATUS_IS_OK(state.status)) {
 933                 DBG_WARNING("g_lock_unlock_fn failed: %s\n",
 934                             nt_errstr(state.status));
 935                 return state.status;
 936         }
 937
 938         if (ctx->lock_order != DBWRAP_LOCK_ORDER_NONE) {
 939                 const char *name = dbwrap_name(ctx->db);
 940                 dbwrap_lock_order_unlock(name, ctx->lock_order);
 941         }
 942
 943         return NT_STATUS_OK;
 944 }
 945
 946 struct g_lock_writev_data_state {
 947         TDB_DATA key;
 948         struct server_id self;
 949         const TDB_DATA *dbufs;
 950         size_t num_dbufs;
 951         NTSTATUS status;
 952 };
 953
 954 static void g_lock_writev_data_fn(
 955         struct db_record *rec,
 956         TDB_DATA value,
 957         void *private_data)
 958 {
 959         struct g_lock_writev_data_state *state = private_data;
 960         struct g_lock lck;
 961         bool exclusive;
 962         bool ok;
 963
 964         ok = g_lock_parse(value.dptr, value.dsize, &lck);
 965         if (!ok) {
 966                 DBG_DEBUG("g_lock_parse for %s failed\n",
 967                           hex_encode_talloc(talloc_tos(),
 968                                             state->key.dptr,
 969                                             state->key.dsize));
 970                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 971                 return;
 972         }
 973
 974         exclusive = server_id_equal(&state->self, &lck.exclusive);
 975
 976         /*
 977          * Make sure we're really exclusive. We are marked as
 978          * exclusive when we are waiting for an exclusive lock
 979          */
 980         exclusive &= (lck.num_shared == 0);
 981
 982         if (!exclusive) {
 983                 struct server_id_buf buf1, buf2;
 984                 DBG_DEBUG("Not locked by us: self=%s, lck.exclusive=%s, "
 985                           "lck.num_shared=%zu\n",
 986                           server_id_str_buf(state->self, &buf1),
 987                           server_id_str_buf(lck.exclusive, &buf2),
 988                           lck.num_shared);
 989                 state->status = NT_STATUS_NOT_LOCKED;
 990                 return;
 991         }
 992
 993         lck.unique_data_epoch = generate_unique_u64(lck.unique_data_epoch);
 994         lck.data = NULL;
 995         lck.datalen = 0;
 996         state->status = g_lock_store(
 997                 rec, &lck, NULL, state->dbufs, state->num_dbufs);
 998 }
 999
1000 NTSTATUS g_lock_writev_data(
1001         struct g_lock_ctx *ctx,
1002         TDB_DATA key,
1003         const TDB_DATA *dbufs,
1004         size_t num_dbufs)
1005 {
1006         struct g_lock_writev_data_state state = {
1007                 .key = key,
1008                 .self = messaging_server_id(ctx->msg),
1009                 .dbufs = dbufs,
1010                 .num_dbufs = num_dbufs,
1011         };
1012         NTSTATUS status;
1013
1014         status = dbwrap_do_locked(
1015                 ctx->db, key, g_lock_writev_data_fn, &state);
1016         if (!NT_STATUS_IS_OK(status)) {
1017                 DBG_WARNING("dbwrap_do_locked failed: %s\n",
1018                             nt_errstr(status));
1019                 return status;
1020         }
1021         if (!NT_STATUS_IS_OK(state.status)) {
1022                 DBG_WARNING("g_lock_writev_data_fn failed: %s\n",
1023                             nt_errstr(state.status));
1024                 return state.status;
1025         }
1026
1027         return NT_STATUS_OK;
1028 }
1029
1030 NTSTATUS g_lock_write_data(struct g_lock_ctx *ctx, TDB_DATA key,
1031                            const uint8_t *buf, size_t buflen)
1032 {
1033         TDB_DATA dbuf = {
1034                 .dptr = discard_const_p(uint8_t, buf),
1035                 .dsize = buflen,
1036         };
1037         return g_lock_writev_data(ctx, key, &dbuf, 1);
1038 }
1039
1040 struct g_lock_locks_state {
1041         int (*fn)(TDB_DATA key, void *private_data);
1042         void *private_data;
1043 };
1044
1045 static int g_lock_locks_fn(struct db_record *rec, void *priv)
1046 {
1047         TDB_DATA key;
1048         struct g_lock_locks_state *state = (struct g_lock_locks_state *)priv;
1049
1050         key = dbwrap_record_get_key(rec);
1051         return state->fn(key, state->private_data);
1052 }
1053
1054 int g_lock_locks(struct g_lock_ctx *ctx,
1055                  int (*fn)(TDB_DATA key, void *private_data),
1056                  void *private_data)
1057 {
1058         struct g_lock_locks_state state;
1059         NTSTATUS status;
1060         int count;
1061
1062         state.fn = fn;
1063         state.private_data = private_data;
1064
1065         status = dbwrap_traverse_read(ctx->db, g_lock_locks_fn, &state, &count);
1066         if (!NT_STATUS_IS_OK(status)) {
1067                 return -1;
1068         }
1069         return count;
1070 }
1071
1072 struct g_lock_dump_state {
1073         TALLOC_CTX *mem_ctx;
1074         TDB_DATA key;
1075         void (*fn)(struct server_id exclusive,
1076                    size_t num_shared,
1077                    struct server_id *shared,
1078                    const uint8_t *data,
1079                    size_t datalen,
1080                    void *private_data);
1081         void *private_data;
1082         NTSTATUS status;
1083         enum dbwrap_req_state req_state;
1084 };
1085
1086 static void g_lock_dump_fn(TDB_DATA key, TDB_DATA data,
1087                            void *private_data)
1088 {
1089         struct g_lock_dump_state *state = private_data;
1090         struct g_lock lck = (struct g_lock) { .exclusive.pid = 0 };
1091         struct server_id *shared = NULL;
1092         size_t i;
1093         bool ok;
1094
1095         ok = g_lock_parse(data.dptr, data.dsize, &lck);
1096         if (!ok) {
1097                 DBG_DEBUG("g_lock_parse failed for %s\n",
1098                           hex_encode_talloc(talloc_tos(),
1099                                             state->key.dptr,
1100                                             state->key.dsize));
1101                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
1102                 return;
1103         }
1104
1105         shared = talloc_array(
1106                 state->mem_ctx, struct server_id, lck.num_shared);
1107         if (shared == NULL) {
1108                 DBG_DEBUG("talloc failed\n");
1109                 state->status = NT_STATUS_NO_MEMORY;
1110                 return;
1111         }
1112
1113         for (i=0; i<lck.num_shared; i++) {
1114                 g_lock_get_shared(&lck, i, &shared[i]);
1115         }
1116
1117         state->fn(lck.exclusive,
1118                   lck.num_shared,
1119                   shared,
1120                   lck.data,
1121                   lck.datalen,
1122                   state->private_data);
1123
1124         TALLOC_FREE(shared);
1125
1126         state->status = NT_STATUS_OK;
1127 }
1128
1129 NTSTATUS g_lock_dump(struct g_lock_ctx *ctx, TDB_DATA key,
1130                      void (*fn)(struct server_id exclusive,
1131                                 size_t num_shared,
1132                                 struct server_id *shared,
1133                                 const uint8_t *data,
1134                                 size_t datalen,
1135                                 void *private_data),
1136                      void *private_data)
1137 {
1138         struct g_lock_dump_state state = {
1139                 .mem_ctx = ctx, .key = key,
1140                 .fn = fn, .private_data = private_data
1141         };
1142         NTSTATUS status;
1143
1144         status = dbwrap_parse_record(ctx->db, key, g_lock_dump_fn, &state);
1145         if (!NT_STATUS_IS_OK(status)) {
1146                 DBG_DEBUG("dbwrap_parse_record returned %s\n",
1147                           nt_errstr(status));
1148                 return status;
1149         }
1150         if (!NT_STATUS_IS_OK(state.status)) {
1151                 DBG_DEBUG("g_lock_dump_fn returned %s\n",
1152                           nt_errstr(state.status));
1153                 return state.status;
1154         }
1155         return NT_STATUS_OK;
1156 }
1157
1158 static void g_lock_dump_done(struct tevent_req *subreq);
1159
1160 struct tevent_req *g_lock_dump_send(
1161         TALLOC_CTX *mem_ctx,
1162         struct tevent_context *ev,
1163         struct g_lock_ctx *ctx,
1164         TDB_DATA key,
1165         void (*fn)(struct server_id exclusive,
1166                    size_t num_shared,
1167                    struct server_id *shared,
1168                    const uint8_t *data,
1169                    size_t datalen,
1170                    void *private_data),
1171         void *private_data)
1172 {
1173         struct tevent_req *req = NULL, *subreq = NULL;
1174         struct g_lock_dump_state *state = NULL;
1175
1176         req = tevent_req_create(mem_ctx, &state, struct g_lock_dump_state);
1177         if (req == NULL) {
1178                 return NULL;
1179         }
1180         state->mem_ctx = state;
1181         state->key = key;
1182         state->fn = fn;
1183         state->private_data = private_data;
1184
1185         subreq = dbwrap_parse_record_send(
1186                 state,
1187                 ev,
1188                 ctx->db,
1189                 key,
1190                 g_lock_dump_fn,
1191                 state,
1192                 &state->req_state);
1193         if (tevent_req_nomem(subreq, req)) {
1194                 return tevent_req_post(req, ev);
1195         }
1196         tevent_req_set_callback(subreq, g_lock_dump_done, req);
1197         return req;
1198 }
1199
1200 static void g_lock_dump_done(struct tevent_req *subreq)
1201 {
1202         struct tevent_req *req = tevent_req_callback_data(
1203                 subreq, struct tevent_req);
1204         struct g_lock_dump_state *state = tevent_req_data(
1205                 req, struct g_lock_dump_state);
1206         NTSTATUS status;
1207
1208         status = dbwrap_parse_record_recv(subreq);
1209         TALLOC_FREE(subreq);
1210         if (tevent_req_nterror(req, status) ||
1211             tevent_req_nterror(req, state->status)) {
1212                 return;
1213         }
1214         tevent_req_done(req);
1215 }
1216
1217 NTSTATUS g_lock_dump_recv(struct tevent_req *req)
1218 {
1219         return tevent_req_simple_recv_ntstatus(req);
1220 }
1221
1222 int g_lock_seqnum(struct g_lock_ctx *ctx)
1223 {
1224         return dbwrap_get_seqnum(ctx->db);
1225 }
1226
1227 struct g_lock_watch_data_state {
1228         struct tevent_context *ev;
1229         struct g_lock_ctx *ctx;
1230         TDB_DATA key;
1231         struct server_id blocker;
1232         bool blockerdead;
1233         uint64_t unique_data_epoch;
1234         NTSTATUS status;
1235 };
1236
1237 static void g_lock_watch_data_done(struct tevent_req *subreq);
1238
1239 static void g_lock_watch_data_send_fn(
1240         struct db_record *rec,
1241         TDB_DATA value,
1242         void *private_data)
1243 {
1244         struct tevent_req *req = talloc_get_type_abort(
1245                 private_data, struct tevent_req);
1246         struct g_lock_watch_data_state *state = tevent_req_data(
1247                 req, struct g_lock_watch_data_state);
1248         struct tevent_req *subreq = NULL;
1249         struct g_lock lck;
1250         bool ok;
1251
1252         ok = g_lock_parse(value.dptr, value.dsize, &lck);
1253         if (!ok) {
1254                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
1255                 return;
1256         }
1257         state->unique_data_epoch = lck.unique_data_epoch;
1258
1259         DBG_DEBUG("state->unique_data_epoch=%"PRIu64"\n", state->unique_data_epoch);
1260
1261         subreq = dbwrap_watched_watch_send(
1262                 state, state->ev, rec, state->blocker);
1263         if (subreq == NULL) {
1264                 state->status = NT_STATUS_NO_MEMORY;
1265                 return;
1266         }
1267         tevent_req_set_callback(subreq, g_lock_watch_data_done, req);
1268
1269         state->status = NT_STATUS_EVENT_PENDING;
1270 }
1271
1272 struct tevent_req *g_lock_watch_data_send(
1273         TALLOC_CTX *mem_ctx,
1274         struct tevent_context *ev,
1275         struct g_lock_ctx *ctx,
1276         TDB_DATA key,
1277         struct server_id blocker)
1278 {
1279         struct tevent_req *req = NULL;
1280         struct g_lock_watch_data_state *state = NULL;
1281         NTSTATUS status;
1282
1283         req = tevent_req_create(
1284                 mem_ctx, &state, struct g_lock_watch_data_state);
1285         if (req == NULL) {
1286                 return NULL;
1287         }
1288         state->ev = ev;
1289         state->ctx = ctx;
1290         state->blocker = blocker;
1291
1292         state->key = tdb_data_talloc_copy(state, key);
1293         if (tevent_req_nomem(state->key.dptr, req)) {
1294                 return tevent_req_post(req, ev);
1295         }
1296
1297         status = dbwrap_do_locked(
1298                 ctx->db, key, g_lock_watch_data_send_fn, req);
1299         if (tevent_req_nterror(req, status)) {
1300                 DBG_DEBUG("dbwrap_do_locked returned %s\n", nt_errstr(status));
1301                 return tevent_req_post(req, ev);
1302         }
1303
1304         if (NT_STATUS_IS_OK(state->status)) {
1305                 tevent_req_done(req);
1306                 return tevent_req_post(req, ev);
1307         }
1308
1309         return req;
1310 }
1311
1312 static void g_lock_watch_data_done_fn(
1313         struct db_record *rec,
1314         TDB_DATA value,
1315         void *private_data)
1316 {
1317         struct tevent_req *req = talloc_get_type_abort(
1318                 private_data, struct tevent_req);
1319         struct g_lock_watch_data_state *state = tevent_req_data(
1320                 req, struct g_lock_watch_data_state);
1321         struct tevent_req *subreq = NULL;
1322         struct g_lock lck;
1323         bool ok;
1324
1325         ok = g_lock_parse(value.dptr, value.dsize, &lck);
1326         if (!ok) {
1327                 state->status = NT_STATUS_INTERNAL_DB_CORRUPTION;
1328                 return;
1329         }
1330
1331         if (lck.unique_data_epoch != state->unique_data_epoch) {
1332                 DBG_DEBUG("lck.unique_data_epoch=%"PRIu64", "
1333                           "state->unique_data_epoch=%"PRIu64"\n",
1334                           lck.unique_data_epoch,
1335                           state->unique_data_epoch);
1336                 state->status = NT_STATUS_OK;
1337                 return;
1338         }
1339
1340         subreq = dbwrap_watched_watch_send(
1341                 state, state->ev, rec, state->blocker);
1342         if (subreq == NULL) {
1343                 state->status = NT_STATUS_NO_MEMORY;
1344                 return;
1345         }
1346         tevent_req_set_callback(subreq, g_lock_watch_data_done, req);
1347
1348         state->status = NT_STATUS_EVENT_PENDING;
1349 }
1350
1351 static void g_lock_watch_data_done(struct tevent_req *subreq)
1352 {
1353         struct tevent_req *req = tevent_req_callback_data(
1354                 subreq, struct tevent_req);
1355         struct g_lock_watch_data_state *state = tevent_req_data(
1356                 req, struct g_lock_watch_data_state);
1357         NTSTATUS status;
1358
1359         status = dbwrap_watched_watch_recv(
1360                 subreq, &state->blockerdead, &state->blocker);
1361         TALLOC_FREE(subreq);
1362         if (tevent_req_nterror(req, status)) {
1363                 DBG_DEBUG("dbwrap_watched_watch_recv returned %s\n",
1364                           nt_errstr(status));
1365                 return;
1366         }
1367
1368         status = dbwrap_do_locked(
1369                 state->ctx->db, state->key, g_lock_watch_data_done_fn, req);
1370         if (tevent_req_nterror(req, status)) {
1371                 DBG_DEBUG("dbwrap_do_locked returned %s\n", nt_errstr(status));
1372                 return;
1373         }
1374         if (NT_STATUS_EQUAL(state->status, NT_STATUS_EVENT_PENDING)) {
1375                 return;
1376         }
1377         if (tevent_req_nterror(req, state->status)) {
1378                 return;
1379         }
1380         tevent_req_done(req);
1381 }
1382
1383 NTSTATUS g_lock_watch_data_recv(
1384         struct tevent_req *req,
1385         bool *blockerdead,
1386         struct server_id *blocker)
1387 {
1388         struct g_lock_watch_data_state *state = tevent_req_data(
1389                 req, struct g_lock_watch_data_state);
1390         NTSTATUS status;
1391
1392         if (tevent_req_is_nterror(req, &status)) {
1393                 return status;
1394         }
1395         if (blockerdead != NULL) {
1396                 *blockerdead = state->blockerdead;
1397         }
1398         if (blocker != NULL) {
1399                 *blocker = state->blocker;
1400         }
1401
1402         return NT_STATUS_OK;
1403 }
1404
1405 static void g_lock_wake_watchers_fn(
1406         struct db_record *rec,
1407         TDB_DATA value,
1408         void *private_data)
1409 {
1410         struct g_lock lck = { .exclusive.pid = 0 };
1411         NTSTATUS status;
1412         bool ok;
1413
1414         ok = g_lock_parse(value.dptr, value.dsize, &lck);
1415         if (!ok) {
1416                 DBG_WARNING("g_lock_parse failed\n");
1417                 return;
1418         }
1419
1420         lck.unique_data_epoch = generate_unique_u64(lck.unique_data_epoch);
1421
1422         status = g_lock_store(rec, &lck, NULL, NULL, 0);
1423         if (!NT_STATUS_IS_OK(status)) {
1424                 DBG_WARNING("g_lock_store failed: %s\n", nt_errstr(status));
1425                 return;
1426         }
1427 }
1428
1429 void g_lock_wake_watchers(struct g_lock_ctx *ctx, TDB_DATA key)
1430 {
1431         NTSTATUS status;
1432
1433         status = dbwrap_do_locked(ctx->db, key, g_lock_wake_watchers_fn, NULL);
1434         if (!NT_STATUS_IS_OK(status)) {
1435                 DBG_DEBUG("dbwrap_do_locked returned %s\n",
1436                           nt_errstr(status));
1437         }
1438 }