X-Git-Url: http://git.samba.org/?p=samba.git;a=blobdiff_plain;f=source4%2Fntvfs%2Fcommon%2Fopendb_tdb.c;h=40a1d27f3016966692eaf1c7663e7328ac013b98;hp=41b5ad2ebe238e7c28dce6c0bf88ab287046337c;hb=4fc27c9969960dfef73ad80438d274e0f45e119e;hpb=61873ce94c172c801a4831de5550a8e0fe54c5f5

diff --git a/source4/ntvfs/common/opendb_tdb.c b/source4/ntvfs/common/opendb_tdb.c
index 41b5ad2ebe2..40a1d27f301 100644
--- a/source4/ntvfs/common/opendb_tdb.c
+++ b/source4/ntvfs/common/opendb_tdb.c
@@ -2,7 +2,8 @@
    Unix SMB/CIFS implementation.
 
    Copyright (C) Andrew Tridgell 2004
-   
+   Copyright (C) Stefan Metzmacher 2008
+
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
@@ -48,11 +49,13 @@
 #include "ntvfs/common/ntvfs_common.h"
 #include "cluster/cluster.h"
 #include "param/param.h"
+#include "ntvfs/sysdep/sys_lease.h"
 
 struct odb_context {
 	struct tdb_wrap *w;
 	struct ntvfs_context *ntvfs_ctx;
 	bool oplocks;
+	struct sys_lease_context *lease_ctx;
 };
 
 /*
@@ -62,8 +65,19 @@ struct odb_context {
 struct odb_lock {
 	struct odb_context *odb;
 	TDB_DATA key;
+
+	struct opendb_file file;
+
+	struct {
+		struct opendb_entry *e;
+		bool attrs_only;
+	} can_open;
 };
 
+static NTSTATUS odb_oplock_break_send(struct messaging_context *msg_ctx,
+				      struct opendb_entry *e,
+				      uint8_t level);
+
 /*
   Open up the openfiles.tdb database. Close it down using
   talloc_free(). We need the messaging_ctx to allow for pending open
@@ -87,8 +101,12 @@ static struct odb_context *odb_tdb_init(TALLOC_CTX *mem_ctx,
 
 	odb->ntvfs_ctx = ntvfs_ctx;
 
-	/* leave oplocks disabled by default until the code is working */
-	odb->oplocks = lp_parm_bool(ntvfs_ctx->lp_ctx, NULL, "opendb", "oplocks", false);
+	odb->oplocks = share_bool_option(ntvfs_ctx->ctx->config, SHARE_OPLOCKS, SHARE_OPLOCKS_DEFAULT);
+
+	odb->lease_ctx = sys_lease_context_create(ntvfs_ctx->config, odb,
+						  ntvfs_ctx->event_ctx,
+						  ntvfs_ctx->msg_ctx,
+						  odb_oplock_break_send);
 
 	return odb;
 }
@@ -102,6 +120,8 @@ static int odb_lock_destructor(struct odb_lock *lck)
 	return 0;
 }
 
+static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file);
+
 /*
   get a lock on a entry in the odb. This call returns a lock handle,
   which the caller should unlock using talloc_free().
@@ -110,6 +130,7 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
 				     struct odb_context *odb, DATA_BLOB *file_key)
 {
 	struct odb_lock *lck;
+	NTSTATUS status;
 
 	lck = talloc(mem_ctx, struct odb_lock);
 	if (lck == NULL) {
@@ -129,17 +150,37 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
 		return NULL;
 	}
 
+	ZERO_STRUCT(lck->can_open);
+
 	talloc_set_destructor(lck, odb_lock_destructor);
+
+	status = odb_pull_record(lck, &lck->file);
+	if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
+		/* initialise a blank structure */
+		ZERO_STRUCT(lck->file);
+	} else if (!NT_STATUS_IS_OK(status)) {
+		talloc_free(lck);
+		return NULL;
+	}
 	
 	return lck;
 }
 
+static DATA_BLOB odb_tdb_get_key(TALLOC_CTX *mem_ctx, struct odb_lock *lck)
+{
+	return data_blob_talloc(mem_ctx, lck->key.dptr, lck->key.dsize);
+}
+
+
 /*
   determine if two odb_entry structures conflict
 
   return NT_STATUS_OK on no conflict
 */
