run the "init" event before we freeze the databases
[metze/ctdb/wip.git] / server / ctdb_daemon.c
index 83fcc0aa53779eba88fb8dd66262a90c6fbd684a..5d73b0d5d9b99a21dba6cc7cb3246b99949f6a08 100644 (file)
 #include "includes.h"
 #include "db_wrap.h"
 #include "lib/tdb/include/tdb.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "lib/util/dlinklist.h"
 #include "system/network.h"
 #include "system/filesys.h"
 #include "system/wait.h"
-#include "../include/ctdb.h"
+#include "../include/ctdb_client.h"
 #include "../include/ctdb_private.h"
 #include <sys/socket.h>
 
-static void daemon_incoming_packet(void *, struct ctdb_req_header *);
-
-/*
-  handler for when a node changes its flags
-*/
-static void flag_change_handler(struct ctdb_context *ctdb, uint64_t srvid, 
-                               TDB_DATA data, void *private_data)
-{
-       struct ctdb_node_flag_change *c = (struct ctdb_node_flag_change *)data.dptr;
-
-       if (data.dsize != sizeof(*c) || !ctdb_validate_pnn(ctdb, c->pnn)) {
-               DEBUG(DEBUG_CRIT,(__location__ "Invalid data in ctdb_node_flag_change\n"));
-               return;
-       }
-
-       if (!ctdb_validate_pnn(ctdb, c->pnn)) {
-               DEBUG(DEBUG_CRIT,("Bad pnn %u in flag_change_handler\n", c->pnn));
-               return;
-       }
-
-       /* don't get the disconnected flag from the other node */
-       ctdb->nodes[c->pnn]->flags = 
-               (ctdb->nodes[c->pnn]->flags&NODE_FLAGS_DISCONNECTED) 
-               | (c->new_flags & ~NODE_FLAGS_DISCONNECTED);    
-       DEBUG(DEBUG_INFO,("Node flags for node %u are now 0x%x\n", c->pnn, ctdb->nodes[c->pnn]->flags));
+struct ctdb_client_pid_list {
+       struct ctdb_client_pid_list *next, *prev;
+       struct ctdb_context *ctdb;
+       pid_t pid;
+       struct ctdb_client *client;
+};
 
-       /* make sure we don't hold any IPs when we shouldn't */
-       if (c->pnn == ctdb->pnn &&
-           (ctdb->nodes[c->pnn]->flags & (NODE_FLAGS_INACTIVE|NODE_FLAGS_BANNED))) {
-               ctdb_release_all_ips(ctdb);
-       }
-}
+static void daemon_incoming_packet(void *, struct ctdb_req_header *);
 
 static void print_exit_message(void)
 {
        DEBUG(DEBUG_NOTICE,("CTDB daemon shutting down\n"));
 }
 
-
 /* called when the "startup" event script has finished */
 static void ctdb_start_transport(struct ctdb_context *ctdb)
 {
@@ -91,10 +66,6 @@ static void ctdb_start_transport(struct ctdb_context *ctdb)
        /* Make sure we log something when the daemon terminates */
        atexit(print_exit_message);
 
-       /* a handler for when nodes are disabled/enabled */
-       ctdb_register_message_handler(ctdb, ctdb, CTDB_SRVID_NODE_FLAGS_CHANGED, 
-                                     flag_change_handler, NULL);
-
        /* start monitoring for connected/disconnected nodes */
        ctdb_start_keepalive(ctdb);
 
@@ -103,6 +74,9 @@ static void ctdb_start_transport(struct ctdb_context *ctdb)
 
        /* start periodic update of tcp tickle lists */
                ctdb_start_tcp_tickle_update(ctdb);
+
+       /* start listening for recovery daemon pings */
+       ctdb_control_recd_ping(ctdb);
 }
 
 static void block_signal(int signum)
