added a ctdb_ltdb_lock_fetch_requeue() function

this will be the core of the non-blocking lock idea for ctdb, it will be used
in place of ctdb_ltdb_fetch(), but will also get a lock. It re-starts a request
if it needs to block
(This used to be ctdb commit afa479026cf6293e6a878c8a329cdac035284672)

diff --git a/ctdb/Makefile.in b/ctdb/Makefile.in
index 93e0c7d2bdc7116ae5549f22296d32fe0bb6b623..52659520d528536d7b1afbe31cdea0ae13b80739 100644 (file)
--- a/ctdb/Makefile.in
+++ b/ctdb/Makefile.in
@@ -21,7 +21,7 @@ LIB_FLAGS=@LDFLAGS@ -Llib @LIBS@ -lpopt @INFINIBAND_LIBS@
 EVENTS_OBJ = lib/events/events.o lib/events/events_standard.o
 
 CTDB_COMMON_OBJ = common/ctdb.o common/ctdb_daemon.o common/ctdb_client.o common/ctdb_io.o common/util.o common/ctdb_util.o \
-       common/ctdb_call.o common/ctdb_ltdb.o common/ctdb_message.o \
+       common/ctdb_call.o common/ctdb_ltdb.o common/ctdb_lockwait.o common/ctdb_message.o \
        lib/util/idtree.o lib/util/db_wrap.o
 
 CTDB_TCP_OBJ = tcp/tcp_connect.o tcp/tcp_io.o tcp/tcp_init.o

diff --git a/ctdb/common/ctdb.c b/ctdb/common/ctdb.c
index 8a8d52f3f1fe2af6b2fd57db22453ffd10a4c38a..559b0ed9a637683b0a10aba717d3ed921e2de1be 100644 (file)
--- a/ctdb/common/ctdb.c
+++ b/ctdb/common/ctdb.c
@@ -190,7 +190,7 @@ uint32_t ctdb_get_num_nodes(struct ctdb_context *ctdb)
 /*
   called by the transport layer when a packet comes in
 */
-static void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
+void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
 {
        struct ctdb_req_header *hdr;
 

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index 785ccad9b39478c29e3fde87fcfb72597ae54986..b18f26748076b53392506d28a4fabc8517a37328 100644 (file)
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -215,3 +215,49 @@ int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
        return tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 }
 
+/*
+  called when we should retry the operation
+ */
+static void lock_fetch_callback(void *p)
+{
+       struct ctdb_req_header *hdr = p;
+       struct ctdb_context *ctdb = talloc_find_parent_bytype(p, struct ctdb_context);
+       ctdb_recv_pkt(ctdb, (uint8_t *)hdr, hdr->length);
+       printf("PACKET REQUEUED\n");
+}
+
+/*
+  do a non-blocking ltdb_fetch with a locked record, deferring this
+  ctdb request until we have the chainlock
+ */
+int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db, 
+                                TDB_DATA key, struct ctdb_ltdb_header *header, 
+                                struct ctdb_req_header *hdr, TDB_DATA *data)
+{
+       int ret;
+       struct tdb_context *tdb = ctdb_db->ltdb->tdb;
+       struct lockwait_handle *h;
+       
+       ret = tdb_chainlock_nonblock(tdb, key);
+
+       /* first the non-contended path */
+       if (ret == 0) {
+               ret = ctdb_ltdb_fetch(ctdb_db, key, header, hdr, data);
+               if (ret != 0) {
+                       tdb_chainunlock(tdb, key);
+               }       
+               return ret;
+       }
+
+       /* now the contended path */
+       h = ctdb_lockwait(ctdb_db, key, lock_fetch_callback, hdr);
+       if (h == NULL) {
+               tdb_chainunlock(tdb, key);
+               return -1;
+       }
+
+       /* we get an extra reference to the packet here, to 
+          stop it being freed in the top level packet handler */
+       (void)talloc_reference(ctdb_db, hdr);
+       return 0;
+}

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index c50b481cf35124cbdb99f396e66a0f124c253ca0..53ab5a83efdfab8efd757ff8ad5af77d2fe68ec7 100644 (file)
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -353,6 +353,10 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
 int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
                    struct ctdb_ltdb_header *header, TDB_DATA data);
 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db, 
+                                TDB_DATA key, struct ctdb_ltdb_header *header, 
+                                struct ctdb_req_header *hdr, TDB_DATA *data);
+void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length);
 
 struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, 
                                             struct ctdb_call *call,
@@ -449,4 +453,8 @@ struct ctdb_record_handle *ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_d
 */
 int ctdb_client_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data);
 
+struct lockwait_handle *ctdb_lockwait(struct ctdb_db_context *ctdb_db,
+                                     TDB_DATA key,
+                                     void (*callback)(void *), void *private_data);
+
 #endif