source/cluster/ctdb/opendb_ctdb.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    Copyright (C) Ronnie Sahlberg 2007
   5    Copyright (C) Andrew Tridgell 2007
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 */
  20
  21 /*
  22   this is the open files database, ctdb backend. It implements shared
  23   storage of what files are open between server instances, and
  24   implements the rules of shared access to files.
  25
  26   The caller needs to provide a file_key, which specifies what file
  27   they are talking about. This needs to be a unique key across all
  28   filesystems, and is usually implemented in terms of a device/inode
  29   pair.
  30
  31   Before any operations can be performed the caller needs to establish
  32   a lock on the record associated with file_key. That is done by
  33   calling odb_lock(). The caller releases this lock by calling
  34   talloc_free() on the returned handle.
  35
  36   All other operations on a record are done by passing the odb_lock()
  37   handle back to this module. The handle contains internal
  38   information about what file_key is being operated on.
  39 */
  40
  41 #include "includes.h"
  42 #include "system/filesys.h"
  43 #include "lib/tdb/include/tdb.h"
  44 #include "messaging/messaging.h"
  45 #include "tdb_wrap.h"
  46 #include "lib/messaging/irpc.h"
  47 #include "librpc/gen_ndr/ndr_opendb.h"
  48 #include "ntvfs/ntvfs.h"
  49 #include "ntvfs/common/ntvfs_common.h"
  50 #include "cluster/cluster.h"
  51 #include "include/ctdb.h"
  52 #include "param/param.h"
  53
  54 struct odb_context {
  55         struct ctdb_context *ctdb;
  56         struct ctdb_db_context *ctdb_db;
  57         struct ntvfs_context *ntvfs_ctx;
  58         bool oplocks;
  59 };
  60
  61 /*
  62   an odb lock handle. You must obtain one of these using odb_lock() before doing
  63   any other operations.
  64 */
  65 struct odb_lock {
  66         struct odb_context *odb;
  67         struct ctdb_record_handle *rec;
  68         TDB_DATA key;
  69         TDB_DATA data;
  70 };
  71
  72 /*
  73   Open up the openfiles.tdb database. Close it down using
  74   talloc_free(). We need the messaging_ctx to allow for pending open
  75   notifications.
  76 */
  77 static struct odb_context *odb_ctdb_init(TALLOC_CTX *mem_ctx,
  78                                         struct ntvfs_context *ntvfs_ctx)
  79 {
  80         struct odb_context *odb;
  81         struct ctdb_context *ctdb = talloc_get_type(cluster_backend_handle(),
  82                                                     struct ctdb_context);
  83
  84         odb = talloc(mem_ctx, struct odb_context);
  85         if (odb == NULL) {
  86                 return NULL;
  87         }
  88
  89         odb->ctdb = ctdb;
  90         odb->ctdb_db = ctdb_attach(ctdb, "opendb");
  91         if (!odb->ctdb_db) {
  92                 DEBUG(0,("Failed to get attached ctdb db handle for opendb\n"));
  93                 talloc_free(odb);
  94                 return NULL;
  95         }
  96
  97         odb->ntvfs_ctx = ntvfs_ctx;
  98
  99         /* leave oplocks disabled by default until the code is working */
 100         odb->oplocks = lp_parm_bool(ntvfs_ctx->lp_ctx, NULL, "opendb", "oplocks", false);
 101
 102         return odb;
 103 }
 104
 105 /*
 106   get a lock on a entry in the odb. This call returns a lock handle,
 107   which the caller should unlock using talloc_free().
 108 */
 109 static struct odb_lock *odb_ctdb_lock(TALLOC_CTX *mem_ctx,
 110                                      struct odb_context *odb, DATA_BLOB *file_key)
 111 {
 112         struct odb_lock *lck;
 113
 114         lck = talloc(mem_ctx, struct odb_lock);
 115         if (lck == NULL) {
 116                 return NULL;
 117         }
 118
 119         lck->odb = talloc_reference(lck, odb);
 120         lck->key.dptr = talloc_memdup(lck, file_key->data, file_key->length);
 121         lck->key.dsize = file_key->length;
 122         if (lck->key.dptr == NULL) {
 123                 talloc_free(lck);
 124                 return NULL;
 125         }
 126
 127         lck->rec = ctdb_fetch_lock(odb->ctdb_db, (TALLOC_CTX *)lck, lck->key, &lck->data);
 128         if (!lck->rec) {
 129                 talloc_free(lck);
 130                 return NULL;
 131         }
 132
 133         return lck;
 134 }
 135
 136 static DATA_BLOB odb_ctdb_get_key(TALLOC_CTX *mem_ctx, struct odb_lock *lck)
 137 {
 138         /*
 139          * as this file will went away and isn't used yet,
 140          * copy the implementation from the tdb backend
 141          * --metze
 142          */
 143         return data_blob_const(NULL, 0);
 144 }
 145
 146 /*
 147   determine if two odb_entry structures conflict
 148
 149   return NT_STATUS_OK on no conflict
 150 */
 151 static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
 152 {
 153         /* if either open involves no read.write or delete access then
 154            it can't conflict */
 155         if (!(e1->access_mask & (SEC_FILE_WRITE_DATA |
 156                                  SEC_FILE_APPEND_DATA |
 157                                  SEC_FILE_READ_DATA |
 158                                  SEC_FILE_EXECUTE |
 159                                  SEC_STD_DELETE))) {
 160                 return NT_STATUS_OK;
 161         }
 162         if (!(e2->access_mask & (SEC_FILE_WRITE_DATA |
 163                                  SEC_FILE_APPEND_DATA |
 164                                  SEC_FILE_READ_DATA |
 165                                  SEC_FILE_EXECUTE |
 166                                  SEC_STD_DELETE))) {
 167                 return NT_STATUS_OK;
 168         }
 169
 170         /* data IO access masks. This is skipped if the two open handles
 171            are on different streams (as in that case the masks don't
 172            interact) */
 173         if (e1->stream_id != e2->stream_id) {
 174                 return NT_STATUS_OK;
 175         }
 176
 177 #define CHECK_MASK(am, right, sa, share) \
 178         if (((am) & (right)) && !((sa) & (share))) return NT_STATUS_SHARING_VIOLATION
 179
 180         CHECK_MASK(e1->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
 181                    e2->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
 182         CHECK_MASK(e2->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
 183                    e1->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
 184
 185         CHECK_MASK(e1->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
 186                    e2->share_access, NTCREATEX_SHARE_ACCESS_READ);
 187         CHECK_MASK(e2->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
 188                    e1->share_access, NTCREATEX_SHARE_ACCESS_READ);
 189
 190         CHECK_MASK(e1->access_mask, SEC_STD_DELETE,
 191                    e2->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
 192         CHECK_MASK(e2->access_mask, SEC_STD_DELETE,
 193                    e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
 194
 195         return NT_STATUS_OK;
 196 }
 197
 198 /*
 199   pull a record, translating from the db format to the opendb_file structure defined
 200   in opendb.idl
 201 */
 202 static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 203 {
 204         TDB_DATA dbuf;
 205         DATA_BLOB blob;
 206         enum ndr_err_code ndr_err;
 207
 208         dbuf = lck->data;
 209
 210         if (dbuf.dsize == 0) {
 211                 /* empty record in ctdb means the record isn't there */
 212                 return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 213         }
 214
 215         blob.data = dbuf.dptr;
 216         blob.length = dbuf.dsize;
 217
 218         ndr_err = ndr_pull_struct_blob(&blob, lck, lp_iconv_convenience(lck->odb->ntvfs_ctx->lp_ctx), file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
 219         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 220                 return ndr_map_error2ntstatus(ndr_err);
 221         }
 222
 223         return NT_STATUS_OK;
 224 }
 225
 226 /*
 227   push a record, translating from the opendb_file structure defined in opendb.idl
 228 */
 229 static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 230 {
 231         TDB_DATA dbuf;
 232         DATA_BLOB blob;
 233         enum ndr_err_code ndr_err;
 234         int ret;
 235
 236         if (!file->num_entries) {
 237                 dbuf.dptr  = NULL;
 238                 dbuf.dsize = 0;
 239                 ctdb_record_store(lck->rec, dbuf);
 240                 talloc_free(lck->rec);
 241                 return NT_STATUS_OK;
 242         }
 243
 244         ndr_err = ndr_push_struct_blob(&blob, lck,
 245                                        lp_iconv_convenience(lck->odb->ntvfs_ctx->lp_ctx),
 246                                        file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
 247         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 248                 return ndr_map_error2ntstatus(ndr_err);
 249         }
 250
 251         dbuf.dptr = blob.data;
 252         dbuf.dsize = blob.length;
 253
 254         ret = ctdb_record_store(lck->rec, dbuf);
 255         talloc_free(lck->rec);
 256         data_blob_free(&blob);
 257         if (ret != 0) {
 258                 return NT_STATUS_INTERNAL_DB_CORRUPTION;
 259         }
 260
 261         return NT_STATUS_OK;
 262 }
 263
 264 /*
 265   send an oplock break to a client
 266 */
 267 static NTSTATUS odb_oplock_break_send(struct odb_context *odb, struct opendb_entry *e)
 268 {
 269         /* tell the server handling this open file about the need to send the client
 270            a break */
 271         return messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, e->server,
 272                                   MSG_NTVFS_OPLOCK_BREAK, e->file_handle);
 273 }
 274
 275 /*
 276   register an open file in the open files database. This implements the share_access
 277   rules
 278
 279   Note that the path is only used by the delete on close logic, not
 280   for comparing with other filenames
 281 */
 282 static NTSTATUS odb_ctdb_open_file(struct odb_lock *lck,
 283                                    void *file_handle, const char *path,
 284                                    uint32_t stream_id, uint32_t share_access,
 285                                    uint32_t access_mask, bool delete_on_close,
 286                                    uint32_t open_disposition, bool break_to_none,
 287                                    bool allow_level_II_oplock,
 288                                    uint32_t oplock_level, uint32_t *oplock_granted)
 289
 290 {
 291         struct odb_context *odb = lck->odb;
 292         struct opendb_entry e;
 293         int i;
 294         struct opendb_file file;
 295         NTSTATUS status;
 296
 297         if (odb->oplocks == false) {
 298                 oplock_level = OPLOCK_NONE;
 299         }
 300
 301         status = odb_pull_record(lck, &file);
 302         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
 303                 /* initialise a blank structure */
 304                 ZERO_STRUCT(file);
 305                 file.path = path;
 306         } else {
 307                 NT_STATUS_NOT_OK_RETURN(status);
 308         }
 309
 310         /* see if it conflicts */
 311         e.server          = odb->ntvfs_ctx->server_id;
 312         e.file_handle     = file_handle;
 313         e.stream_id       = stream_id;
 314         e.share_access    = share_access;
 315         e.access_mask     = access_mask;
 316         e.delete_on_close = delete_on_close;
 317         e.oplock_level    = OPLOCK_NONE;
 318
 319         /* see if anyone has an oplock, which we need to break */
 320         for (i=0;i<file.num_entries;i++) {
 321                 if (file.entries[i].oplock_level == OPLOCK_BATCH) {
 322                         /* a batch oplock caches close calls, which
 323                            means the client application might have
 324                            already closed the file. We have to allow
 325                            this close to propogate by sending a oplock
 326                            break request and suspending this call
 327                            until the break is acknowledged or the file
 328                            is closed */
 329                         odb_oplock_break_send(odb, &file.entries[i]);
 330                         return NT_STATUS_OPLOCK_NOT_GRANTED;
 331                 }
 332         }
 333
 334         if (file.delete_on_close ||
 335             (file.num_entries != 0 && delete_on_close)) {
 336                 /* while delete on close is set, no new opens are allowed */
 337                 return NT_STATUS_DELETE_PENDING;
 338         }
 339
 340         /* check for sharing violations */
 341         for (i=0;i<file.num_entries;i++) {
 342                 status = share_conflict(&file.entries[i], &e);
 343                 NT_STATUS_NOT_OK_RETURN(status);
 344         }
 345
 346         /* we now know the open could succeed, but we need to check
 347            for any exclusive oplocks. We can't grant a second open
 348            till these are broken. Note that we check for batch oplocks
 349            before checking for sharing violations, and check for
 350            exclusive oplocks afterwards. */
 351         for (i=0;i<file.num_entries;i++) {
 352                 if (file.entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
 353                         odb_oplock_break_send(odb, &file.entries[i]);
 354                         return NT_STATUS_OPLOCK_NOT_GRANTED;
 355                 }
 356         }
 357
 358         /*
 359           possibly grant an exclusive or batch oplock if this is the only client
 360           with the file open. We don't yet grant levelII oplocks.
 361         */
 362         if (oplock_granted != NULL) {
 363                 if ((oplock_level == OPLOCK_BATCH ||
 364                      oplock_level == OPLOCK_EXCLUSIVE) &&
 365                     file.num_entries == 0) {
 366                         (*oplock_granted) = oplock_level;
 367                 } else {
 368                         (*oplock_granted) = OPLOCK_NONE;
 369                 }
 370                 e.oplock_level = (*oplock_granted);
 371         }
 372
 373         /* it doesn't conflict, so add it to the end */
 374         file.entries = talloc_realloc(lck, file.entries, struct opendb_entry,
 375                                       file.num_entries+1);
 376         NT_STATUS_HAVE_NO_MEMORY(file.entries);
 377
 378         file.entries[file.num_entries] = e;
 379         file.num_entries++;
 380
 381         return odb_push_record(lck, &file);
 382 }
 383
 384
 385 /*
 386   register a pending open file in the open files database
 387 */
 388 static NTSTATUS odb_ctdb_open_file_pending(struct odb_lock *lck, void *private)
 389 {
 390         struct odb_context *odb = lck->odb;
 391         struct opendb_file file;
 392         NTSTATUS status;
 393
 394         status = odb_pull_record(lck, &file);
 395         NT_STATUS_NOT_OK_RETURN(status);
 396
 397         file.pending = talloc_realloc(lck, file.pending, struct opendb_pending,
 398                                       file.num_pending+1);
 399         NT_STATUS_HAVE_NO_MEMORY(file.pending);
 400
 401         file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
 402         file.pending[file.num_pending].notify_ptr = private;
 403
 404         file.num_pending++;
 405
 406         return odb_push_record(lck, &file);
 407 }
 408
 409
 410 /*
 411   remove a opendb entry
 412 */
 413 static NTSTATUS odb_ctdb_close_file(struct odb_lock *lck, void *file_handle,
 414                                     const char **_delete_path)
 415 {
 416         struct odb_context *odb = lck->odb;
 417         struct opendb_file file;
 418         const char *delete_path = NULL;
 419         int i;
 420         NTSTATUS status;
 421
 422         status = odb_pull_record(lck, &file);
 423         NT_STATUS_NOT_OK_RETURN(status);
 424
 425         /* find the entry, and delete it */
 426         for (i=0;i<file.num_entries;i++) {
 427                 if (file_handle == file.entries[i].file_handle &&
 428                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
 429                         if (file.entries[i].delete_on_close) {
 430                                 file.delete_on_close = true;
 431                         }
 432                         if (i < file.num_entries-1) {
 433                                 memmove(file.entries+i, file.entries+i+1,
 434                                         (file.num_entries - (i+1)) *
 435                                         sizeof(struct opendb_entry));
 436                         }
 437                         break;
 438                 }
 439         }
 440
 441         if (i == file.num_entries) {
 442                 return NT_STATUS_UNSUCCESSFUL;
 443         }
 444
 445         /* send any pending notifications, removing them once sent */
 446         for (i=0;i<file.num_pending;i++) {
 447                 messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server,
 448                                    MSG_PVFS_RETRY_OPEN,
 449                                    file.pending[i].notify_ptr);
 450         }
 451         file.num_pending = 0;
 452
 453         file.num_entries--;
 454
 455         if (file.num_entries == 0 && file.delete_on_close) {
 456                 delete_path = talloc_strdup(lck, file.path);
 457                 NT_STATUS_HAVE_NO_MEMORY(delete_path);
 458         }
 459
 460         if (_delete_path) {
 461                 *_delete_path = delete_path;
 462         }
 463
 464         return odb_push_record(lck, &file);
 465 }
 466
 467 /*
 468   update the oplock level of the client
 469 */
 470 static NTSTATUS odb_ctdb_update_oplock(struct odb_lock *lck, void *file_handle,
 471                                        uint32_t oplock_level)
 472 {
 473         /*
 474          * as this file will went away and isn't used yet,
 475          * copy the implementation from the tdb backend
 476          * --metze
 477          */
 478         return NT_STATUS_FOOBAR;
 479 }
 480
 481 static NTSTATUS odb_ctdb_break_oplocks(struct odb_lock *lck)
 482 {
 483         /*
 484          * as this file will went away and isn't used yet,
 485          * copy the implementation from the tdb backend
 486          * --metze
 487          */
 488         return NT_STATUS_FOOBAR;
 489 }
 490
 491 /*
 492   remove a pending opendb entry
 493 */
 494 static NTSTATUS odb_ctdb_remove_pending(struct odb_lock *lck, void *private)
 495 {
 496         struct odb_context *odb = lck->odb;
 497         int i;
 498         NTSTATUS status;
 499         struct opendb_file file;
 500
 501         status = odb_pull_record(lck, &file);
 502         NT_STATUS_NOT_OK_RETURN(status);
 503
 504         /* find the entry, and delete it */
 505         for (i=0;i<file.num_pending;i++) {
 506                 if (private == file.pending[i].notify_ptr &&
 507                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
 508                         if (i < file.num_pending-1) {
 509                                 memmove(file.pending+i, file.pending+i+1,
 510                                         (file.num_pending - (i+1)) *
 511                                         sizeof(struct opendb_pending));
 512                         }
 513                         break;
 514                 }
 515         }
 516
 517         if (i == file.num_pending) {
 518                 return NT_STATUS_UNSUCCESSFUL;
 519         }
 520
 521         file.num_pending--;
 522
 523         return odb_push_record(lck, &file);
 524 }
 525
 526
 527 /*
 528   rename the path in a open file
 529 */
 530 static NTSTATUS odb_ctdb_rename(struct odb_lock *lck, const char *path)
 531 {
 532         struct opendb_file file;
 533         NTSTATUS status;
 534
 535         status = odb_pull_record(lck, &file);
 536         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
 537                 /* not having the record at all is OK */
 538                 return NT_STATUS_OK;
 539         }
 540         NT_STATUS_NOT_OK_RETURN(status);
 541
 542         file.path = path;
 543         return odb_push_record(lck, &file);
 544 }
 545
 546 /*
 547   get the path of an open file
 548 */
 549 static NTSTATUS odb_ctdb_get_path(struct odb_lock *lck, const char **path)
 550 {
 551         struct opendb_file file;
 552         NTSTATUS status;
 553
 554         *path = NULL;
 555
 556         status = odb_pull_record(lck, &file);
 557         /* we don't ignore NT_STATUS_OBJECT_NAME_NOT_FOUND here */
 558         NT_STATUS_NOT_OK_RETURN(status);
 559
 560         *path = file.path;
 561
 562         return NT_STATUS_OK;
 563 }
 564
 565 /*
 566   update delete on close flag on an open file
 567 */
 568 static NTSTATUS odb_ctdb_set_delete_on_close(struct odb_lock *lck, bool del_on_close)
 569 {
 570         NTSTATUS status;
 571         struct opendb_file file;
 572
 573         status = odb_pull_record(lck, &file);
 574         NT_STATUS_NOT_OK_RETURN(status);
 575
 576         file.delete_on_close = del_on_close;
 577
 578         return odb_push_record(lck, &file);
 579 }
 580
 581 /*
 582   return the current value of the delete_on_close bit, and how many
 583   people still have the file open
 584 */
 585 static NTSTATUS odb_ctdb_get_delete_on_close(struct odb_context *odb,
 586                                             DATA_BLOB *key, bool *del_on_close)
 587 {
 588         NTSTATUS status;
 589         struct opendb_file file;
 590         struct odb_lock *lck;
 591
 592         (*del_on_close) = false;
 593
 594         lck = odb_lock(odb, odb, key);
 595         NT_STATUS_HAVE_NO_MEMORY(lck);
 596
 597         status = odb_pull_record(lck, &file);
 598         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
 599                 talloc_free(lck);
 600                 return NT_STATUS_OK;
 601         }
 602         if (!NT_STATUS_IS_OK(status)) {
 603                 talloc_free(lck);
 604                 return status;
 605         }
 606
 607         (*del_on_close) = file.delete_on_close;
 608
 609         talloc_free(lck);
 610
 611         return NT_STATUS_OK;
 612 }
 613
 614
 615 /*
 616   determine if a file can be opened with the given share_access,
 617   create_options and access_mask
 618 */
 619 static NTSTATUS odb_ctdb_can_open(struct odb_lock *lck,
 620                                   uint32_t stream_id, uint32_t share_access,
 621                                   uint32_t access_mask, bool delete_on_close,
 622                                   uint32_t open_disposition, bool break_to_none)
 623 {
 624         struct odb_context *odb = lck->odb;
 625         NTSTATUS status;
 626         struct opendb_file file;
 627         struct opendb_entry e;
 628         int i;
 629
 630         status = odb_pull_record(lck, &file);
 631         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
 632                 return NT_STATUS_OK;
 633         }
 634         NT_STATUS_NOT_OK_RETURN(status);
 635
 636         if (delete_on_close &&
 637             file.num_entries != 0) {
 638                 return NT_STATUS_SHARING_VIOLATION;
 639         }
 640
 641         if (file.delete_on_close) {
 642                 return NT_STATUS_DELETE_PENDING;
 643         }
 644
 645         e.server       = odb->ntvfs_ctx->server_id;
 646         e.file_handle  = NULL;
 647         e.stream_id    = 0;
 648         e.share_access = share_access;
 649         e.access_mask  = access_mask;
 650
 651         for (i=0;i<file.num_entries;i++) {
 652                 status = share_conflict(&file.entries[i], &e);
 653                 if (!NT_STATUS_IS_OK(status)) {
 654                         /* note that we discard the error code
 655                            here. We do this as unless we are actually
 656                            doing an open (which comes via a different
 657                            function), we need to return a sharing
 658                            violation */
 659                         return NT_STATUS_SHARING_VIOLATION;
 660                 }
 661         }
 662
 663         return NT_STATUS_OK;
 664 }
 665
 666
 667 static const struct opendb_ops opendb_ctdb_ops = {
 668         .odb_init                = odb_ctdb_init,
 669         .odb_lock                = odb_ctdb_lock,
 670         .odb_get_key             = odb_ctdb_get_key,
 671         .odb_open_file           = odb_ctdb_open_file,
 672         .odb_open_file_pending   = odb_ctdb_open_file_pending,
 673         .odb_close_file          = odb_ctdb_close_file,
 674         .odb_remove_pending      = odb_ctdb_remove_pending,
 675         .odb_rename              = odb_ctdb_rename,
 676         .odb_get_path            = odb_ctdb_get_path,
 677         .odb_set_delete_on_close = odb_ctdb_set_delete_on_close,
 678         .odb_get_delete_on_close = odb_ctdb_get_delete_on_close,
 679         .odb_can_open            = odb_ctdb_can_open,
 680         .odb_update_oplock       = odb_ctdb_update_oplock,
 681         .odb_break_oplocks       = odb_ctdb_break_oplocks
 682 };
 683
 684
 685 void odb_ctdb_init_ops(void)
 686 {
 687         odb_set_ops(&opendb_ctdb_ops);
 688 }