@@ -124,6 +98,13 @@ static void block_signal(int signum)
 static int daemon_queue_send(struct ctdb_client *client, struct ctdb_req_header *hdr)
 {
        client->ctdb->statistics.client_packets_sent++;
+       if (hdr->operation == CTDB_REQ_MESSAGE) {
+               if (ctdb_queue_length(client->queue) > client->ctdb->tunable.max_queue_depth_drop_msg) {
+                       DEBUG(DEBUG_ERR,("CTDB_REQ_MESSAGE queue full - killing client connection.\n"));
+                       talloc_free(client);
+                       return -1;
+               }
+       }
        return ctdb_queue_send(client->queue, (uint8_t *)hdr, hdr->length);
 }
 
@@ -154,7 +135,6 @@ static void daemon_message_handler(struct ctdb_context *ctdb, uint64_t srvid,
 
        talloc_free(r);
 }
-                                          
 
 /*
   this is called when the ctdb daemon received a ctdb request to 
@@ -177,13 +157,6 @@ int daemon_register_message_handler(struct ctdb_context *ctdb, uint32_t client_i
                         (unsigned long long)srvid));
        }
 
-       /* this is a hack for Samba - we now know the pid of the Samba client */
-       if ((srvid & 0xFFFFFFFF) == srvid &&
-           kill(srvid, 0) == 0) {
-               client->pid = srvid;
-               DEBUG(DEBUG_INFO,(__location__ " Registered PID %u for client %u\n",
-                        (unsigned)client->pid, client_id));
-       }
        return res;
 }
 
@@ -207,9 +180,26 @@ int daemon_deregister_message_handler(struct ctdb_context *ctdb, uint32_t client
 */
 static int ctdb_client_destructor(struct ctdb_client *client)
 {
+       struct ctdb_db_context *ctdb_db;
+
        ctdb_takeover_client_destructor_hook(client);
        ctdb_reqid_remove(client->ctdb, client->client_id);
-       client->ctdb->statistics.num_clients--;
+       if (client->ctdb->statistics.num_clients) {
+               client->ctdb->statistics.num_clients--;
+       }
+
+       if (client->num_persistent_updates != 0) {
+               DEBUG(DEBUG_ERR,(__location__ " Client disconnecting with %u persistent updates in flight. Starting recovery\n", client->num_persistent_updates));
+               client->ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
+       }
+       ctdb_db = find_ctdb_db(client->ctdb, client->db_id);
+       if (ctdb_db) {
+               DEBUG(DEBUG_ERR, (__location__ " client exit while transaction "
+                                 "commit active. Forcing recovery.\n"));
+               client->ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
+               ctdb_db->transaction_active = false;
+       }
+
        return 0;
 }
 
@@ -260,6 +250,7 @@ static void daemon_call_from_client_callback(struct ctdb_call_state *state)
        int res;
        uint32_t length;
        struct ctdb_client *client = dstate->client;
+       struct ctdb_db_context *ctdb_db = state->ctdb_db;
 
        talloc_steal(client, dstate);
        talloc_steal(dstate, dstate->call);
@@ -267,8 +258,10 @@ static void daemon_call_from_client_callback(struct ctdb_call_state *state)
        res = ctdb_daemon_call_recv(state, dstate->call);
        if (res != 0) {
                DEBUG(DEBUG_ERR, (__location__ " ctdbd_call_recv() returned error\n"));
-               client->ctdb->statistics.pending_calls--;
-               ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       client->ctdb->statistics.pending_calls--;
+               }
+               ctdb_latency(ctdb_db, "call_from_client_cb 1", &client->ctdb->statistics.max_call_latency, dstate->start_time);
                return;
        }
 
@@ -277,8 +270,10 @@ static void daemon_call_from_client_callback(struct ctdb_call_state *state)
                               length, struct ctdb_reply_call);
        if (r == NULL) {
                DEBUG(DEBUG_ERR, (__location__ " Failed to allocate reply_call in ctdb daemon\n"));
-               client->ctdb->statistics.pending_calls--;
-               ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       client->ctdb->statistics.pending_calls--;
+               }
+               ctdb_latency(ctdb_db, "call_from_client_cb 2", &client->ctdb->statistics.max_call_latency, dstate->start_time);
                return;
        }
        r->hdr.reqid        = dstate->reqid;
