source4/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 "db_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(global_loadparm, 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 /*
 137   determine if two odb_entry structures conflict
 138
 139   return NT_STATUS_OK on no conflict
 140 */
 141 static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
 142 {
 143         /* if either open involves no read.write or delete access then
 144            it can't conflict */
 145         if (!(e1->access_mask & (SEC_FILE_WRITE_DATA |
 146                                  SEC_FILE_APPEND_DATA |
 147                                  SEC_FILE_READ_DATA |
 148                                  SEC_FILE_EXECUTE |
 149                                  SEC_STD_DELETE))) {
 150                 return NT_STATUS_OK;
 151         }
 152         if (!(e2->access_mask & (SEC_FILE_WRITE_DATA |
 153                                  SEC_FILE_APPEND_DATA |
 154                                  SEC_FILE_READ_DATA |
 155                                  SEC_FILE_EXECUTE |
 156                                  SEC_STD_DELETE))) {
 157                 return NT_STATUS_OK;
 158         }
 159
 160         /* data IO access masks. This is skipped if the two open handles
 161            are on different streams (as in that case the masks don't
 162            interact) */
 163         if (e1->stream_id != e2->stream_id) {
 164                 return NT_STATUS_OK;
 165         }
 166
 167 #define CHECK_MASK(am, right, sa, share) \
 168         if (((am) & (right)) && !((sa) & (share))) return NT_STATUS_SHARING_VIOLATION
 169
 170         CHECK_MASK(e1->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
 171                    e2->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
 172         CHECK_MASK(e2->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
 173                    e1->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
 174
 175         CHECK_MASK(e1->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
 176                    e2->share_access, NTCREATEX_SHARE_ACCESS_READ);
 177         CHECK_MASK(e2->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
 178                    e1->share_access, NTCREATEX_SHARE_ACCESS_READ);
 179
 180         CHECK_MASK(e1->access_mask, SEC_STD_DELETE,
 181                    e2->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
 182         CHECK_MASK(e2->access_mask, SEC_STD_DELETE,
 183                    e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
 184
 185         return NT_STATUS_OK;
 186 }
 187
 188 /*
 189   pull a record, translating from the db format to the opendb_file structure defined
 190   in opendb.idl
 191 */
 192 static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 193 {
 194         TDB_DATA dbuf;
 195         DATA_BLOB blob;
 196         NTSTATUS status;
 197
 198         dbuf = lck->data;
 199
 200         if (dbuf.dsize == 0) {
 201                 /* empty record in ctdb means the record isn't there */
 202                 return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 203         }
 204
 205         blob.data = dbuf.dptr;
 206         blob.length = dbuf.dsize;
 207
 208         status = ndr_pull_struct_blob(&blob, lck, file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
 209
 210         return status;
 211 }
 212
 213 /*
 214   push a record, translating from the opendb_file structure defined in opendb.idl
 215 */
 216 static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 217 {
 218         TDB_DATA dbuf;
 219         DATA_BLOB blob;
 220         NTSTATUS status;
 221         int ret;
 222
 223         if (!file->num_entries) {
 224                 dbuf.dptr  = NULL;
 225                 dbuf.dsize = 0;
 226                 ctdb_record_store(lck->rec, dbuf);
 227                 talloc_free(lck->rec);
 228                 return NT_STATUS_OK;
 229         }
 230
 231         status = ndr_push_struct_blob(&blob, lck, file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
 232         NT_STATUS_NOT_OK_RETURN(status);
 233
 234         dbuf.dptr = blob.data;
 235         dbuf.dsize = blob.length;
 236
 237         ret = ctdb_record_store(lck->rec, dbuf);
 238         talloc_free(lck->rec);
 239         data_blob_free(&blob);
 240         if (ret != 0) {
 241                 return NT_STATUS_INTERNAL_DB_CORRUPTION;
 242         }
 243
 244         return NT_STATUS_OK;
 245 }
 246
 247 /*
 248   send an oplock break to a client
 249 */
 250 static NTSTATUS odb_oplock_break_send(struct odb_context *odb, struct opendb_entry *e)
 251 {
 252         /* tell the server handling this open file about the need to send the client
 253            a break */
 254         return messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, e->server,
 255                                   MSG_NTVFS_OPLOCK_BREAK, e->file_handle);
 256 }
 257
 258 /*
 259   register an open file in the open files database. This implements the share_access
 260   rules
 261
 262   Note that the path is only used by the delete on close logic, not
 263   for comparing with other filenames
 264 */
 265 static NTSTATUS odb_ctdb_open_file(struct odb_lock *lck, void *file_handle,
 266                                   uint32_t stream_id, uint32_t share_access,
 267                                   uint32_t access_mask, BOOL delete_on_close,
 268                                   const char *path,
 269                                   uint32_t oplock_level, uint32_t *oplock_granted)
 270 {
 271         struct odb_context *odb = lck->odb;
 272         struct opendb_entry e;
 273         int i;
 274         struct opendb_file file;
 275         NTSTATUS status;
 276
 277         if (odb->oplocks == False) {
 278                 oplock_level = OPLOCK_NONE;
 279         }
 280
 281         status = odb_pull_record(lck, &file);
 282         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
 283                 /* initialise a blank structure */
 284                 ZERO_STRUCT(file);
 285                 file.path = path;
 286         } else {
 287                 NT_STATUS_NOT_OK_RETURN(status);
 288         }
 289
 290         /* see if it conflicts */
 291         e.server          = odb->ntvfs_ctx->server_id;
 292         e.file_handle     = file_handle;
 293         e.stream_id       = stream_id;
 294         e.share_access    = share_access;
 295         e.access_mask     = access_mask;
 296         e.delete_on_close = delete_on_close;
 297         e.oplock_level    = OPLOCK_NONE;
 298
 299         /* see if anyone has an oplock, which we need to break */
 300         for (i=0;i<file.num_entries;i++) {
 301                 if (file.entries[i].oplock_level == OPLOCK_BATCH) {
 302                         /* a batch oplock caches close calls, which
 303                            means the client application might have
 304                            already closed the file. We have to allow
 305                            this close to propogate by sending a oplock
 306                            break request and suspending this call
 307                            until the break is acknowledged or the file
 308                            is closed */
 309                         odb_oplock_break_send(odb, &file.entries[i]);
 310                         return NT_STATUS_OPLOCK_NOT_GRANTED;
 311                 }
 312         }
 313
 314         if (file.delete_on_close ||
 315             (file.num_entries != 0 && delete_on_close)) {
 316                 /* while delete on close is set, no new opens are allowed */
 317                 return NT_STATUS_DELETE_PENDING;
 318         }
 319
 320         /* check for sharing violations */
 321         for (i=0;i<file.num_entries;i++) {
 322                 status = share_conflict(&file.entries[i], &e);
 323                 NT_STATUS_NOT_OK_RETURN(status);
 324         }
 325
 326         /* we now know the open could succeed, but we need to check
 327            for any exclusive oplocks. We can't grant a second open
 328            till these are broken. Note that we check for batch oplocks
 329            before checking for sharing violations, and check for
 330            exclusive oplocks afterwards. */
 331         for (i=0;i<file.num_entries;i++) {
 332                 if (file.entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
 333                         odb_oplock_break_send(odb, &file.entries[i]);
 334                         return NT_STATUS_OPLOCK_NOT_GRANTED;
 335                 }
 336         }
 337
 338         /*
 339           possibly grant an exclusive or batch oplock if this is the only client
 340           with the file open. We don't yet grant levelII oplocks.
 341         */
 342         if (oplock_granted != NULL) {
 343                 if ((oplock_level == OPLOCK_BATCH ||
 344                      oplock_level == OPLOCK_EXCLUSIVE) &&
 345                     file.num_entries == 0) {
 346                         (*oplock_granted) = oplock_level;
 347                 } else {
 348                         (*oplock_granted) = OPLOCK_NONE;
 349                 }
 350                 e.oplock_level = (*oplock_granted);
 351         }
 352
 353         /* it doesn't conflict, so add it to the end */
 354         file.entries = talloc_realloc(lck, file.entries, struct opendb_entry,
 355                                       file.num_entries+1);
 356         NT_STATUS_HAVE_NO_MEMORY(file.entries);
 357
 358         file.entries[file.num_entries] = e;
 359         file.num_entries++;
 360
 361         return odb_push_record(lck, &file);
 362 }
 363
 364
 365 /*
 366   register a pending open file in the open files database
 367 */
 368 static NTSTATUS odb_ctdb_open_file_pending(struct odb_lock *lck, void *private)
 369 {
 370         struct odb_context *odb = lck->odb;
 371         struct opendb_file file;
 372         NTSTATUS status;
 373
 374         status = odb_pull_record(lck, &file);
 375         NT_STATUS_NOT_OK_RETURN(status);
 376
 377         file.pending = talloc_realloc(lck, file.pending, struct opendb_pending,
 378                                       file.num_pending+1);
 379         NT_STATUS_HAVE_NO_MEMORY(file.pending);
 380
 381         file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
 382         file.pending[file.num_pending].notify_ptr = private;
 383
 384         file.num_pending++;
 385
 386         return odb_push_record(lck, &file);
 387 }
 388
 389
 390 /*
 391   remove a opendb entry
 392 */
 393 static NTSTATUS odb_ctdb_close_file(struct odb_lock *lck, void *file_handle)
 394 {
 395         struct odb_context *odb = lck->odb;
 396         struct opendb_file file;
 397         int i;
 398         NTSTATUS status;
 399
 400         status = odb_pull_record(lck, &file);
 401         NT_STATUS_NOT_OK_RETURN(status);
 402
 403         /* find the entry, and delete it */
 404         for (i=0;i<file.num_entries;i++) {
 405                 if (file_handle == file.entries[i].file_handle &&
 406                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
 407                         if (file.entries[i].delete_on_close) {
 408                                 file.delete_on_close = True;
 409                         }
 410                         if (i < file.num_entries-1) {
 411                                 memmove(file.entries+i, file.entries+i+1,
 412                                         (file.num_entries - (i+1)) *
 413                                         sizeof(struct opendb_entry));
 414                         }
 415                         break;
 416                 }
 417         }
 418
 419         if (i == file.num_entries) {
 420                 return NT_STATUS_UNSUCCESSFUL;
 421         }
 422
 423         /* send any pending notifications, removing them once sent */
 424         for (i=0;i<file.num_pending;i++) {
 425                 messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server,
 426                                    MSG_PVFS_RETRY_OPEN,
 427                                    file.pending[i].notify_ptr);
 428         }
 429         file.num_pending = 0;
 430
 431         file.num_entries--;
 432
 433         return odb_push_record(lck, &file);
 434 }
 435
 436
 437 /*
 438   remove a pending opendb entry
 439 */
 440 static NTSTATUS odb_ctdb_remove_pending(struct odb_lock *lck, void *private)
 441 {
 442         struct odb_context *odb = lck->odb;
 443         int i;
 444         NTSTATUS status;
 445         struct opendb_file file;
 446
 447         status = odb_pull_record(lck, &file);
 448         NT_STATUS_NOT_OK_RETURN(status);
 449
 450         /* find the entry, and delete it */
 451         for (i=0;i<file.num_pending;i++) {
 452                 if (private == file.pending[i].notify_ptr &&
 453                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
 454                         if (i < file.num_pending-1) {
 455                                 memmove(file.pending+i, file.pending+i+1,
 456                                         (file.num_pending - (i+1)) *
 457                                         sizeof(struct opendb_pending));
 458                         }
 459                         break;
 460                 }
 461         }
 462
 463         if (i == file.num_pending) {
 464                 return NT_STATUS_UNSUCCESSFUL;
 465         }
 466
 467         file.num_pending--;
 468
 469         return odb_push_record(lck, &file);
 470 }
 471
 472
 473 /*
 474   rename the path in a open file
 475 */
 476 static NTSTATUS odb_ctdb_rename(struct odb_lock *lck, const char *path)
 477 {
 478         struct opendb_file file;
 479         NTSTATUS status;
 480
 481         status = odb_pull_record(lck, &file);
 482         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
 483                 /* not having the record at all is OK */
 484                 return NT_STATUS_OK;
 485         }
 486         NT_STATUS_NOT_OK_RETURN(status);
 487
 488         file.path = path;
 489         return odb_push_record(lck, &file);
 490 }
 491
 492 /*
 493   update delete on close flag on an open file
 494 */
 495 static NTSTATUS odb_ctdb_set_delete_on_close(struct odb_lock *lck, BOOL del_on_close)
 496 {
 497         NTSTATUS status;
 498         struct opendb_file file;
 499
 500         status = odb_pull_record(lck, &file);
 501         NT_STATUS_NOT_OK_RETURN(status);
 502
 503         file.delete_on_close = del_on_close;
 504
 505         return odb_push_record(lck, &file);
 506 }
 507
 508 /*
 509   return the current value of the delete_on_close bit, and how many
 510   people still have the file open
 511 */
 512 static NTSTATUS odb_ctdb_get_delete_on_close(struct odb_context *odb,
 513                                             DATA_BLOB *key, BOOL *del_on_close,
 514                                             int *open_count, char **path)
 515 {
 516         NTSTATUS status;
 517         struct opendb_file file;
 518         struct odb_lock *lck;
 519
 520         lck = odb_lock(odb, odb, key);
 521         NT_STATUS_HAVE_NO_MEMORY(lck);
 522
 523         status = odb_pull_record(lck, &file);
 524         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
 525                 talloc_free(lck);
 526                 (*del_on_close) = False;
 527                 return NT_STATUS_OK;
 528         }
 529         if (!NT_STATUS_IS_OK(status)) {
 530                 talloc_free(lck);
 531                 return status;
 532         }
 533
 534         (*del_on_close) = file.delete_on_close;
 535         if (open_count != NULL) {
 536                 (*open_count) = file.num_entries;
 537         }
 538         if (path != NULL) {
 539                 *path = talloc_strdup(odb, file.path);
 540                 NT_STATUS_HAVE_NO_MEMORY(*path);
 541                 if (file.num_entries == 1 && file.entries[0].delete_on_close) {
 542                         (*del_on_close) = True;
 543                 }
 544         }
 545
 546         talloc_free(lck);
 547
 548         return NT_STATUS_OK;
 549 }
 550
 551
 552 /*
 553   determine if a file can be opened with the given share_access,
 554   create_options and access_mask
 555 */
 556 static NTSTATUS odb_ctdb_can_open(struct odb_lock *lck,
 557                                  uint32_t share_access, uint32_t create_options,
 558                                  uint32_t access_mask)
 559 {
 560         struct odb_context *odb = lck->odb;
 561         NTSTATUS status;
 562         struct opendb_file file;
 563         struct opendb_entry e;
 564         int i;
 565
 566         status = odb_pull_record(lck, &file);
 567         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
 568                 return NT_STATUS_OK;
 569         }
 570         NT_STATUS_NOT_OK_RETURN(status);
 571
 572         if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) &&
 573             file.num_entries != 0) {
 574                 return NT_STATUS_SHARING_VIOLATION;
 575         }
 576
 577         if (file.delete_on_close) {
 578                 return NT_STATUS_DELETE_PENDING;
 579         }
 580
 581         e.server       = odb->ntvfs_ctx->server_id;
 582         e.file_handle  = NULL;
 583         e.stream_id    = 0;
 584         e.share_access = share_access;
 585         e.access_mask  = access_mask;
 586
 587         for (i=0;i<file.num_entries;i++) {
 588                 status = share_conflict(&file.entries[i], &e);
 589                 if (!NT_STATUS_IS_OK(status)) {
 590                         /* note that we discard the error code
 591                            here. We do this as unless we are actually
 592                            doing an open (which comes via a different
 593                            function), we need to return a sharing
 594                            violation */
 595                         return NT_STATUS_SHARING_VIOLATION;
 596                 }
 597         }
 598
 599         return NT_STATUS_OK;
 600 }
 601
 602
 603 static const struct opendb_ops opendb_ctdb_ops = {
 604         .odb_init                = odb_ctdb_init,
 605         .odb_lock                = odb_ctdb_lock,
 606         .odb_open_file           = odb_ctdb_open_file,
 607         .odb_open_file_pending   = odb_ctdb_open_file_pending,
 608         .odb_close_file          = odb_ctdb_close_file,
 609         .odb_remove_pending      = odb_ctdb_remove_pending,
 610         .odb_rename              = odb_ctdb_rename,
 611         .odb_set_delete_on_close = odb_ctdb_set_delete_on_close,
 612         .odb_get_delete_on_close = odb_ctdb_get_delete_on_close,
 613         .odb_can_open            = odb_ctdb_can_open
 614 };
 615
 616
 617 void odb_ctdb_init_ops(void)
 618 {
 619         odb_set_ops(&opendb_ctdb_ops);
 620 }