Fix up the libnet tests.
[samba.git] / 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 "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, void *file_handle,
283                                   uint32_t stream_id, uint32_t share_access, 
284                                   uint32_t access_mask, bool delete_on_close,
285                                   const char *path, 
286                                   uint32_t oplock_level, uint32_t *oplock_granted)
287 {
288         struct odb_context *odb = lck->odb;
289         struct opendb_entry e;
290         int i;
291         struct opendb_file file;
292         NTSTATUS status;
293
294         if (odb->oplocks == false) {
295                 oplock_level = OPLOCK_NONE;
296         }
297
298         status = odb_pull_record(lck, &file);
299         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
300                 /* initialise a blank structure */
301                 ZERO_STRUCT(file);
302                 file.path = path;
303         } else {
304                 NT_STATUS_NOT_OK_RETURN(status);
305         }
306
307         /* see if it conflicts */
308         e.server          = odb->ntvfs_ctx->server_id;
309         e.file_handle     = file_handle;
310         e.stream_id       = stream_id;
311         e.share_access    = share_access;
312         e.access_mask     = access_mask;
313         e.delete_on_close = delete_on_close;
314         e.oplock_level    = OPLOCK_NONE;
315                 
316         /* see if anyone has an oplock, which we need to break */
317         for (i=0;i<file.num_entries;i++) {
318                 if (file.entries[i].oplock_level == OPLOCK_BATCH) {
319                         /* a batch oplock caches close calls, which
320                            means the client application might have
321                            already closed the file. We have to allow
322                            this close to propogate by sending a oplock
323                            break request and suspending this call
324                            until the break is acknowledged or the file
325                            is closed */
326                         odb_oplock_break_send(odb, &file.entries[i]);
327                         return NT_STATUS_OPLOCK_NOT_GRANTED;
328                 }
329         }
330
331         if (file.delete_on_close || 
332             (file.num_entries != 0 && delete_on_close)) {
333                 /* while delete on close is set, no new opens are allowed */
334                 return NT_STATUS_DELETE_PENDING;
335         }
336
337         /* check for sharing violations */
338         for (i=0;i<file.num_entries;i++) {
339                 status = share_conflict(&file.entries[i], &e);
340                 NT_STATUS_NOT_OK_RETURN(status);
341         }
342
343         /* we now know the open could succeed, but we need to check
344            for any exclusive oplocks. We can't grant a second open
345            till these are broken. Note that we check for batch oplocks
346            before checking for sharing violations, and check for
347            exclusive oplocks afterwards. */
348         for (i=0;i<file.num_entries;i++) {
349                 if (file.entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
350                         odb_oplock_break_send(odb, &file.entries[i]);
351                         return NT_STATUS_OPLOCK_NOT_GRANTED;
352                 }
353         }
354
355         /*
356           possibly grant an exclusive or batch oplock if this is the only client
357           with the file open. We don't yet grant levelII oplocks.
358         */
359         if (oplock_granted != NULL) {
360                 if ((oplock_level == OPLOCK_BATCH ||
361                      oplock_level == OPLOCK_EXCLUSIVE) &&
362                     file.num_entries == 0) {
363                         (*oplock_granted) = oplock_level;
364                 } else {
365                         (*oplock_granted) = OPLOCK_NONE;
366                 }
367                 e.oplock_level = (*oplock_granted);
368         }
369
370         /* it doesn't conflict, so add it to the end */
371         file.entries = talloc_realloc(lck, file.entries, struct opendb_entry, 
372                                       file.num_entries+1);
373         NT_STATUS_HAVE_NO_MEMORY(file.entries);
374
375         file.entries[file.num_entries] = e;
376         file.num_entries++;
377
378         return odb_push_record(lck, &file);
379 }
380
381
382 /*
383   register a pending open file in the open files database
384 */
385 static NTSTATUS odb_ctdb_open_file_pending(struct odb_lock *lck, void *private)
386 {
387         struct odb_context *odb = lck->odb;
388         struct opendb_file file;
389         NTSTATUS status;
390                 
391         status = odb_pull_record(lck, &file);
392         NT_STATUS_NOT_OK_RETURN(status);
393
394         file.pending = talloc_realloc(lck, file.pending, struct opendb_pending, 
395                                       file.num_pending+1);
396         NT_STATUS_HAVE_NO_MEMORY(file.pending);
397
398         file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
399         file.pending[file.num_pending].notify_ptr = private;
400
401         file.num_pending++;
402
403         return odb_push_record(lck, &file);
404 }
405
406
407 /*
408   remove a opendb entry
409 */
410 static NTSTATUS odb_ctdb_close_file(struct odb_lock *lck, void *file_handle)
411 {
412         struct odb_context *odb = lck->odb;
413         struct opendb_file file;
414         int i;
415         NTSTATUS status;
416
417         status = odb_pull_record(lck, &file);
418         NT_STATUS_NOT_OK_RETURN(status);
419
420         /* find the entry, and delete it */
421         for (i=0;i<file.num_entries;i++) {
422                 if (file_handle == file.entries[i].file_handle &&
423                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
424                         if (file.entries[i].delete_on_close) {
425                                 file.delete_on_close = true;
426                         }
427                         if (i < file.num_entries-1) {
428                                 memmove(file.entries+i, file.entries+i+1, 
429                                         (file.num_entries - (i+1)) * 
430                                         sizeof(struct opendb_entry));
431                         }
432                         break;
433                 }
434         }
435
436         if (i == file.num_entries) {
437                 return NT_STATUS_UNSUCCESSFUL;
438         }
439
440         /* send any pending notifications, removing them once sent */
441         for (i=0;i<file.num_pending;i++) {
442                 messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server, 
443                                    MSG_PVFS_RETRY_OPEN, 
444                                    file.pending[i].notify_ptr);
445         }
446         file.num_pending = 0;
447
448         file.num_entries--;
449         
450         return odb_push_record(lck, &file);
451 }
452
453 /*
454   update the oplock level of the client
455 */
456 static NTSTATUS odb_ctdb_update_oplock(struct odb_lock *lck, void *file_handle,
457                                        uint32_t oplock_level)
458 {
459         /*
460          * as this file will went away and isn't used yet,
461          * copy the implementation from the tdb backend
462          * --metze
463          */
464         return NT_STATUS_FOOBAR;
465 }
466
467 /*
468   remove a pending opendb entry
469 */
470 static NTSTATUS odb_ctdb_remove_pending(struct odb_lock *lck, void *private)
471 {
472         struct odb_context *odb = lck->odb;
473         int i;
474         NTSTATUS status;
475         struct opendb_file file;
476
477         status = odb_pull_record(lck, &file);
478         NT_STATUS_NOT_OK_RETURN(status);
479
480         /* find the entry, and delete it */
481         for (i=0;i<file.num_pending;i++) {
482                 if (private == file.pending[i].notify_ptr &&
483                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
484                         if (i < file.num_pending-1) {
485                                 memmove(file.pending+i, file.pending+i+1, 
486                                         (file.num_pending - (i+1)) * 
487                                         sizeof(struct opendb_pending));
488                         }
489                         break;
490                 }
491         }
492
493         if (i == file.num_pending) {
494                 return NT_STATUS_UNSUCCESSFUL;
495         }
496
497         file.num_pending--;
498         
499         return odb_push_record(lck, &file);
500 }
501
502
503 /*
504   rename the path in a open file
505 */
506 static NTSTATUS odb_ctdb_rename(struct odb_lock *lck, const char *path)
507 {
508         struct opendb_file file;
509         NTSTATUS status;
510
511         status = odb_pull_record(lck, &file);
512         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
513                 /* not having the record at all is OK */
514                 return NT_STATUS_OK;
515         }
516         NT_STATUS_NOT_OK_RETURN(status);
517
518         file.path = path;
519         return odb_push_record(lck, &file);
520 }
521
522 /*
523   update delete on close flag on an open file
524 */
525 static NTSTATUS odb_ctdb_set_delete_on_close(struct odb_lock *lck, bool del_on_close)
526 {
527         NTSTATUS status;
528         struct opendb_file file;
529
530         status = odb_pull_record(lck, &file);
531         NT_STATUS_NOT_OK_RETURN(status);
532
533         file.delete_on_close = del_on_close;
534
535         return odb_push_record(lck, &file);
536 }
537
538 /*
539   return the current value of the delete_on_close bit, and how many
540   people still have the file open
541 */
542 static NTSTATUS odb_ctdb_get_delete_on_close(struct odb_context *odb, 
543                                             DATA_BLOB *key, bool *del_on_close, 
544                                             int *open_count, char **path)
545 {
546         NTSTATUS status;
547         struct opendb_file file;
548         struct odb_lock *lck;
549
550         lck = odb_lock(odb, odb, key);
551         NT_STATUS_HAVE_NO_MEMORY(lck);
552
553         status = odb_pull_record(lck, &file);
554         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
555                 talloc_free(lck);
556                 (*del_on_close) = false;
557                 return NT_STATUS_OK;
558         }
559         if (!NT_STATUS_IS_OK(status)) {
560                 talloc_free(lck);
561                 return status;
562         }
563
564         (*del_on_close) = file.delete_on_close;
565         if (open_count != NULL) {
566                 (*open_count) = file.num_entries;
567         }
568         if (path != NULL) {
569                 *path = talloc_strdup(odb, file.path);
570                 NT_STATUS_HAVE_NO_MEMORY(*path);
571                 if (file.num_entries == 1 && file.entries[0].delete_on_close) {
572                         (*del_on_close) = true;
573                 }
574         }
575
576         talloc_free(lck);
577
578         return NT_STATUS_OK;
579 }
580
581
582 /*
583   determine if a file can be opened with the given share_access,
584   create_options and access_mask
585 */
586 static NTSTATUS odb_ctdb_can_open(struct odb_lock *lck,
587                                  uint32_t share_access, uint32_t create_options, 
588                                  uint32_t access_mask)
589 {
590         struct odb_context *odb = lck->odb;
591         NTSTATUS status;
592         struct opendb_file file;
593         struct opendb_entry e;
594         int i;
595
596         status = odb_pull_record(lck, &file);
597         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
598                 return NT_STATUS_OK;
599         }
600         NT_STATUS_NOT_OK_RETURN(status);
601
602         if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) && 
603             file.num_entries != 0) {
604                 return NT_STATUS_SHARING_VIOLATION;
605         }
606
607         if (file.delete_on_close) {
608                 return NT_STATUS_DELETE_PENDING;
609         }
610
611         e.server       = odb->ntvfs_ctx->server_id;
612         e.file_handle  = NULL;
613         e.stream_id    = 0;
614         e.share_access = share_access;
615         e.access_mask  = access_mask;
616                 
617         for (i=0;i<file.num_entries;i++) {
618                 status = share_conflict(&file.entries[i], &e);
619                 if (!NT_STATUS_IS_OK(status)) {
620                         /* note that we discard the error code
621                            here. We do this as unless we are actually
622                            doing an open (which comes via a different
623                            function), we need to return a sharing
624                            violation */
625                         return NT_STATUS_SHARING_VIOLATION;
626                 }
627         }
628
629         return NT_STATUS_OK;
630 }
631
632
633 static const struct opendb_ops opendb_ctdb_ops = {
634         .odb_init                = odb_ctdb_init,
635         .odb_lock                = odb_ctdb_lock,
636         .odb_get_key             = odb_ctdb_get_key,
637         .odb_open_file           = odb_ctdb_open_file,
638         .odb_open_file_pending   = odb_ctdb_open_file_pending,
639         .odb_close_file          = odb_ctdb_close_file,
640         .odb_remove_pending      = odb_ctdb_remove_pending,
641         .odb_rename              = odb_ctdb_rename,
642         .odb_set_delete_on_close = odb_ctdb_set_delete_on_close,
643         .odb_get_delete_on_close = odb_ctdb_get_delete_on_close,
644         .odb_can_open            = odb_ctdb_can_open,
645         .odb_update_oplock       = odb_ctdb_update_oplock
646 };
647
648
649 void odb_ctdb_init_ops(void)
650 {
651         odb_set_ops(&opendb_ctdb_ops);
652 }