@@ -286,12 +281,18 @@ static void daemon_call_from_client_callback(struct ctdb_call_state *state)
        memcpy(&r->data[0], dstate->call->reply_data.dptr, r->datalen);
 
        res = daemon_queue_send(client, &r->hdr);
+       if (res == -1) {
+               /* client is dead - return immediately */
+               return;
+       }
        if (res != 0) {
                DEBUG(DEBUG_ERR, (__location__ " Failed to queue packet from daemon to client\n"));
        }
-       ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
+       ctdb_latency(ctdb_db, "call_from_client_cb 3", &client->ctdb->statistics.max_call_latency, dstate->start_time);
        talloc_free(dstate);
-       client->ctdb->statistics.pending_calls--;
+       if (client->ctdb->statistics.pending_calls > 0) {
+               client->ctdb->statistics.pending_calls--;
+       }
 }
 
 struct ctdb_daemon_packet_wrap {
@@ -344,16 +345,30 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
        struct ctdb_daemon_packet_wrap *w;
 
        ctdb->statistics.total_calls++;
-       ctdb->statistics.pending_calls++;
+       if (client->ctdb->statistics.pending_calls > 0) {
+               ctdb->statistics.pending_calls++;
+       }
 
        ctdb_db = find_ctdb_db(client->ctdb, c->db_id);
        if (!ctdb_db) {
                DEBUG(DEBUG_ERR, (__location__ " Unknown database in request. db_id==0x%08x",
                          c->db_id));
-               ctdb->statistics.pending_calls--;
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
                return;
        }
 
+       if (ctdb_db->unhealthy_reason) {
+               /*
+                * this is just a warning, as the tdb should be empty anyway,
+                * and only persistent databases can be unhealthy, which doesn't
+                * use this code patch
+                */
+               DEBUG(DEBUG_WARNING,("warn: db(%s) unhealty in daemon_request_call_from_client(): %s\n",
+                                    ctdb_db->db_name, ctdb_db->unhealthy_reason));
+       }
+
        key.dptr = c->data;
        key.dsize = c->keylen;
 
@@ -368,7 +383,9 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
                                           daemon_incoming_packet_wrap, w, True);
        if (ret == -2) {
                /* will retry later */
-               ctdb->statistics.pending_calls--;
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
                return;
        }
 
@@ -376,15 +393,23 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
 
        if (ret != 0) {
                DEBUG(DEBUG_ERR,(__location__ " Unable to fetch record\n"));
-               ctdb->statistics.pending_calls--;
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
                return;
        }
 
        dstate = talloc(client, struct daemon_call_state);
        if (dstate == NULL) {
-               ctdb_ltdb_unlock(ctdb_db, key);
+               ret = ctdb_ltdb_unlock(ctdb_db, key);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+               }
+
                DEBUG(DEBUG_ERR,(__location__ " Unable to allocate dstate\n"));
-               ctdb->statistics.pending_calls--;
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
                return;
        }
        dstate->start_time = timeval_current();
@@ -394,10 +419,16 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
 
        call = dstate->call = talloc_zero(dstate, struct ctdb_call);
        if (call == NULL) {
-               ctdb_ltdb_unlock(ctdb_db, key);
+               ret = ctdb_ltdb_unlock(ctdb_db, key);
+               if (ret != 0) {
+                       DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+               }
+
                DEBUG(DEBUG_ERR,(__location__ " Unable to allocate call\n"));
-               ctdb->statistics.pending_calls--;
-               ctdb_latency(&ctdb->statistics.max_call_latency, dstate->start_time);
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
+               ctdb_latency(ctdb_db, "call_from_client 1", &ctdb->statistics.max_call_latency, dstate->start_time);
                return;
        }
 
@@ -413,12 +444,17 @@ static void daemon_request_call_from_client(struct ctdb_client *client,
                state = ctdb_daemon_call_send_remote(ctdb_db, call, &header);
        }
 