-static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
+static NTSTATUS share_conflict(struct opendb_entry *e1,
+			       uint32_t stream_id,
+			       uint32_t share_access,
+			       uint32_t access_mask)
 {
 	/* if either open involves no read.write or delete access then
 	   it can't conflict */
@@ -150,18 +191,18 @@ static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
 				 SEC_STD_DELETE))) {
 		return NT_STATUS_OK;
 	}
-	if (!(e2->access_mask & (SEC_FILE_WRITE_DATA |
-				 SEC_FILE_APPEND_DATA |
-				 SEC_FILE_READ_DATA |
-				 SEC_FILE_EXECUTE |
-				 SEC_STD_DELETE))) {
+	if (!(access_mask & (SEC_FILE_WRITE_DATA |
+			     SEC_FILE_APPEND_DATA |
+			     SEC_FILE_READ_DATA |
+			     SEC_FILE_EXECUTE |
+			     SEC_STD_DELETE))) {
 		return NT_STATUS_OK;
 	}
 
 	/* data IO access masks. This is skipped if the two open handles
 	   are on different streams (as in that case the masks don't
 	   interact) */
-	if (e1->stream_id != e2->stream_id) {
+	if (e1->stream_id != stream_id) {
 		return NT_STATUS_OK;
 	}
 
@@ -169,20 +210,20 @@ static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
 	if (((am) & (right)) && !((sa) & (share))) return NT_STATUS_SHARING_VIOLATION
 
 	CHECK_MASK(e1->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
-		   e2->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
-	CHECK_MASK(e2->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
+		   share_access, NTCREATEX_SHARE_ACCESS_WRITE);
+	CHECK_MASK(access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
 		   e1->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
 	
 	CHECK_MASK(e1->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
-		   e2->share_access, NTCREATEX_SHARE_ACCESS_READ);
-	CHECK_MASK(e2->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
+		   share_access, NTCREATEX_SHARE_ACCESS_READ);
+	CHECK_MASK(access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
 		   e1->share_access, NTCREATEX_SHARE_ACCESS_READ);
 
 	CHECK_MASK(e1->access_mask, SEC_STD_DELETE,
-		   e2->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
-	CHECK_MASK(e2->access_mask, SEC_STD_DELETE,
+		   share_access, NTCREATEX_SHARE_ACCESS_DELETE);
+	CHECK_MASK(access_mask, SEC_STD_DELETE,
 		   e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
-
+#undef CHECK_MASK
 	return NT_STATUS_OK;
 }
 
@@ -205,7 +246,7 @@ static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 	blob.data = dbuf.dptr;
 	blob.length = dbuf.dsize;
 
-	ndr_err = ndr_pull_struct_blob(&blob, lck, file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
+	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);
 	free(dbuf.dptr);
 	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 		return ndr_map_error2ntstatus(ndr_err);
@@ -233,7 +274,7 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 		return NT_STATUS_OK;
 	}
 
-	ndr_err = ndr_push_struct_blob(&blob, lck, file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
+	ndr_err = ndr_push_struct_blob(&blob, lck, lp_iconv_convenience(lck->odb->ntvfs_ctx->lp_ctx), file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
 	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 		return ndr_map_error2ntstatus(ndr_err);
 	}
@@ -253,58 +294,80 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 /*
   send an oplock break to a client
 */
-static NTSTATUS odb_oplock_break_send(struct odb_context *odb, struct opendb_entry *e)
+static NTSTATUS odb_oplock_break_send(struct messaging_context *msg_ctx,
+				      struct opendb_entry *e,
+				      uint8_t level)
 {
+	NTSTATUS status;
+	struct opendb_oplock_break op_break;
+	DATA_BLOB blob;
+
+	ZERO_STRUCT(op_break);
+
 	/* tell the server handling this open file about the need to send the client
 	   a break */
-	return messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, e->server, 
-				  MSG_NTVFS_OPLOCK_BREAK, e->file_handle);
-}
+	op_break.file_handle	= e->file_handle;
+	op_break.level		= level;
 
-/*
-  register an open file in the open files database. This implements the share_access
-  rules
+	blob = data_blob_const(&op_break, sizeof(op_break));
 
-  Note that the path is only used by the delete on close logic, not
-  for comparing with other filenames
-*/
-static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
-				  uint32_t stream_id, uint32_t share_access, 
-				  uint32_t access_mask, bool delete_on_close,
-				  const char *path, 
-				  uint32_t oplock_level, uint32_t *oplock_granted)
-{
-	struct odb_context *odb = lck->odb;
-	struct opendb_entry e;
-	int i;
-	struct opendb_file file;
-	NTSTATUS status;
+	status = messaging_send(msg_ctx, e->server,
+				MSG_NTVFS_OPLOCK_BREAK, &blob);
+	NT_STATUS_NOT_OK_RETURN(status);
 
-	if (odb->oplocks == false) {
-		oplock_level = OPLOCK_NONE;
+	return NT_STATUS_OK;
+}
+
+static bool access_attributes_only(uint32_t access_mask,
+				   uint32_t open_disposition,
+				   bool break_to_none)
+{
+	switch (open_disposition) {
+	case NTCREATEX_DISP_SUPERSEDE:
+	case NTCREATEX_DISP_OVERWRITE_IF:
+	case NTCREATEX_DISP_OVERWRITE:
+		return false;
+	default:
+		break;
 	}
 
-	status = odb_pull_record(lck, &file);
-	if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
-		/* initialise a blank structure */
-		ZERO_STRUCT(file);
-		file.path = path;
-	} else {
-		NT_STATUS_NOT_OK_RETURN(status);
+	if (break_to_none) {
+		return false;
 	}
 
-	/* see if it conflicts */
-	e.server          = odb->ntvfs_ctx->server_id;
-	e.file_handle     = file_handle;
-	e.stream_id       = stream_id;
-	e.share_access    = share_access;
-	e.access_mask     = access_mask;
-	e.delete_on_close = delete_on_close;
-	e.oplock_level    = OPLOCK_NONE;
-		
+#define CHECK_MASK(m,g) ((m) && (((m) & ~(g))==0) && (((m) & (g)) != 0))
+	return CHECK_MASK(access_mask,
+			  SEC_STD_SYNCHRONIZE |
+			  SEC_FILE_READ_ATTRIBUTE |
+			  SEC_FILE_WRITE_ATTRIBUTE);
+#undef CHECK_MASK
+}
+
+static NTSTATUS odb_tdb_open_can_internal(struct odb_context *odb,
+					  const struct opendb_file *file,
+					  uint32_t stream_id, uint32_t share_access,
+					  uint32_t access_mask, bool delete_on_close,
+					  uint32_t open_disposition, bool break_to_none,
+					  bool *_attrs_only)
+{
+	NTSTATUS status;
+	uint32_t i;
+	bool attrs_only = false;
+
 	/* see if anyone has an oplock, which we need to break */
-	for (i=0;i<file.num_entries;i++) {
-		if (file.entries[i].oplock_level == OPLOCK_BATCH) {
+	for (i=0;i<file->num_entries;i++) {
+		if (file->entries[i].oplock_level == OPLOCK_BATCH) {
+			bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
+			/* if this is an attribute only access
+			 * it doesn't conflict with a BACTCH oplock
+			 * but we'll not grant the oplock below
+			 */
+			attrs_only = access_attributes_only(access_mask,
+							    open_disposition,
+							    break_to_none);
+			if (attrs_only) {
+				break;
+			}
 			/* a batch oplock caches close calls, which
 			   means the client application might have
 			   already closed the file. We have to allow
@@ -312,20 +375,30 @@ static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
 			   break request and suspending this call
 			   until the break is acknowledged or the file
 			   is closed */
-			odb_oplock_break_send(odb, &file.entries[i]);
+			if (break_to_none ||
+			    !file->entries[i].allow_level_II_oplock) {
+				oplock_return = OPLOCK_BREAK_TO_NONE;
+			}
+			odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+					      &file->entries[i],
+					      oplock_return);
 			return NT_STATUS_OPLOCK_NOT_GRANTED;
 		}
 	}
 
-	if (file.delete_on_close || 
-	    (file.num_entries != 0 && delete_on_close)) {
+	if (file->delete_on_close) {
 		/* while delete on close is set, no new opens are allowed */
 		return NT_STATUS_DELETE_PENDING;
 	}
 
+	if (file->num_entries != 0 && delete_on_close) {
+		return NT_STATUS_SHARING_VIOLATION;
+	}
+
 	/* check for sharing violations */
-	for (i=0;i<file.num_entries;i++) {
-		status = share_conflict(&file.entries[i], &e);
+	for (i=0;i<file->num_entries;i++) {
+		status = share_conflict(&file->entries[i], stream_id,
+					share_access, access_mask);
 		NT_STATUS_NOT_OK_RETURN(status);
 	}
 
@@ -334,37 +407,136 @@ static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
 	   till these are broken. Note that we check for batch oplocks
 	   before checking for sharing violations, and check for
 	   exclusive oplocks afterwards. */
-	for (i=0;i<file.num_entries;i++) {
-		if (file.entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
-			odb_oplock_break_send(odb, &file.entries[i]);
+	for (i=0;i<file->num_entries;i++) {
+		if (file->entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
+			bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
+			/* if this is an attribute only access
+			 * it doesn't conflict with an EXCLUSIVE oplock
+			 * but we'll not grant the oplock below
+			 */
+			attrs_only = access_attributes_only(access_mask,
+							    open_disposition,
+							    break_to_none);
+			if (attrs_only) {
+				break;
+			}
+			/*
+			 * send an oplock break to the holder of the
+			 * oplock and tell caller to retry later
+			 */
+			if (break_to_none ||
+			    !file->entries[i].allow_level_II_oplock) {
+				oplock_return = OPLOCK_BREAK_TO_NONE;
+			}
+			odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+					      &file->entries[i],
+					      oplock_return);
 			return NT_STATUS_OPLOCK_NOT_GRANTED;
 		}
 	}
 
+	if (_attrs_only) {
+		*_attrs_only = attrs_only;
+	}
+	return NT_STATUS_OK;
+}
+
+/*
+  register an open file in the open files database.
+  The share_access rules are implemented by odb_can_open()
+  and it's needed to call odb_can_open() before
+  odb_open_file() otherwise NT_STATUS_INTERNAL_ERROR is returned
+
+  Note that the path is only used by the delete on close logic, not
+  for comparing with other filenames
+*/
+static NTSTATUS odb_tdb_open_file(struct odb_lock *lck,
+				  void *file_handle, const char *path,
+				  int *fd, bool allow_level_II_oplock,
+				  uint32_t oplock_level, uint32_t *oplock_granted)
+{
+	struct odb_context *odb = lck->odb;
+
+	if (!lck->can_open.e) {
+		return NT_STATUS_INTERNAL_ERROR;
+	}
+
+	if (odb->oplocks == false) {
+		oplock_level = OPLOCK_NONE;
+	}
+
+	if (!oplock_granted) {
+		oplock_level = OPLOCK_NONE;
+	}
+
+	if (lck->file.path == NULL) {
+		lck->file.path = talloc_strdup(lck, path);
+		NT_STATUS_HAVE_NO_MEMORY(lck->file.path);
+	}
+
 	/*
-	  possibly grant an exclusive or batch oplock if this is the only client
-	  with the file open. We don't yet grant levelII oplocks.
+	  possibly grant an exclusive, batch or level2 oplock
 	*/
-	if (oplock_granted != NULL) {
-		if ((oplock_level == OPLOCK_BATCH ||
-		     oplock_level == OPLOCK_EXCLUSIVE) &&
-		    file.num_entries == 0) {
-			(*oplock_granted) = oplock_level;
+	if (lck->can_open.attrs_only) {
+		oplock_level	= OPLOCK_NONE;
+	} else if (oplock_level == OPLOCK_EXCLUSIVE) {
+		if (lck->file.num_entries == 0) {
+			oplock_level	= OPLOCK_EXCLUSIVE;
+		} else if (allow_level_II_oplock) {
+			oplock_level	= OPLOCK_LEVEL_II;
+		} else {
+			oplock_level	= OPLOCK_NONE;
+		}
+	} else if (oplock_level == OPLOCK_BATCH) {
+		if (lck->file.num_entries == 0) {
+			oplock_level	= OPLOCK_BATCH;
+		} else if (allow_level_II_oplock) {
+			oplock_level	= OPLOCK_LEVEL_II;
+		} else {
+			oplock_level	= OPLOCK_NONE;
+		}
+	} else if (oplock_level == OPLOCK_LEVEL_II) {
+		oplock_level	= OPLOCK_LEVEL_II;
+	} else {
+		oplock_level	= OPLOCK_NONE;
+	}
+
+	lck->can_open.e->file_handle		= file_handle;
+	lck->can_open.e->fd			= fd;
+	lck->can_open.e->allow_level_II_oplock	= allow_level_II_oplock;
+	lck->can_open.e->oplock_level		= oplock_level;
+
+	if (odb->lease_ctx && fd) {
+		NTSTATUS status;
+		status = sys_lease_setup(odb->lease_ctx, lck->can_open.e);
+		NT_STATUS_NOT_OK_RETURN(status);
+	}
+
+	if (oplock_granted) {
+		if (lck->can_open.e->oplock_level == OPLOCK_EXCLUSIVE) {
+			*oplock_granted	= EXCLUSIVE_OPLOCK_RETURN;
+		} else if (lck->can_open.e->oplock_level == OPLOCK_BATCH) {
+			*oplock_granted	= BATCH_OPLOCK_RETURN;
+		} else if (lck->can_open.e->oplock_level == OPLOCK_LEVEL_II) {
+			*oplock_granted	= LEVEL_II_OPLOCK_RETURN;
 		} else {
-			(*oplock_granted) = OPLOCK_NONE;
+			*oplock_granted	= NO_OPLOCK_RETURN;
 		}
-		e.oplock_level = (*oplock_granted);
 	}
 
 	/* it doesn't conflict, so add it to the end */
-	file.entries = talloc_realloc(lck, file.entries, struct opendb_entry, 
-				      file.num_entries+1);
-	NT_STATUS_HAVE_NO_MEMORY(file.entries);
+	lck->file.entries = talloc_realloc(lck, lck->file.entries,
+					   struct opendb_entry,
+					   lck->file.num_entries+1);
+	NT_STATUS_HAVE_NO_MEMORY(lck->file.entries);
+
+	lck->file.entries[lck->file.num_entries] = *lck->can_open.e;
+	lck->file.num_entries++;
 
-	file.entries[file.num_entries] = e;
-	file.num_entries++;
+	talloc_free(lck->can_open.e);
+	lck->can_open.e = NULL;
 
-	return odb_push_record(lck, &file);
+	return odb_push_record(lck, &lck->file);
 }
 
 
@@ -374,71 +546,161 @@ static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
 static NTSTATUS odb_tdb_open_file_pending(struct odb_lock *lck, void *private)
 {
 	struct odb_context *odb = lck->odb;
-	struct opendb_file file;
-	NTSTATUS status;
-		
-	status = odb_pull_record(lck, &file);
-	NT_STATUS_NOT_OK_RETURN(status);
 
-	file.pending = talloc_realloc(lck, file.pending, struct opendb_pending, 
-				      file.num_pending+1);
-	NT_STATUS_HAVE_NO_MEMORY(file.pending);
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
 
-	file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
-	file.pending[file.num_pending].notify_ptr = private;
+	lck->file.pending = talloc_realloc(lck, lck->file.pending,
+					   struct opendb_pending,
+					   lck->file.num_pending+1);
+	NT_STATUS_HAVE_NO_MEMORY(lck->file.pending);
 
-	file.num_pending++;
+	lck->file.pending[lck->file.num_pending].server = odb->ntvfs_ctx->server_id;
+	lck->file.pending[lck->file.num_pending].notify_ptr = private;
 
-	return odb_push_record(lck, &file);
+	lck->file.num_pending++;
+
+	return odb_push_record(lck, &lck->file);
 }
 
 
 /*
   remove a opendb entry
 */
-static NTSTATUS odb_tdb_close_file(struct odb_lock *lck, void *file_handle)
+static NTSTATUS odb_tdb_close_file(struct odb_lock *lck, void *file_handle,
+				   const char **_delete_path)
 {
 	struct odb_context *odb = lck->odb;
-	struct opendb_file file;
+	const char *delete_path = NULL;
 	int i;
-	NTSTATUS status;
 
-	status = odb_pull_record(lck, &file);
-	NT_STATUS_NOT_OK_RETURN(status);
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
 
 	/* find the entry, and delete it */
-	for (i=0;i<file.num_entries;i++) {
-		if (file_handle == file.entries[i].file_handle &&
-		    cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
-			if (file.entries[i].delete_on_close) {
-				file.delete_on_close = true;
+	for (i=0;i<lck->file.num_entries;i++) {
+		if (file_handle == lck->file.entries[i].file_handle &&
+		    cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.entries[i].server)) {
+			if (lck->file.entries[i].delete_on_close) {
+				lck->file.delete_on_close = true;
+			}
+			if (odb->lease_ctx && lck->file.entries[i].fd) {
+				NTSTATUS status;
+				status = sys_lease_remove(odb->lease_ctx, &lck->file.entries[i]);
+				NT_STATUS_NOT_OK_RETURN(status);
 			}
-			if (i < file.num_entries-1) {
-				memmove(file.entries+i, file.entries+i+1, 
-					(file.num_entries - (i+1)) * 
+			if (i < lck->file.num_entries-1) {
+				memmove(lck->file.entries+i, lck->file.entries+i+1,
+					(lck->file.num_entries - (i+1)) *
 					sizeof(struct opendb_entry));
 			}
 			break;
 		}
 	}
 
-	if (i == file.num_entries) {
+	if (i == lck->file.num_entries) {
 		return NT_STATUS_UNSUCCESSFUL;
 	}
 
 	/* send any pending notifications, removing them once sent */
-	for (i=0;i<file.num_pending;i++) {
-		messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server, 
-				   MSG_PVFS_RETRY_OPEN, 
-				   file.pending[i].notify_ptr);
+	for (i=0;i<lck->file.num_pending;i++) {
+		messaging_send_ptr(odb->ntvfs_ctx->msg_ctx,
+				   lck->file.pending[i].server,
+				   MSG_PVFS_RETRY_OPEN,
+				   lck->file.pending[i].notify_ptr);
 	}
-	file.num_pending = 0;
+	lck->file.num_pending = 0;
 
-	file.num_entries--;
-	
-	return odb_push_record(lck, &file);
+	lck->file.num_entries--;
+
+	if (lck->file.num_entries == 0 && lck->file.delete_on_close) {
+		delete_path = lck->file.path;
+	}
+
+	if (_delete_path) {
+		*_delete_path = delete_path;
+	}
+
+	return odb_push_record(lck, &lck->file);
+}
+
+/*
+  update the oplock level of the client
+*/
+static NTSTATUS odb_tdb_update_oplock(struct odb_lock *lck, void *file_handle,
+				      uint32_t oplock_level)
+{
+	struct odb_context *odb = lck->odb;
+	int i;
+
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
+
+	/* find the entry, and update it */
+	for (i=0;i<lck->file.num_entries;i++) {
+		if (file_handle == lck->file.entries[i].file_handle &&
+		    cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.entries[i].server)) {
+			lck->file.entries[i].oplock_level = oplock_level;
+
+			if (odb->lease_ctx && lck->file.entries[i].fd) {
+				NTSTATUS status;
+				status = sys_lease_update(odb->lease_ctx, &lck->file.entries[i]);
+				NT_STATUS_NOT_OK_RETURN(status);
+			}
+
+			break;
+		}
+	}
+
+	if (i == lck->file.num_entries) {
+		return NT_STATUS_UNSUCCESSFUL;
+	}
+
+	/* send any pending notifications, removing them once sent */
+	for (i=0;i<lck->file.num_pending;i++) {
+		messaging_send_ptr(odb->ntvfs_ctx->msg_ctx,
+				   lck->file.pending[i].server,
+				   MSG_PVFS_RETRY_OPEN,
+				   lck->file.pending[i].notify_ptr);
+	}
+	lck->file.num_pending = 0;
+
+	return odb_push_record(lck, &lck->file);
 }
 
+/*
+  send oplocks breaks to none to all level2 holders
+*/
+static NTSTATUS odb_tdb_break_oplocks(struct odb_lock *lck)
+{
+	struct odb_context *odb = lck->odb;
+	int i;
+	bool modified = false;
+
+	/* see if anyone has an oplock, which we need to break */
+	for (i=0;i<lck->file.num_entries;i++) {
+		if (lck->file.entries[i].oplock_level == OPLOCK_LEVEL_II) {
+			/*
+			 * there could be multiple level2 oplocks
+			 * and we just send a break to none to all of them
+			 * without waiting for a release
+			 */
+			odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+					      &lck->file.entries[i],
+					      OPLOCK_BREAK_TO_NONE);
+			lck->file.entries[i].oplock_level = OPLOCK_NONE;
+			modified = true;
+		}
+	}
+
+	if (modified) {
+		return odb_push_record(lck, &lck->file);
+	}
+	return NT_STATUS_OK;
+}
 
 /*
   remove a pending opendb entry
@@ -447,32 +709,31 @@ static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private)
 {
 	struct odb_context *odb = lck->odb;
 	int i;
-	NTSTATUS status;
-	struct opendb_file file;
 
-	status = odb_pull_record(lck, &file);
-	NT_STATUS_NOT_OK_RETURN(status);
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
 
 	/* find the entry, and delete it */
-	for (i=0;i<file.num_pending;i++) {
-		if (private == file.pending[i].notify_ptr &&
-		    cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
-			if (i < file.num_pending-1) {
-				memmove(file.pending+i, file.pending+i+1, 
-					(file.num_pending - (i+1)) * 
+	for (i=0;i<lck->file.num_pending;i++) {
+		if (private == lck->file.pending[i].notify_ptr &&
+		    cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.pending[i].server)) {
+			if (i < lck->file.num_pending-1) {
+				memmove(lck->file.pending+i, lck->file.pending+i+1,
+					(lck->file.num_pending - (i+1)) *
 					sizeof(struct opendb_pending));
 			}
 			break;
 		}
 	}
 
-	if (i == file.num_pending) {
+	if (i == lck->file.num_pending) {
 		return NT_STATUS_UNSUCCESSFUL;
 	}
 
-	file.num_pending--;
+	lck->file.num_pending--;
 	
-	return odb_push_record(lck, &file);
+	return odb_push_record(lck, &lck->file);
 }
 
 
@@ -481,18 +742,32 @@ static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private)
 */
 static NTSTATUS odb_tdb_rename(struct odb_lock *lck, const char *path)
 {
-	struct opendb_file file;
-	NTSTATUS status;
-
-	status = odb_pull_record(lck, &file);
-	if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
+	if (lck->file.path == NULL) {
 		/* not having the record at all is OK */
 		return NT_STATUS_OK;
 	}
-	NT_STATUS_NOT_OK_RETURN(status);
 
-	file.path = path;
-	return odb_push_record(lck, &file);
+	lck->file.path = talloc_strdup(lck, path);
+	NT_STATUS_HAVE_NO_MEMORY(lck->file.path);
+
+	return odb_push_record(lck, &lck->file);
+}
+
+/*
+  get the path of an open file
+*/
+static NTSTATUS odb_tdb_get_path(struct odb_lock *lck, const char **path)
+{
+	*path = NULL;
+
+	/* we don't ignore NT_STATUS_OBJECT_NAME_NOT_FOUND here */
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
+
+	*path = lck->file.path;
+
+	return NT_STATUS_OK;
 }
 
 /*
@@ -500,15 +775,13 @@ static NTSTATUS odb_tdb_rename(struct odb_lock *lck, const char *path)
 */
 static NTSTATUS odb_tdb_set_delete_on_close(struct odb_lock *lck, bool del_on_close)
 {
-	NTSTATUS status;
-	struct opendb_file file;
-
-	status = odb_pull_record(lck, &file);
-	NT_STATUS_NOT_OK_RETURN(status);
+	if (lck->file.path == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+	}
 
-	file.delete_on_close = del_on_close;
+	lck->file.delete_on_close = del_on_close;
 
-	return odb_push_record(lck, &file);
+	return odb_push_record(lck, &lck->file);
 }
 
 /*
@@ -516,38 +789,16 @@ static NTSTATUS odb_tdb_set_delete_on_close(struct odb_lock *lck, bool del_on_cl
   people still have the file open
 */
 static NTSTATUS odb_tdb_get_delete_on_close(struct odb_context *odb, 
-					    DATA_BLOB *key, bool *del_on_close, 
-					    int *open_count, char **path)
+					    DATA_BLOB *key, bool *del_on_close)
 {
-	NTSTATUS status;
-	struct opendb_file file;
 	struct odb_lock *lck;
 
+	(*del_on_close) = false;
+
 	lck = odb_lock(odb, odb, key);
 	NT_STATUS_HAVE_NO_MEMORY(lck);
 
-	status = odb_pull_record(lck, &file);
-	if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
-		talloc_free(lck);
-		(*del_on_close) = false;
-		return NT_STATUS_OK;
-	}
-	if (!NT_STATUS_IS_OK(status)) {
-		talloc_free(lck);
-		return status;
-	}
-
-	(*del_on_close) = file.delete_on_close;
-	if (open_count != NULL) {
-		(*open_count) = file.num_entries;
-	}
-	if (path != NULL) {
-		*path = talloc_strdup(odb, file.path);
-		NT_STATUS_HAVE_NO_MEMORY(*path);
-		if (file.num_entries == 1 && file.entries[0].delete_on_close) {
-			(*del_on_close) = true;
-		}
-	}
+	(*del_on_close) = lck->file.delete_on_close;
 
 	talloc_free(lck);
 
@@ -560,47 +811,31 @@ static NTSTATUS odb_tdb_get_delete_on_close(struct odb_context *odb,
   create_options and access_mask
 */
 static NTSTATUS odb_tdb_can_open(struct odb_lock *lck,
-				 uint32_t share_access, uint32_t create_options, 
-				 uint32_t access_mask)
+				 uint32_t stream_id, uint32_t share_access,
+				 uint32_t access_mask, bool delete_on_close,
+				 uint32_t open_disposition, bool break_to_none)
 {
 	struct odb_context *odb = lck->odb;
 	NTSTATUS status;
-	struct opendb_file file;
-	struct opendb_entry e;
-	int i;
 
-	status = odb_pull_record(lck, &file);
-	if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
-		return NT_STATUS_OK;
-	}
+	status = odb_tdb_open_can_internal(odb, &lck->file, stream_id,
+					   share_access, access_mask,
+					   delete_on_close, open_disposition,
+					   break_to_none, &lck->can_open.attrs_only);
 	NT_STATUS_NOT_OK_RETURN(status);
 
-	if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) && 
-	    file.num_entries != 0) {
-		return NT_STATUS_SHARING_VIOLATION;
-	}
+	lck->can_open.e	= talloc(lck, struct opendb_entry);
+	NT_STATUS_HAVE_NO_MEMORY(lck->can_open.e);
 
-	if (file.delete_on_close) {
-		return NT_STATUS_DELETE_PENDING;
-	}
-
-	e.server       = odb->ntvfs_ctx->server_id;
-	e.file_handle  = NULL;
-	e.stream_id    = 0;
-	e.share_access = share_access;
-	e.access_mask  = access_mask;
-		
-	for (i=0;i<file.num_entries;i++) {
-		status = share_conflict(&file.entries[i], &e);
-		if (!NT_STATUS_IS_OK(status)) {
-			/* note that we discard the error code
-			   here. We do this as unless we are actually
-			   doing an open (which comes via a different
-			   function), we need to return a sharing
-			   violation */
-			return NT_STATUS_SHARING_VIOLATION;
-		}
-	}
+	lck->can_open.e->server			= odb->ntvfs_ctx->server_id;
+	lck->can_open.e->file_handle		= NULL;
+	lck->can_open.e->fd			= NULL;
+	lck->can_open.e->stream_id		= stream_id;
+	lck->can_open.e->share_access		= share_access;
+	lck->can_open.e->access_mask		= access_mask;
+	lck->can_open.e->delete_on_close	= delete_on_close;
+	lck->can_open.e->allow_level_II_oplock	= false;
+	lck->can_open.e->oplock_level		= OPLOCK_NONE;
 
 	return NT_STATUS_OK;
 }
@@ -609,18 +844,23 @@ static NTSTATUS odb_tdb_can_open(struct odb_lock *lck,
 static const struct opendb_ops opendb_tdb_ops = {
 	.odb_init                = odb_tdb_init,
 	.odb_lock                = odb_tdb_lock,
+	.odb_get_key             = odb_tdb_get_key,
 	.odb_open_file           = odb_tdb_open_file,
 	.odb_open_file_pending   = odb_tdb_open_file_pending,
 	.odb_close_file          = odb_tdb_close_file,
 	.odb_remove_pending      = odb_tdb_remove_pending,
 	.odb_rename              = odb_tdb_rename,
+	.odb_get_path            = odb_tdb_get_path,
 	.odb_set_delete_on_close = odb_tdb_set_delete_on_close,
 	.odb_get_delete_on_close = odb_tdb_get_delete_on_close,
-	.odb_can_open            = odb_tdb_can_open
+	.odb_can_open            = odb_tdb_can_open,
+	.odb_update_oplock       = odb_tdb_update_oplock,
+	.odb_break_oplocks       = odb_tdb_break_oplocks
 };
 
 
 void odb_tdb_init_ops(void)
 {
+	sys_lease_init();
 	odb_set_ops(&opendb_tdb_ops);
 }