server: when we migrate off a record with data, set the MIGRATED_WITH_DATA flag
[samba.git] / ctdb / server / ctdb_call.c
1 /* 
2    ctdb_call protocol code
3
4    Copyright (C) Andrew Tridgell  2006
5
6    This program is free software; you can redistribute it and/or modify
7    it under the terms of the GNU General Public License as published by
8    the Free Software Foundation; either version 3 of the License, or
9    (at your option) any later version.
10    
11    This program is distributed in the hope that it will be useful,
12    but WITHOUT ANY WARRANTY; without even the implied warranty of
13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14    GNU General Public License for more details.
15    
16    You should have received a copy of the GNU General Public License
17    along with this program; if not, see <http://www.gnu.org/licenses/>.
18 */
19 /*
20   see http://wiki.samba.org/index.php/Samba_%26_Clustering for
21   protocol design and packet details
22 */
23 #include "includes.h"
24 #include "lib/tevent/tevent.h"
25 #include "lib/tdb/include/tdb.h"
26 #include "lib/util/dlinklist.h"
27 #include "system/network.h"
28 #include "system/filesys.h"
29 #include "../include/ctdb_private.h"
30
31 /*
32   find the ctdb_db from a db index
33  */
34  struct ctdb_db_context *find_ctdb_db(struct ctdb_context *ctdb, uint32_t id)
35 {
36         struct ctdb_db_context *ctdb_db;
37
38         for (ctdb_db=ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
39                 if (ctdb_db->db_id == id) {
40                         break;
41                 }
42         }
43         return ctdb_db;
44 }
45
46
47 /*
48   a varient of input packet that can be used in lock requeue
49 */
50 static void ctdb_call_input_pkt(void *p, struct ctdb_req_header *hdr)
51 {
52         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
53         ctdb_input_pkt(ctdb, hdr);
54 }
55
56
57 /*
58   send an error reply
59 */
60 static void ctdb_send_error(struct ctdb_context *ctdb, 
61                             struct ctdb_req_header *hdr, uint32_t status,
62                             const char *fmt, ...) PRINTF_ATTRIBUTE(4,5);
63 static void ctdb_send_error(struct ctdb_context *ctdb, 
64                             struct ctdb_req_header *hdr, uint32_t status,
65                             const char *fmt, ...)
66 {
67         va_list ap;
68         struct ctdb_reply_error *r;
69         char *msg;
70         int msglen, len;
71
72         if (ctdb->methods == NULL) {
73                 DEBUG(DEBUG_INFO,(__location__ " Failed to send error. Transport is DOWN\n"));
74                 return;
75         }
76
77         va_start(ap, fmt);
78         msg = talloc_vasprintf(ctdb, fmt, ap);
79         if (msg == NULL) {
80                 ctdb_fatal(ctdb, "Unable to allocate error in ctdb_send_error\n");
81         }
82         va_end(ap);
83
84         msglen = strlen(msg)+1;
85         len = offsetof(struct ctdb_reply_error, msg);
86         r = ctdb_transport_allocate(ctdb, msg, CTDB_REPLY_ERROR, len + msglen, 
87                                     struct ctdb_reply_error);
88         CTDB_NO_MEMORY_FATAL(ctdb, r);
89
90         r->hdr.destnode  = hdr->srcnode;
91         r->hdr.reqid     = hdr->reqid;
92         r->status        = status;
93         r->msglen        = msglen;
94         memcpy(&r->msg[0], msg, msglen);
95
96         ctdb_queue_packet(ctdb, &r->hdr);
97
98         talloc_free(msg);
99 }
100
101
102 /**
103  * send a redirect reply
104  *
105  * The logic behind this function is this:
106  *
107  * A client wants to grab a record and sends a CTDB_REQ_CALL packet
108  * to its local ctdb (ctdb_request_call). If the node is not itself
109  * the record's DMASTER, it first redirects the packet to  the
110  * record's LMASTER. The LMASTER then redirects the call packet to
111  * the current DMASTER. But there is a race: The record may have
112  * been migrated off the DMASTER while the redirected packet is
113  * on the wire (or in the local queue). So in case the record has
114  * migrated off the new destinaton of the call packet, instead of
115  * going back to the LMASTER to get the new DMASTER, we try to
116  * reduce rountrips by fist chasing the record a couple of times
117  * before giving up the direct chase and finally going back to the
118  * LMASTER (again). Note that this works because auf this: When
119  * a record is migrated off a node, then the new DMASTER is stored
120  * in the record's copy on the former DMASTER.
121  *
122  * The maxiumum number of attempts for direct chase to make before
123  * going back to the LMASTER is configurable by the tunable
124  * "MaxRedirectCount".
125  */
126 static void ctdb_call_send_redirect(struct ctdb_context *ctdb, 
127                                     TDB_DATA key,
128                                     struct ctdb_req_call *c, 
129                                     struct ctdb_ltdb_header *header)
130 {
131         
132         uint32_t lmaster = ctdb_lmaster(ctdb, &key);
133         if (ctdb->pnn == lmaster) {
134                 c->hdr.destnode = header->dmaster;
135         } else if ((c->hopcount % ctdb->tunable.max_redirect_count) == 0) {
136                 c->hdr.destnode = lmaster;
137         } else {
138                 c->hdr.destnode = header->dmaster;
139         }
140         c->hopcount++;
141         ctdb_queue_packet(ctdb, &c->hdr);
142 }
143
144
145 /*
146   send a dmaster reply
147
148   caller must have the chainlock before calling this routine. Caller must be
149   the lmaster
150 */
151 static void ctdb_send_dmaster_reply(struct ctdb_db_context *ctdb_db,
152                                     struct ctdb_ltdb_header *header,
153                                     TDB_DATA key, TDB_DATA data,
154                                     uint32_t new_dmaster,
155                                     uint32_t reqid)
156 {
157         struct ctdb_context *ctdb = ctdb_db->ctdb;
158         struct ctdb_reply_dmaster *r;
159         int ret, len;
160         TALLOC_CTX *tmp_ctx;
161
162         if (ctdb->pnn != ctdb_lmaster(ctdb, &key)) {
163                 DEBUG(DEBUG_ALERT,(__location__ " Caller is not lmaster!\n"));
164                 return;
165         }
166
167         header->dmaster = new_dmaster;
168         ret = ctdb_ltdb_store(ctdb_db, key, header, data);
169         if (ret != 0) {
170                 ctdb_fatal(ctdb, "ctdb_send_dmaster_reply unable to update dmaster");
171                 return;
172         }
173
174         if (ctdb->methods == NULL) {
175                 ctdb_fatal(ctdb, "ctdb_send_dmaster_reply cant update dmaster since transport is down");
176                 return;
177         }
178
179         /* put the packet on a temporary context, allowing us to safely free
180            it below even if ctdb_reply_dmaster() has freed it already */
181         tmp_ctx = talloc_new(ctdb);
182
183         /* send the CTDB_REPLY_DMASTER */
184         len = offsetof(struct ctdb_reply_dmaster, data) + key.dsize + data.dsize;
185         r = ctdb_transport_allocate(ctdb, tmp_ctx, CTDB_REPLY_DMASTER, len,
186                                     struct ctdb_reply_dmaster);
187         CTDB_NO_MEMORY_FATAL(ctdb, r);
188
189         r->hdr.destnode  = new_dmaster;
190         r->hdr.reqid     = reqid;
191         r->rsn           = header->rsn;
192         r->keylen        = key.dsize;
193         r->datalen       = data.dsize;
194         r->db_id         = ctdb_db->db_id;
195         memcpy(&r->data[0], key.dptr, key.dsize);
196         memcpy(&r->data[key.dsize], data.dptr, data.dsize);
197
198         ctdb_queue_packet(ctdb, &r->hdr);
199
200         talloc_free(tmp_ctx);
201 }
202
203 /*
204   send a dmaster request (give another node the dmaster for a record)
205
206   This is always sent to the lmaster, which ensures that the lmaster
207   always knows who the dmaster is. The lmaster will then send a
208   CTDB_REPLY_DMASTER to the new dmaster
209 */
210 static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db, 
211                                    struct ctdb_req_call *c, 
212                                    struct ctdb_ltdb_header *header,
213                                    TDB_DATA *key, TDB_DATA *data)
214 {
215         struct ctdb_req_dmaster *r;
216         struct ctdb_context *ctdb = ctdb_db->ctdb;
217         int len;
218         uint32_t lmaster = ctdb_lmaster(ctdb, key);
219
220         if (ctdb->methods == NULL) {
221                 ctdb_fatal(ctdb, "Failed ctdb_call_send_dmaster since transport is down");
222                 return;
223         }
224
225         if (data->dsize != 0) {
226                 header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
227         }
228
229         if (lmaster == ctdb->pnn) {
230                 ctdb_send_dmaster_reply(ctdb_db, header, *key, *data, 
231                                         c->hdr.srcnode, c->hdr.reqid);
232                 return;
233         }
234         
235         len = offsetof(struct ctdb_req_dmaster, data) + key->dsize + data->dsize;
236         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REQ_DMASTER, len, 
237                                     struct ctdb_req_dmaster);
238         CTDB_NO_MEMORY_FATAL(ctdb, r);
239         r->hdr.destnode  = lmaster;
240         r->hdr.reqid     = c->hdr.reqid;
241         r->db_id         = c->db_id;
242         r->rsn           = header->rsn;
243         r->dmaster       = c->hdr.srcnode;
244         r->keylen        = key->dsize;
245         r->datalen       = data->dsize;
246         memcpy(&r->data[0], key->dptr, key->dsize);
247         memcpy(&r->data[key->dsize], data->dptr, data->dsize);
248
249         header->dmaster = c->hdr.srcnode;
250         if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
251                 ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
252         }
253         
254         ctdb_queue_packet(ctdb, &r->hdr);
255
256         talloc_free(r);
257 }
258
259 /*
260   called when a CTDB_REPLY_DMASTER packet comes in, or when the lmaster
261   gets a CTDB_REQUEST_DMASTER for itself. We become the dmaster.
262
263   must be called with the chainlock held. This function releases the chainlock
264 */
265 static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db, 
266                                 struct ctdb_req_header *hdr,
267                                 TDB_DATA key, TDB_DATA data,
268                                 uint64_t rsn)
269 {
270         struct ctdb_call_state *state;
271         struct ctdb_context *ctdb = ctdb_db->ctdb;
272         struct ctdb_ltdb_header header;
273         int ret;
274
275         DEBUG(DEBUG_DEBUG,("pnn %u dmaster response %08x\n", ctdb->pnn, ctdb_hash(&key)));
276
277         ZERO_STRUCT(header);
278         header.rsn = rsn + 1;
279         header.dmaster = ctdb->pnn;
280
281         if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
282                 ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
283
284                 ret = ctdb_ltdb_unlock(ctdb_db, key);
285                 if (ret != 0) {
286                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
287                 }
288                 return;
289         }
290
291         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
292
293         if (state == NULL) {
294                 DEBUG(DEBUG_ERR,("pnn %u Invalid reqid %u in ctdb_become_dmaster from node %u\n",
295                          ctdb->pnn, hdr->reqid, hdr->srcnode));
296
297                 ret = ctdb_ltdb_unlock(ctdb_db, key);
298                 if (ret != 0) {
299                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
300                 }
301                 return;
302         }
303
304         if (key.dsize != state->call->key.dsize || memcmp(key.dptr, state->call->key.dptr, key.dsize)) {
305                 DEBUG(DEBUG_ERR, ("Got bogus DMASTER packet reqid:%u from node %u. Key does not match key held in matching idr.\n", hdr->reqid, hdr->srcnode));
306
307                 ret = ctdb_ltdb_unlock(ctdb_db, key);
308                 if (ret != 0) {
309                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
310                 }
311                 return;
312         }
313
314         if (hdr->reqid != state->reqid) {
315                 /* we found a record  but it was the wrong one */
316                 DEBUG(DEBUG_ERR, ("Dropped orphan in ctdb_become_dmaster with reqid:%u\n from node %u", hdr->reqid, hdr->srcnode));
317
318                 ret = ctdb_ltdb_unlock(ctdb_db, key);
319                 if (ret != 0) {
320                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
321                 }
322                 return;
323         }
324
325         ctdb_call_local(ctdb_db, state->call, &header, state, &data);
326
327         ret = ctdb_ltdb_unlock(ctdb_db, state->call->key);
328         if (ret != 0) {
329                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
330         }
331
332         state->state = CTDB_CALL_DONE;
333         if (state->async.fn) {
334                 state->async.fn(state);
335         }
336 }
337
338
339
340 /*
341   called when a CTDB_REQ_DMASTER packet comes in
342
343   this comes into the lmaster for a record when the current dmaster
344   wants to give up the dmaster role and give it to someone else
345 */
346 void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
347 {
348         struct ctdb_req_dmaster *c = (struct ctdb_req_dmaster *)hdr;
349         TDB_DATA key, data, data2;
350         struct ctdb_ltdb_header header;
351         struct ctdb_db_context *ctdb_db;
352         int ret;
353
354         key.dptr = c->data;
355         key.dsize = c->keylen;
356         data.dptr = c->data + c->keylen;
357         data.dsize = c->datalen;
358
359         ctdb_db = find_ctdb_db(ctdb, c->db_id);
360         if (!ctdb_db) {
361                 ctdb_send_error(ctdb, hdr, -1,
362                                 "Unknown database in request. db_id==0x%08x",
363                                 c->db_id);
364                 return;
365         }
366         
367         /* fetch the current record */
368         ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, key, &header, hdr, &data2,
369                                            ctdb_call_input_pkt, ctdb, False);
370         if (ret == -1) {
371                 ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
372                 return;
373         }
374         if (ret == -2) {
375                 DEBUG(DEBUG_INFO,(__location__ " deferring ctdb_request_dmaster\n"));
376                 return;
377         }
378
379         if (ctdb_lmaster(ctdb, &key) != ctdb->pnn) {
380                 DEBUG(DEBUG_ALERT,("pnn %u dmaster request to non-lmaster lmaster=%u gen=%u curgen=%u\n",
381                          ctdb->pnn, ctdb_lmaster(ctdb, &key), 
382                          hdr->generation, ctdb->vnn_map->generation));
383                 ctdb_fatal(ctdb, "ctdb_req_dmaster to non-lmaster");
384         }
385
386         DEBUG(DEBUG_DEBUG,("pnn %u dmaster request on %08x for %u from %u\n", 
387                  ctdb->pnn, ctdb_hash(&key), c->dmaster, c->hdr.srcnode));
388
389         /* its a protocol error if the sending node is not the current dmaster */
390         if (header.dmaster != hdr->srcnode) {
391                 DEBUG(DEBUG_ALERT,("pnn %u dmaster request for new-dmaster %u from non-master %u real-dmaster=%u key %08x dbid 0x%08x gen=%u curgen=%u c->rsn=%llu header.rsn=%llu reqid=%u keyval=0x%08x\n",
392                          ctdb->pnn, c->dmaster, hdr->srcnode, header.dmaster, ctdb_hash(&key),
393                          ctdb_db->db_id, hdr->generation, ctdb->vnn_map->generation,
394                          (unsigned long long)c->rsn, (unsigned long long)header.rsn, c->hdr.reqid,
395                          (key.dsize >= 4)?(*(uint32_t *)key.dptr):0));
396                 if (header.rsn != 0 || header.dmaster != ctdb->pnn) {
397                         DEBUG(DEBUG_ERR,("ctdb_req_dmaster from non-master. Force a recovery.\n"));
398
399                         ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
400                         return;
401                 }
402         }
403
404         if (header.rsn > c->rsn) {
405                 DEBUG(DEBUG_ALERT,("pnn %u dmaster request with older RSN new-dmaster %u from %u real-dmaster=%u key %08x dbid 0x%08x gen=%u curgen=%u c->rsn=%llu header.rsn=%llu reqid=%u\n",
406                          ctdb->pnn, c->dmaster, hdr->srcnode, header.dmaster, ctdb_hash(&key),
407                          ctdb_db->db_id, hdr->generation, ctdb->vnn_map->generation,
408                          (unsigned long long)c->rsn, (unsigned long long)header.rsn, c->hdr.reqid));
409         }
410
411         /* use the rsn from the sending node */
412         header.rsn = c->rsn;
413
414         /* check if the new dmaster is the lmaster, in which case we
415            skip the dmaster reply */
416         if (c->dmaster == ctdb->pnn) {
417                 ctdb_become_dmaster(ctdb_db, hdr, key, data, c->rsn);
418         } else {
419                 ctdb_send_dmaster_reply(ctdb_db, &header, key, data, c->dmaster, hdr->reqid);
420
421                 ret = ctdb_ltdb_unlock(ctdb_db, key);
422                 if (ret != 0) {
423                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
424                 }
425         }
426 }
427
428
429 /*
430   called when a CTDB_REQ_CALL packet comes in
431 */
432 void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
433 {
434         struct ctdb_req_call *c = (struct ctdb_req_call *)hdr;
435         TDB_DATA data;
436         struct ctdb_reply_call *r;
437         int ret, len;
438         struct ctdb_ltdb_header header;
439         struct ctdb_call *call;
440         struct ctdb_db_context *ctdb_db;
441
442         if (ctdb->methods == NULL) {
443                 DEBUG(DEBUG_INFO,(__location__ " Failed ctdb_request_call. Transport is DOWN\n"));
444                 return;
445         }
446
447
448         ctdb_db = find_ctdb_db(ctdb, c->db_id);
449         if (!ctdb_db) {
450                 ctdb_send_error(ctdb, hdr, -1,
451                                 "Unknown database in request. db_id==0x%08x",
452                                 c->db_id);
453                 return;
454         }
455
456         call = talloc(hdr, struct ctdb_call);
457         CTDB_NO_MEMORY_FATAL(ctdb, call);
458
459         call->call_id  = c->callid;
460         call->key.dptr = c->data;
461         call->key.dsize = c->keylen;
462         call->call_data.dptr = c->data + c->keylen;
463         call->call_data.dsize = c->calldatalen;
464
465         /* determine if we are the dmaster for this key. This also
466            fetches the record data (if any), thus avoiding a 2nd fetch of the data 
467            if the call will be answered locally */
468
469         ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, call->key, &header, hdr, &data,
470                                            ctdb_call_input_pkt, ctdb, False);
471         if (ret == -1) {
472                 ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call");
473                 return;
474         }
475         if (ret == -2) {
476                 DEBUG(DEBUG_INFO,(__location__ " deferred ctdb_request_call\n"));
477                 return;
478         }
479
480         /* if we are not the dmaster, then send a redirect to the
481            requesting node */
482         if (header.dmaster != ctdb->pnn) {
483                 talloc_free(data.dptr);
484                 ctdb_call_send_redirect(ctdb, call->key, c, &header);
485
486                 ret = ctdb_ltdb_unlock(ctdb_db, call->key);
487                 if (ret != 0) {
488                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
489                 }
490                 return;
491         }
492
493         CTDB_UPDATE_STAT(ctdb, max_hop_count, c->hopcount);
494
495         /* Try if possible to migrate the record off to the caller node.
496          * From the clients perspective a fetch of the data is just as 
497          * expensive as a migration.
498          */
499         if (c->hdr.srcnode != ctdb->pnn) {
500                 if (ctdb_db->transaction_active) {
501                         DEBUG(DEBUG_INFO, (__location__ " refusing migration"
502                               " of key %s while transaction is active\n",
503                               (char *)call->key.dptr));
504                 } else {
505                         DEBUG(DEBUG_DEBUG,("pnn %u starting migration of %08x to %u\n",
506                                  ctdb->pnn, ctdb_hash(&(call->key)), c->hdr.srcnode));
507                         ctdb_call_send_dmaster(ctdb_db, c, &header, &(call->key), &data);
508                         talloc_free(data.dptr);
509
510                         ret = ctdb_ltdb_unlock(ctdb_db, call->key);
511                         if (ret != 0) {
512                                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
513                         }
514                         return;
515                 }
516         }
517
518         ctdb_call_local(ctdb_db, call, &header, hdr, &data);
519
520         ret = ctdb_ltdb_unlock(ctdb_db, call->key);
521         if (ret != 0) {
522                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
523         }
524
525         len = offsetof(struct ctdb_reply_call, data) + call->reply_data.dsize;
526         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REPLY_CALL, len, 
527                                     struct ctdb_reply_call);
528         CTDB_NO_MEMORY_FATAL(ctdb, r);
529         r->hdr.destnode  = hdr->srcnode;
530         r->hdr.reqid     = hdr->reqid;
531         r->status        = call->status;
532         r->datalen       = call->reply_data.dsize;
533         if (call->reply_data.dsize) {
534                 memcpy(&r->data[0], call->reply_data.dptr, call->reply_data.dsize);
535         }
536
537         ctdb_queue_packet(ctdb, &r->hdr);
538
539         talloc_free(r);
540 }
541
542 /*
543   called when a CTDB_REPLY_CALL packet comes in
544
545   This packet comes in response to a CTDB_REQ_CALL request packet. It
546   contains any reply data from the call
547 */
548 void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
549 {
550         struct ctdb_reply_call *c = (struct ctdb_reply_call *)hdr;
551         struct ctdb_call_state *state;
552
553         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
554         if (state == NULL) {
555                 DEBUG(DEBUG_ERR, (__location__ " reqid %u not found\n", hdr->reqid));
556                 return;
557         }
558
559         if (hdr->reqid != state->reqid) {
560                 /* we found a record  but it was the wrong one */
561                 DEBUG(DEBUG_ERR, ("Dropped orphaned call reply with reqid:%u\n",hdr->reqid));
562                 return;
563         }
564
565         state->call->reply_data.dptr = c->data;
566         state->call->reply_data.dsize = c->datalen;
567         state->call->status = c->status;
568
569         talloc_steal(state, c);
570
571         state->state = CTDB_CALL_DONE;
572         if (state->async.fn) {
573                 state->async.fn(state);
574         }
575 }
576
577
578 /*
579   called when a CTDB_REPLY_DMASTER packet comes in
580
581   This packet comes in from the lmaster response to a CTDB_REQ_CALL
582   request packet. It means that the current dmaster wants to give us
583   the dmaster role
584 */
585 void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
586 {
587         struct ctdb_reply_dmaster *c = (struct ctdb_reply_dmaster *)hdr;
588         struct ctdb_db_context *ctdb_db;
589         TDB_DATA key, data;
590         int ret;
591
592         ctdb_db = find_ctdb_db(ctdb, c->db_id);
593         if (ctdb_db == NULL) {
594                 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in ctdb_reply_dmaster\n", c->db_id));
595                 return;
596         }
597         
598         key.dptr = c->data;
599         key.dsize = c->keylen;
600         data.dptr = &c->data[key.dsize];
601         data.dsize = c->datalen;
602
603         ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr,
604                                      ctdb_call_input_pkt, ctdb, False);
605         if (ret == -2) {
606                 return;
607         }
608         if (ret != 0) {
609                 DEBUG(DEBUG_ERR,(__location__ " Failed to get lock in ctdb_reply_dmaster\n"));
610                 return;
611         }
612
613         ctdb_become_dmaster(ctdb_db, hdr, key, data, c->rsn);
614 }
615
616
617 /*
618   called when a CTDB_REPLY_ERROR packet comes in
619 */
620 void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
621 {
622         struct ctdb_reply_error *c = (struct ctdb_reply_error *)hdr;
623         struct ctdb_call_state *state;
624
625         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
626         if (state == NULL) {
627                 DEBUG(DEBUG_ERR,("pnn %u Invalid reqid %u in ctdb_reply_error\n",
628                          ctdb->pnn, hdr->reqid));
629                 return;
630         }
631
632         if (hdr->reqid != state->reqid) {
633                 /* we found a record  but it was the wrong one */
634                 DEBUG(DEBUG_ERR, ("Dropped orphaned error reply with reqid:%u\n",hdr->reqid));
635                 return;
636         }
637
638         talloc_steal(state, c);
639
640         state->state  = CTDB_CALL_ERROR;
641         state->errmsg = (char *)c->msg;
642         if (state->async.fn) {
643                 state->async.fn(state);
644         }
645 }
646
647
648 /*
649   destroy a ctdb_call
650 */
651 static int ctdb_call_destructor(struct ctdb_call_state *state)
652 {
653         DLIST_REMOVE(state->ctdb_db->ctdb->pending_calls, state);
654         ctdb_reqid_remove(state->ctdb_db->ctdb, state->reqid);
655         return 0;
656 }
657
658
659 /*
660   called when a ctdb_call needs to be resent after a reconfigure event
661 */
662 static void ctdb_call_resend(struct ctdb_call_state *state)
663 {
664         struct ctdb_context *ctdb = state->ctdb_db->ctdb;
665
666         state->generation = ctdb->vnn_map->generation;
667
668         /* use a new reqid, in case the old reply does eventually come in */
669         ctdb_reqid_remove(ctdb, state->reqid);
670         state->reqid = ctdb_reqid_new(ctdb, state);
671         state->c->hdr.reqid = state->reqid;
672
673         /* update the generation count for this request, so its valid with the new vnn_map */
674         state->c->hdr.generation = state->generation;
675
676         /* send the packet to ourselves, it will be redirected appropriately */
677         state->c->hdr.destnode = ctdb->pnn;
678
679         ctdb_queue_packet(ctdb, &state->c->hdr);
680         DEBUG(DEBUG_NOTICE,("resent ctdb_call\n"));
681 }
682
683 /*
684   resend all pending calls on recovery
685  */
686 void ctdb_call_resend_all(struct ctdb_context *ctdb)
687 {
688         struct ctdb_call_state *state, *next;
689         for (state=ctdb->pending_calls;state;state=next) {
690                 next = state->next;
691                 ctdb_call_resend(state);
692         }
693 }
694
695 /*
696   this allows the caller to setup a async.fn 
697 */
698 static void call_local_trigger(struct event_context *ev, struct timed_event *te, 
699                        struct timeval t, void *private_data)
700 {
701         struct ctdb_call_state *state = talloc_get_type(private_data, struct ctdb_call_state);
702         if (state->async.fn) {
703                 state->async.fn(state);
704         }
705 }       
706
707
708 /*
709   construct an event driven local ctdb_call
710
711   this is used so that locally processed ctdb_call requests are processed
712   in an event driven manner
713 */
714 struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, 
715                                              struct ctdb_call *call,
716                                              struct ctdb_ltdb_header *header,
717                                              TDB_DATA *data)
718 {
719         struct ctdb_call_state *state;
720         struct ctdb_context *ctdb = ctdb_db->ctdb;
721         int ret;
722
723         state = talloc_zero(ctdb_db, struct ctdb_call_state);
724         CTDB_NO_MEMORY_NULL(ctdb, state);
725
726         talloc_steal(state, data->dptr);
727
728         state->state = CTDB_CALL_DONE;
729         state->call  = talloc(state, struct ctdb_call);
730         CTDB_NO_MEMORY_NULL(ctdb, state->call);
731         *(state->call) = *call;
732         state->ctdb_db = ctdb_db;
733
734         ret = ctdb_call_local(ctdb_db, state->call, header, state, data);
735
736         event_add_timed(ctdb->ev, state, timeval_zero(), call_local_trigger, state);
737
738         return state;
739 }
740
741
742 /*
743   make a remote ctdb call - async send. Called in daemon context.
744
745   This constructs a ctdb_call request and queues it for processing. 
746   This call never blocks.
747 */
748 struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctdb_db, 
749                                                      struct ctdb_call *call, 
750                                                      struct ctdb_ltdb_header *header)
751 {
752         uint32_t len;
753         struct ctdb_call_state *state;
754         struct ctdb_context *ctdb = ctdb_db->ctdb;
755
756         if (ctdb->methods == NULL) {
757                 DEBUG(DEBUG_INFO,(__location__ " Failed send packet. Transport is down\n"));
758                 return NULL;
759         }
760
761         state = talloc_zero(ctdb_db, struct ctdb_call_state);
762         CTDB_NO_MEMORY_NULL(ctdb, state);
763         state->call = talloc(state, struct ctdb_call);
764         CTDB_NO_MEMORY_NULL(ctdb, state->call);
765
766         state->reqid = ctdb_reqid_new(ctdb, state);
767         state->ctdb_db = ctdb_db;
768         talloc_set_destructor(state, ctdb_call_destructor);
769
770         len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
771         state->c = ctdb_transport_allocate(ctdb, state, CTDB_REQ_CALL, len, 
772                                            struct ctdb_req_call);
773         CTDB_NO_MEMORY_NULL(ctdb, state->c);
774         state->c->hdr.destnode  = header->dmaster;
775
776         /* this limits us to 16k outstanding messages - not unreasonable */
777         state->c->hdr.reqid     = state->reqid;
778         state->c->flags         = call->flags;
779         state->c->db_id         = ctdb_db->db_id;
780         state->c->callid        = call->call_id;
781         state->c->hopcount      = 0;
782         state->c->keylen        = call->key.dsize;
783         state->c->calldatalen   = call->call_data.dsize;
784         memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
785         memcpy(&state->c->data[call->key.dsize], 
786                call->call_data.dptr, call->call_data.dsize);
787         *(state->call)              = *call;
788         state->call->call_data.dptr = &state->c->data[call->key.dsize];
789         state->call->key.dptr       = &state->c->data[0];
790
791         state->state  = CTDB_CALL_WAIT;
792         state->generation = ctdb->vnn_map->generation;
793
794         DLIST_ADD(ctdb->pending_calls, state);
795
796         ctdb_queue_packet(ctdb, &state->c->hdr);
797
798         return state;
799 }
800
801 /*
802   make a remote ctdb call - async recv - called in daemon context
803
804   This is called when the program wants to wait for a ctdb_call to complete and get the 
805   results. This call will block unless the call has already completed.
806 */
807 int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
808 {
809         while (state->state < CTDB_CALL_DONE) {
810                 event_loop_once(state->ctdb_db->ctdb->ev);
811         }
812         if (state->state != CTDB_CALL_DONE) {
813                 ctdb_set_error(state->ctdb_db->ctdb, "%s", state->errmsg);
814                 talloc_free(state);
815                 return -1;
816         }
817
818         if (state->call->reply_data.dsize) {
819                 call->reply_data.dptr = talloc_memdup(call,
820                                                       state->call->reply_data.dptr,
821                                                       state->call->reply_data.dsize);
822                 call->reply_data.dsize = state->call->reply_data.dsize;
823         } else {
824                 call->reply_data.dptr = NULL;
825                 call->reply_data.dsize = 0;
826         }
827         call->status = state->call->status;
828         talloc_free(state);
829         return 0;
830 }
831
832
833 /* 
834    send a keepalive packet to the other node
835 */
836 void ctdb_send_keepalive(struct ctdb_context *ctdb, uint32_t destnode)
837 {
838         struct ctdb_req_keepalive *r;
839         
840         if (ctdb->methods == NULL) {
841                 DEBUG(DEBUG_INFO,(__location__ " Failed to send keepalive. Transport is DOWN\n"));
842                 return;
843         }
844
845         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REQ_KEEPALIVE,
846                                     sizeof(struct ctdb_req_keepalive), 
847                                     struct ctdb_req_keepalive);
848         CTDB_NO_MEMORY_FATAL(ctdb, r);
849         r->hdr.destnode  = destnode;
850         r->hdr.reqid     = 0;
851         
852         CTDB_INCREMENT_STAT(ctdb, keepalive_packets_sent);
853
854         ctdb_queue_packet(ctdb, &r->hdr);
855
856         talloc_free(r);
857 }