-       ctdb_ltdb_unlock(ctdb_db, key);
+       ret = ctdb_ltdb_unlock(ctdb_db, key);
+       if (ret != 0) {
+               DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
+       }
 
        if (state == NULL) {
                DEBUG(DEBUG_ERR,(__location__ " Unable to setup call send\n"));
-               ctdb->statistics.pending_calls--;
-               ctdb_latency(&ctdb->statistics.max_call_latency, dstate->start_time);
+               if (client->ctdb->statistics.pending_calls > 0) {
+                       ctdb->statistics.pending_calls--;
+               }
+               ctdb_latency(ctdb_db, "call_from_client 2", &ctdb->statistics.max_call_latency, dstate->start_time);
                return;
        }
        talloc_steal(state, dstate);
@@ -526,14 +562,26 @@ static void ctdb_daemon_read_cb(uint8_t *data, size_t cnt, void *args)
        daemon_incoming_packet(client, hdr);
 }
 
+
+static int ctdb_clientpid_destructor(struct ctdb_client_pid_list *client_pid)
+{
+       if (client_pid->ctdb->client_pids != NULL) {
+               DLIST_REMOVE(client_pid->ctdb->client_pids, client_pid);
+       }
+
+       return 0;
+}
+
+
 static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde, 
                         uint16_t flags, void *private_data)
 {
-       struct sockaddr_in addr;
+       struct sockaddr_un addr;
        socklen_t len;
        int fd;
        struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
        struct ctdb_client *client;
+       struct ctdb_client_pid_list *client_pid;
 #ifdef _AIX
        struct peercred_struct cr;
        socklen_t crl = sizeof(struct peercred_struct);
@@ -552,24 +600,42 @@ static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde,
        set_nonblocking(fd);
        set_close_on_exec(fd);
 
+       DEBUG(DEBUG_DEBUG,(__location__ " Created SOCKET FD:%d to connected child\n", fd));
+
        client = talloc_zero(ctdb, struct ctdb_client);
 #ifdef _AIX
        if (getsockopt(fd, SOL_SOCKET, SO_PEERID, &cr, &crl) == 0) {
 #else
        if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &cr, &crl) == 0) {
 #endif
-               talloc_asprintf(client, "struct ctdb_client: pid:%u", (unsigned)cr.pid);
+               DEBUG(DEBUG_INFO,("Connected client with pid:%u\n", (unsigned)cr.pid));
        }
 
        client->ctdb = ctdb;
        client->fd = fd;
        client->client_id = ctdb_reqid_new(ctdb, client);
-       ctdb->statistics.num_clients++;
+       client->pid = cr.pid;
+
+       client_pid = talloc(client, struct ctdb_client_pid_list);
+       if (client_pid == NULL) {
+               DEBUG(DEBUG_ERR,("Failed to allocate client pid structure\n"));
+               close(fd);
+               talloc_free(client);
+               return;
+       }               
+       client_pid->ctdb   = ctdb;
+       client_pid->pid    = cr.pid;
+       client_pid->client = client;
+
+       DLIST_ADD(ctdb->client_pids, client_pid);
 
        client->queue = ctdb_queue_setup(ctdb, client, fd, CTDB_DS_ALIGNMENT, 
-                                        ctdb_daemon_read_cb, client);
+                                        ctdb_daemon_read_cb, client,
+                                        "client-%u", client->pid);
 
        talloc_set_destructor(client, ctdb_client_destructor);
+       talloc_set_destructor(client_pid, ctdb_clientpid_destructor);
+       ctdb->statistics.num_clients++;
 }
 
 
@@ -606,7 +672,7 @@ static int ux_socket_bind(struct ctdb_context *ctdb)
        } 
 
 
-       if (listen(ctdb->daemon.sd, 10) != 0) {
+       if (listen(ctdb->daemon.sd, 100) != 0) {
                DEBUG(DEBUG_CRIT,("Unable to listen on ctdb socket '%s'\n", ctdb->daemon.name));
                goto failed;
        }
@@ -619,23 +685,52 @@ failed:
        return -1;      
 }
 
-/*
-  delete the socket on exit - called on destruction of autofree context
- */
-static int unlink_destructor(const char *name)
+static void sig_child_handler(struct event_context *ev,
+       struct signal_event *se, int signum, int count,
+       void *dont_care, 
+       void *private_data)
 {
-       unlink(name);
-       return 0;
+//     struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
+       int status;
+       pid_t pid = -1;
+
+       while (pid != 0) {
+               pid = waitpid(-1, &status, WNOHANG);
+               if (pid == -1) {
+                       DEBUG(DEBUG_ERR, (__location__ " waitpid() returned error. errno:%d\n", errno));
+                       return;
+               }
+               if (pid > 0) {
+                       DEBUG(DEBUG_DEBUG, ("SIGCHLD from %d\n", (int)pid));
+               }
+       }
+}
+
+static void ctdb_setup_event_callback(struct ctdb_context *ctdb, int status,
+                                     void *private_data)
+{
+       if (status != 0) {
+               ctdb_fatal(ctdb, "Failed to run setup event\n");
+               return;
+       }
+       ctdb_run_notification_script(ctdb, "setup");
+
+       /* tell all other nodes we've just started up */
+       ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL,
+                                0, CTDB_CONTROL_STARTUP, 0,
+                                CTDB_CTRL_FLAG_NOREPLY,
+                                tdb_null, NULL, NULL);
 }
 
 /*
   start the protocol going as a daemon
 */
-int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork)
+int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork, bool use_syslog)
 {
        int res, ret = -1;
        struct fd_event *fde;
        const char *domain_socket_name;
+       struct signal_event *se;
 
        /* get rid of any old sockets */
        unlink(ctdb->daemon.name);
@@ -663,16 +758,22 @@ int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork)
        }
        block_signal(SIGPIPE);
 
-       if (ctdb->do_setsched) {
-               /* try to set us up as realtime */
-               ctdb_set_scheduler(ctdb);
-       }
+       ctdbd_pid = getpid();
+
+
+       DEBUG(DEBUG_ERR, ("Starting CTDBD as pid : %u\n", ctdbd_pid));
+
+       ctdb_high_priority(ctdb);
 
        /* ensure the socket is deleted on exit of the daemon */
        domain_socket_name = talloc_strdup(talloc_autofree_context(), ctdb->daemon.name);
-       talloc_set_destructor(domain_socket_name, unlink_destructor);   
+       if (domain_socket_name == NULL) {
+               DEBUG(DEBUG_ALERT,(__location__ " talloc_strdup failed.\n"));
+               exit(12);
+       }
 
        ctdb->ev = event_context_init(NULL);
+       tevent_loop_allow_nesting(ctdb->ev);
 
        ctdb_set_child_logging(ctdb);
 
@@ -704,26 +805,27 @@ int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork)
                ctdb_fatal(ctdb, "transport failed to initialise");
        }
 
-       /* attach to any existing persistent databases */
-       if (ctdb_attach_persistent(ctdb) != 0) {
-               ctdb_fatal(ctdb, "Failed to attach to persistent databases\n");         
+       /* attach to existing databases */
+       if (ctdb_attach_databases(ctdb) != 0) {
+               ctdb_fatal(ctdb, "Failed to attach to databases\n");
+       }
+
+       ret = ctdb_event_script(ctdb, CTDB_EVENT_INIT);
+       if (ret != 0) {
+               ctdb_fatal(ctdb, "Failed to run init event\n");
        }
+       ctdb_run_notification_script(ctdb, "init");
 
        /* start frozen, then let the first election sort things out */
-       if (!ctdb_blocking_freeze(ctdb)) {
+       if (ctdb_blocking_freeze(ctdb)) {
                ctdb_fatal(ctdb, "Failed to get initial freeze\n");
        }
 
        /* now start accepting clients, only can do this once frozen */
        fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, 
-                          EVENT_FD_READ|EVENT_FD_AUTOCLOSE, 
+                          EVENT_FD_READ,
                           ctdb_accept_client, ctdb);
-
-       /* tell all other nodes we've just started up */
-       ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL,
-                                0, CTDB_CONTROL_STARTUP, 0,
-                                CTDB_CTRL_FLAG_NOREPLY,
-                                tdb_null, NULL, NULL);
+       tevent_fd_set_auto_close(fde);
 
        /* release any IPs we hold from previous runs of the daemon */
        ctdb_release_all_ips(ctdb);
@@ -731,9 +833,37 @@ int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork)
        /* start the transport going */
        ctdb_start_transport(ctdb);
 
-       /* we dont want any SIGCHLD */
-       signal(SIGCHLD, SIG_IGN);
+       /* set up a handler to pick up sigchld */
+       se = event_add_signal(ctdb->ev, ctdb,
+                                    SIGCHLD, 0,
+                                    sig_child_handler,
+                                    ctdb);
+       if (se == NULL) {
+               DEBUG(DEBUG_CRIT,("Failed to set up signal handler for SIGCHLD\n"));
+               exit(1);
+       }
+
+       ret = ctdb_event_script_callback(ctdb,
+                                        ctdb,
+                                        ctdb_setup_event_callback,
+                                        ctdb,
+                                        false,
+                                        CTDB_EVENT_SETUP,
+                                        "");
+       if (ret != 0) {
+               DEBUG(DEBUG_CRIT,("Failed to set up 'setup' event\n"));
+               exit(1);
+       }
 
+       if (use_syslog) {
+               if (start_syslog_daemon(ctdb)) {
+                       DEBUG(DEBUG_CRIT, ("Failed to start syslog daemon\n"));
+                       exit(10);
+               }
+       }
+
+       ctdb_lockdown_memory(ctdb);
+         
        /* go into a wait loop to allow other nodes to complete */
        event_loop_wait(ctdb->ev);
 
@@ -801,6 +931,7 @@ static void daemon_control_callback(struct ctdb_context *ctdb,
        struct ctdb_client *client = state->client;
        struct ctdb_reply_control *r;
        size_t len;
+       int ret;
 
        /* construct a message to send to the client containing the data */
        len = offsetof(struct ctdb_reply_control, data) + data.dsize;
@@ -821,9 +952,10 @@ static void daemon_control_callback(struct ctdb_context *ctdb,
                memcpy(&r->data[r->datalen], errormsg, r->errorlen);
        }
 
-       daemon_queue_send(client, &r->hdr);
-
-       talloc_free(state);
+       ret = daemon_queue_send(client, &r->hdr);
+       if (ret != -1) {
+               talloc_free(state);
+       }
 }
 
 /*
@@ -979,6 +1111,11 @@ int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t pnn,
        struct ctdb_req_message *r;
        int len;
 
+       if (ctdb->methods == NULL) {
+               DEBUG(DEBUG_ERR,(__location__ " Failed to send message. Transport is DOWN\n"));
+               return -1;
+       }
+
        /* see if this is a message to ourselves */
        if (pnn == ctdb->pnn) {
                return ctdb_local_message(ctdb, srvid, data);
@@ -1000,3 +1137,144 @@ int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t pnn,
        return 0;
 }
 
+
+
+struct ctdb_client_notify_list {
+       struct ctdb_client_notify_list *next, *prev;
+       struct ctdb_context *ctdb;
+       uint64_t srvid;
+       TDB_DATA data;
+};
+
+
+static int ctdb_client_notify_destructor(struct ctdb_client_notify_list *nl)
+{
+       int ret;
+
+       DEBUG(DEBUG_ERR,("Sending client notify message for srvid:%llu\n", (unsigned long long)nl->srvid));
+
+       ret = ctdb_daemon_send_message(nl->ctdb, CTDB_BROADCAST_CONNECTED, (unsigned long long)nl->srvid, nl->data);
+       if (ret != 0) {
+               DEBUG(DEBUG_ERR,("Failed to send client notify message\n"));
+       }
+
+       return 0;
+}
+
+int32_t ctdb_control_register_notify(struct ctdb_context *ctdb, uint32_t client_id, TDB_DATA indata)
+{
+       struct ctdb_client_notify_register *notify = (struct ctdb_client_notify_register *)indata.dptr;
+        struct ctdb_client *client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client); 
+       struct ctdb_client_notify_list *nl;
+
+       DEBUG(DEBUG_INFO,("Register srvid %llu for client %d\n", (unsigned long long)notify->srvid, client_id));
+
+       if (indata.dsize < offsetof(struct ctdb_client_notify_register, notify_data)) {
+               DEBUG(DEBUG_ERR,(__location__ " Too little data in control : %d\n", (int)indata.dsize));
+               return -1;
+       }
+
+       if (indata.dsize != (notify->len + offsetof(struct ctdb_client_notify_register, notify_data))) {
+               DEBUG(DEBUG_ERR,(__location__ " Wrong amount of data in control. Got %d, expected %d\n", (int)indata.dsize, (int)(notify->len + offsetof(struct ctdb_client_notify_register, notify_data))));
+               return -1;
+       }
+
+
+        if (client == NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " Could not find client parent structure. You can not send this control to a remote node\n"));
+                return -1;
+        }
+
+       for(nl=client->notify; nl; nl=nl->next) {
+               if (nl->srvid == notify->srvid) {
+                       break;
+               }
+       }
+       if (nl != NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " Notification for srvid:%llu already exists for this client\n", (unsigned long long)notify->srvid));
+                return -1;
+        }
+
+       nl = talloc(client, struct ctdb_client_notify_list);
+       CTDB_NO_MEMORY(ctdb, nl);
+       nl->ctdb       = ctdb;
+       nl->srvid      = notify->srvid;
+       nl->data.dsize = notify->len;
+       nl->data.dptr  = talloc_size(nl, nl->data.dsize);
+       CTDB_NO_MEMORY(ctdb, nl->data.dptr);
+       memcpy(nl->data.dptr, notify->notify_data, nl->data.dsize);
+       
+       DLIST_ADD(client->notify, nl);
+       talloc_set_destructor(nl, ctdb_client_notify_destructor);
+
+       return 0;
+}
+
+int32_t ctdb_control_deregister_notify(struct ctdb_context *ctdb, uint32_t client_id, TDB_DATA indata)
+{
+       struct ctdb_client_notify_deregister *notify = (struct ctdb_client_notify_deregister *)indata.dptr;
+        struct ctdb_client *client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client); 
+       struct ctdb_client_notify_list *nl;
+
+       DEBUG(DEBUG_INFO,("Deregister srvid %llu for client %d\n", (unsigned long long)notify->srvid, client_id));
+
+        if (client == NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " Could not find client parent structure. You can not send this control to a remote node\n"));
+                return -1;
+        }
+
+       for(nl=client->notify; nl; nl=nl->next) {
+               if (nl->srvid == notify->srvid) {
+                       break;
+               }
+       }
+       if (nl == NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " No notification for srvid:%llu found for this client\n", (unsigned long long)notify->srvid));
+                return -1;
+        }
+
+       DLIST_REMOVE(client->notify, nl);
+       talloc_set_destructor(nl, NULL);
+       talloc_free(nl);
+
+       return 0;
+}
+
+struct ctdb_client *ctdb_find_client_by_pid(struct ctdb_context *ctdb, pid_t pid)
+{
+       struct ctdb_client_pid_list *client_pid;
+
+       for (client_pid = ctdb->client_pids; client_pid; client_pid=client_pid->next) {
+               if (client_pid->pid == pid) {
+                       return client_pid->client;
+               }
+       }
+       return NULL;
+}
+
+
+/* This control is used by samba when probing if a process (of a samba daemon)
+   exists on the node.
+   Samba does this when it needs/wants to check if a subrecord in one of the
+   databases is still valied, or if it is stale and can be removed.
+   If the node is in unhealthy or stopped state we just kill of the samba
+   process holding htis sub-record and return to the calling samba that
+   the process does not exist.
+   This allows us to forcefully recall subrecords registered by samba processes
+   on banned and stopped nodes.
+*/
+int32_t ctdb_control_process_exists(struct ctdb_context *ctdb, pid_t pid)
+{
+        struct ctdb_client *client;
+
+       if (ctdb->nodes[ctdb->pnn]->flags & (NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
+               client = ctdb_find_client_by_pid(ctdb, pid);
+               if (client != NULL) {
+                       DEBUG(DEBUG_NOTICE,(__location__ " Killing client with pid:%d on banned/stopped node\n", (int)pid));
+                       talloc_free(client);
+               }
+               return -1;
+       }
+
+       return kill(pid, 0);
+}