server: add a comment explaining the call redirect logic in ctdb_call_send_redirect().
[sahlberg/ctdb.git] / 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 (lmaster == ctdb->pnn) {
226                 ctdb_send_dmaster_reply(ctdb_db, header, *key, *data, 
227                                         c->hdr.srcnode, c->hdr.reqid);
228                 return;
229         }
230         
231         len = offsetof(struct ctdb_req_dmaster, data) + key->dsize + data->dsize;
232         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REQ_DMASTER, len, 
233                                     struct ctdb_req_dmaster);
234         CTDB_NO_MEMORY_FATAL(ctdb, r);
235         r->hdr.destnode  = lmaster;
236         r->hdr.reqid     = c->hdr.reqid;
237         r->db_id         = c->db_id;
238         r->rsn           = header->rsn;
239         r->dmaster       = c->hdr.srcnode;
240         r->keylen        = key->dsize;
241         r->datalen       = data->dsize;
242         memcpy(&r->data[0], key->dptr, key->dsize);
243         memcpy(&r->data[key->dsize], data->dptr, data->dsize);
244
245         header->dmaster = c->hdr.srcnode;
246         if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
247                 ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
248         }
249         
250         ctdb_queue_packet(ctdb, &r->hdr);
251
252         talloc_free(r);
253 }
254
255 /*
256   called when a CTDB_REPLY_DMASTER packet comes in, or when the lmaster
257   gets a CTDB_REQUEST_DMASTER for itself. We become the dmaster.
258
259   must be called with the chainlock held. This function releases the chainlock
260 */
261 static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db, 
262                                 struct ctdb_req_header *hdr,
263                                 TDB_DATA key, TDB_DATA data,
264                                 uint64_t rsn)
265 {
266         struct ctdb_call_state *state;
267         struct ctdb_context *ctdb = ctdb_db->ctdb;
268         struct ctdb_ltdb_header header;
269         int ret;
270
271         DEBUG(DEBUG_DEBUG,("pnn %u dmaster response %08x\n", ctdb->pnn, ctdb_hash(&key)));
272
273         ZERO_STRUCT(header);
274         header.rsn = rsn + 1;
275         header.dmaster = ctdb->pnn;
276
277         if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
278                 ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
279
280                 ret = ctdb_ltdb_unlock(ctdb_db, key);
281                 if (ret != 0) {
282                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
283                 }
284                 return;
285         }
286
287         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
288
289         if (state == NULL) {
290                 DEBUG(DEBUG_ERR,("pnn %u Invalid reqid %u in ctdb_become_dmaster from node %u\n",
291                          ctdb->pnn, hdr->reqid, hdr->srcnode));
292
293                 ret = ctdb_ltdb_unlock(ctdb_db, key);
294                 if (ret != 0) {
295                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
296                 }
297                 return;
298         }
299
300         if (key.dsize != state->call->key.dsize || memcmp(key.dptr, state->call->key.dptr, key.dsize)) {
301                 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));
302
303                 ret = ctdb_ltdb_unlock(ctdb_db, key);
304                 if (ret != 0) {
305                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
306                 }
307                 return;
308         }
309
310         if (hdr->reqid != state->reqid) {
311                 /* we found a record  but it was the wrong one */
312                 DEBUG(DEBUG_ERR, ("Dropped orphan in ctdb_become_dmaster with reqid:%u\n from node %u", hdr->reqid, hdr->srcnode));
313
314                 ret = ctdb_ltdb_unlock(ctdb_db, key);
315                 if (ret != 0) {
316                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
317                 }
318                 return;
319         }
320
321         ctdb_call_local(ctdb_db, state->call, &header, state, &data);
322
323         ret = ctdb_ltdb_unlock(ctdb_db, state->call->key);
324         if (ret != 0) {
325                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
326         }
327
328         state->state = CTDB_CALL_DONE;
329         if (state->async.fn) {
330                 state->async.fn(state);
331         }
332 }
333
334
335
336 /*
337   called when a CTDB_REQ_DMASTER packet comes in
338
339   this comes into the lmaster for a record when the current dmaster
340   wants to give up the dmaster role and give it to someone else
341 */
342 void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
343 {
344         struct ctdb_req_dmaster *c = (struct ctdb_req_dmaster *)hdr;
345         TDB_DATA key, data, data2;
346         struct ctdb_ltdb_header header;
347         struct ctdb_db_context *ctdb_db;
348         int ret;
349
350         key.dptr = c->data;
351         key.dsize = c->keylen;
352         data.dptr = c->data + c->keylen;
353         data.dsize = c->datalen;
354
355         ctdb_db = find_ctdb_db(ctdb, c->db_id);
356         if (!ctdb_db) {
357                 ctdb_send_error(ctdb, hdr, -1,
358                                 "Unknown database in request. db_id==0x%08x",
359                                 c->db_id);
360                 return;
361         }
362         
363         /* fetch the current record */
364         ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, key, &header, hdr, &data2,
365                                            ctdb_call_input_pkt, ctdb, False);
366         if (ret == -1) {
367                 ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
368                 return;
369         }
370         if (ret == -2) {
371                 DEBUG(DEBUG_INFO,(__location__ " deferring ctdb_request_dmaster\n"));
372                 return;
373         }
374
375         if (ctdb_lmaster(ctdb, &key) != ctdb->pnn) {
376                 DEBUG(DEBUG_ALERT,("pnn %u dmaster request to non-lmaster lmaster=%u gen=%u curgen=%u\n",
377                          ctdb->pnn, ctdb_lmaster(ctdb, &key), 
378                          hdr->generation, ctdb->vnn_map->generation));
379                 ctdb_fatal(ctdb, "ctdb_req_dmaster to non-lmaster");
380         }
381
382         DEBUG(DEBUG_DEBUG,("pnn %u dmaster request on %08x for %u from %u\n", 
383                  ctdb->pnn, ctdb_hash(&key), c->dmaster, c->hdr.srcnode));
384
385         /* its a protocol error if the sending node is not the current dmaster */
386         if (header.dmaster != hdr->srcnode) {
387                 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",
388                          ctdb->pnn, c->dmaster, hdr->srcnode, header.dmaster, ctdb_hash(&key),
389                          ctdb_db->db_id, hdr->generation, ctdb->vnn_map->generation,
390                          (unsigned long long)c->rsn, (unsigned long long)header.rsn, c->hdr.reqid,
391                          (key.dsize >= 4)?(*(uint32_t *)key.dptr):0));
392                 if (header.rsn != 0 || header.dmaster != ctdb->pnn) {
393                         DEBUG(DEBUG_ERR,("ctdb_req_dmaster from non-master. Force a recovery.\n"));
394
395                         ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
396                         return;
397                 }
398         }
399
400         if (header.rsn > c->rsn) {
401                 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",
402                          ctdb->pnn, c->dmaster, hdr->srcnode, header.dmaster, ctdb_hash(&key),
403                          ctdb_db->db_id, hdr->generation, ctdb->vnn_map->generation,
404                          (unsigned long long)c->rsn, (unsigned long long)header.rsn, c->hdr.reqid));
405         }
406
407         /* use the rsn from the sending node */
408         header.rsn = c->rsn;
409
410         /* check if the new dmaster is the lmaster, in which case we
411            skip the dmaster reply */
412         if (c->dmaster == ctdb->pnn) {
413                 ctdb_become_dmaster(ctdb_db, hdr, key, data, c->rsn);
414         } else {
415                 ctdb_send_dmaster_reply(ctdb_db, &header, key, data, c->dmaster, hdr->reqid);
416
417                 ret = ctdb_ltdb_unlock(ctdb_db, key);
418                 if (ret != 0) {
419                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
420                 }
421         }
422 }
423
424
425 /*
426   called when a CTDB_REQ_CALL packet comes in
427 */
428 void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
429 {
430         struct ctdb_req_call *c = (struct ctdb_req_call *)hdr;
431         TDB_DATA data;
432         struct ctdb_reply_call *r;
433         int ret, len;
434         struct ctdb_ltdb_header header;
435         struct ctdb_call *call;
436         struct ctdb_db_context *ctdb_db;
437
438         if (ctdb->methods == NULL) {
439                 DEBUG(DEBUG_INFO,(__location__ " Failed ctdb_request_call. Transport is DOWN\n"));
440                 return;
441         }
442
443
444         ctdb_db = find_ctdb_db(ctdb, c->db_id);
445         if (!ctdb_db) {
446                 ctdb_send_error(ctdb, hdr, -1,
447                                 "Unknown database in request. db_id==0x%08x",
448                                 c->db_id);
449                 return;
450         }
451
452         call = talloc(hdr, struct ctdb_call);
453         CTDB_NO_MEMORY_FATAL(ctdb, call);
454
455         call->call_id  = c->callid;
456         call->key.dptr = c->data;
457         call->key.dsize = c->keylen;
458         call->call_data.dptr = c->data + c->keylen;
459         call->call_data.dsize = c->calldatalen;
460
461         /* determine if we are the dmaster for this key. This also
462            fetches the record data (if any), thus avoiding a 2nd fetch of the data 
463            if the call will be answered locally */
464
465         ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, call->key, &header, hdr, &data,
466                                            ctdb_call_input_pkt, ctdb, False);
467         if (ret == -1) {
468                 ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call");
469                 return;
470         }
471         if (ret == -2) {
472                 DEBUG(DEBUG_INFO,(__location__ " deferred ctdb_request_call\n"));
473                 return;
474         }
475
476         /* if we are not the dmaster, then send a redirect to the
477            requesting node */
478         if (header.dmaster != ctdb->pnn) {
479                 talloc_free(data.dptr);
480                 ctdb_call_send_redirect(ctdb, call->key, c, &header);
481
482                 ret = ctdb_ltdb_unlock(ctdb_db, call->key);
483                 if (ret != 0) {
484                         DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
485                 }
486                 return;
487         }
488
489         CTDB_UPDATE_STAT(ctdb, max_hop_count, c->hopcount);
490
491         /* Try if possible to migrate the record off to the caller node.
492          * From the clients perspective a fetch of the data is just as 
493          * expensive as a migration.
494          */
495         if (c->hdr.srcnode != ctdb->pnn) {
496                 if (ctdb_db->transaction_active) {
497                         DEBUG(DEBUG_INFO, (__location__ " refusing migration"
498                               " of key %s while transaction is active\n",
499                               (char *)call->key.dptr));
500                 } else {
501                         DEBUG(DEBUG_DEBUG,("pnn %u starting migration of %08x to %u\n",
502                                  ctdb->pnn, ctdb_hash(&(call->key)), c->hdr.srcnode));
503                         ctdb_call_send_dmaster(ctdb_db, c, &header, &(call->key), &data);
504                         talloc_free(data.dptr);
505
506                         ret = ctdb_ltdb_unlock(ctdb_db, call->key);
507                         if (ret != 0) {
508                                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
509                         }
510                         return;
511                 }
512         }
513
514         ctdb_call_local(ctdb_db, call, &header, hdr, &data);
515
516         ret = ctdb_ltdb_unlock(ctdb_db, call->key);
517         if (ret != 0) {
518                 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", ret));
519         }
520
521         len = offsetof(struct ctdb_reply_call, data) + call->reply_data.dsize;
522         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REPLY_CALL, len, 
523                                     struct ctdb_reply_call);
524         CTDB_NO_MEMORY_FATAL(ctdb, r);
525         r->hdr.destnode  = hdr->srcnode;
526         r->hdr.reqid     = hdr->reqid;
527         r->status        = call->status;
528         r->datalen       = call->reply_data.dsize;
529         if (call->reply_data.dsize) {
530                 memcpy(&r->data[0], call->reply_data.dptr, call->reply_data.dsize);
531         }
532
533         ctdb_queue_packet(ctdb, &r->hdr);
534
535         talloc_free(r);
536 }
537
538 /*
539   called when a CTDB_REPLY_CALL packet comes in
540
541   This packet comes in response to a CTDB_REQ_CALL request packet. It
542   contains any reply data from the call
543 */
544 void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
545 {
546         struct ctdb_reply_call *c = (struct ctdb_reply_call *)hdr;
547         struct ctdb_call_state *state;
548
549         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
550         if (state == NULL) {
551                 DEBUG(DEBUG_ERR, (__location__ " reqid %u not found\n", hdr->reqid));
552                 return;
553         }
554
555         if (hdr->reqid != state->reqid) {
556                 /* we found a record  but it was the wrong one */
557                 DEBUG(DEBUG_ERR, ("Dropped orphaned call reply with reqid:%u\n",hdr->reqid));
558                 return;
559         }
560
561         state->call->reply_data.dptr = c->data;
562         state->call->reply_data.dsize = c->datalen;
563         state->call->status = c->status;
564
565         talloc_steal(state, c);
566
567         state->state = CTDB_CALL_DONE;
568         if (state->async.fn) {
569                 state->async.fn(state);
570         }
571 }
572
573
574 /*
575   called when a CTDB_REPLY_DMASTER packet comes in
576
577   This packet comes in from the lmaster response to a CTDB_REQ_CALL
578   request packet. It means that the current dmaster wants to give us
579   the dmaster role
580 */
581 void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
582 {
583         struct ctdb_reply_dmaster *c = (struct ctdb_reply_dmaster *)hdr;
584         struct ctdb_db_context *ctdb_db;
585         TDB_DATA key, data;
586         int ret;
587
588         ctdb_db = find_ctdb_db(ctdb, c->db_id);
589         if (ctdb_db == NULL) {
590                 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in ctdb_reply_dmaster\n", c->db_id));
591                 return;
592         }
593         
594         key.dptr = c->data;
595         key.dsize = c->keylen;
596         data.dptr = &c->data[key.dsize];
597         data.dsize = c->datalen;
598
599         ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr,
600                                      ctdb_call_input_pkt, ctdb, False);
601         if (ret == -2) {
602                 return;
603         }
604         if (ret != 0) {
605                 DEBUG(DEBUG_ERR,(__location__ " Failed to get lock in ctdb_reply_dmaster\n"));
606                 return;
607         }
608
609         ctdb_become_dmaster(ctdb_db, hdr, key, data, c->rsn);
610 }
611
612
613 /*
614   called when a CTDB_REPLY_ERROR packet comes in
615 */
616 void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
617 {
618         struct ctdb_reply_error *c = (struct ctdb_reply_error *)hdr;
619         struct ctdb_call_state *state;
620
621         state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
622         if (state == NULL) {
623                 DEBUG(DEBUG_ERR,("pnn %u Invalid reqid %u in ctdb_reply_error\n",
624                          ctdb->pnn, hdr->reqid));
625                 return;
626         }
627
628         if (hdr->reqid != state->reqid) {
629                 /* we found a record  but it was the wrong one */
630                 DEBUG(DEBUG_ERR, ("Dropped orphaned error reply with reqid:%u\n",hdr->reqid));
631                 return;
632         }
633
634         talloc_steal(state, c);
635
636         state->state  = CTDB_CALL_ERROR;
637         state->errmsg = (char *)c->msg;
638         if (state->async.fn) {
639                 state->async.fn(state);
640         }
641 }
642
643
644 /*
645   destroy a ctdb_call
646 */
647 static int ctdb_call_destructor(struct ctdb_call_state *state)
648 {
649         DLIST_REMOVE(state->ctdb_db->ctdb->pending_calls, state);
650         ctdb_reqid_remove(state->ctdb_db->ctdb, state->reqid);
651         return 0;
652 }
653
654
655 /*
656   called when a ctdb_call needs to be resent after a reconfigure event
657 */
658 static void ctdb_call_resend(struct ctdb_call_state *state)
659 {
660         struct ctdb_context *ctdb = state->ctdb_db->ctdb;
661
662         state->generation = ctdb->vnn_map->generation;
663
664         /* use a new reqid, in case the old reply does eventually come in */
665         ctdb_reqid_remove(ctdb, state->reqid);
666         state->reqid = ctdb_reqid_new(ctdb, state);
667         state->c->hdr.reqid = state->reqid;
668
669         /* update the generation count for this request, so its valid with the new vnn_map */
670         state->c->hdr.generation = state->generation;
671
672         /* send the packet to ourselves, it will be redirected appropriately */
673         state->c->hdr.destnode = ctdb->pnn;
674
675         ctdb_queue_packet(ctdb, &state->c->hdr);
676         DEBUG(DEBUG_NOTICE,("resent ctdb_call\n"));
677 }
678
679 /*
680   resend all pending calls on recovery
681  */
682 void ctdb_call_resend_all(struct ctdb_context *ctdb)
683 {
684         struct ctdb_call_state *state, *next;
685         for (state=ctdb->pending_calls;state;state=next) {
686                 next = state->next;
687                 ctdb_call_resend(state);
688         }
689 }
690
691 /*
692   this allows the caller to setup a async.fn 
693 */
694 static void call_local_trigger(struct event_context *ev, struct timed_event *te, 
695                        struct timeval t, void *private_data)
696 {
697         struct ctdb_call_state *state = talloc_get_type(private_data, struct ctdb_call_state);
698         if (state->async.fn) {
699                 state->async.fn(state);
700         }
701 }       
702
703
704 /*
705   construct an event driven local ctdb_call
706
707   this is used so that locally processed ctdb_call requests are processed
708   in an event driven manner
709 */
710 struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, 
711                                              struct ctdb_call *call,
712                                              struct ctdb_ltdb_header *header,
713                                              TDB_DATA *data)
714 {
715         struct ctdb_call_state *state;
716         struct ctdb_context *ctdb = ctdb_db->ctdb;
717         int ret;
718
719         state = talloc_zero(ctdb_db, struct ctdb_call_state);
720         CTDB_NO_MEMORY_NULL(ctdb, state);
721
722         talloc_steal(state, data->dptr);
723
724         state->state = CTDB_CALL_DONE;
725         state->call  = talloc(state, struct ctdb_call);
726         CTDB_NO_MEMORY_NULL(ctdb, state->call);
727         *(state->call) = *call;
728         state->ctdb_db = ctdb_db;
729
730         ret = ctdb_call_local(ctdb_db, state->call, header, state, data);
731
732         event_add_timed(ctdb->ev, state, timeval_zero(), call_local_trigger, state);
733
734         return state;
735 }
736
737
738 /*
739   make a remote ctdb call - async send. Called in daemon context.
740
741   This constructs a ctdb_call request and queues it for processing. 
742   This call never blocks.
743 */
744 struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctdb_db, 
745                                                      struct ctdb_call *call, 
746                                                      struct ctdb_ltdb_header *header)
747 {
748         uint32_t len;
749         struct ctdb_call_state *state;
750         struct ctdb_context *ctdb = ctdb_db->ctdb;
751
752         if (ctdb->methods == NULL) {
753                 DEBUG(DEBUG_INFO,(__location__ " Failed send packet. Transport is down\n"));
754                 return NULL;
755         }
756
757         state = talloc_zero(ctdb_db, struct ctdb_call_state);
758         CTDB_NO_MEMORY_NULL(ctdb, state);
759         state->call = talloc(state, struct ctdb_call);
760         CTDB_NO_MEMORY_NULL(ctdb, state->call);
761
762         state->reqid = ctdb_reqid_new(ctdb, state);
763         state->ctdb_db = ctdb_db;
764         talloc_set_destructor(state, ctdb_call_destructor);
765
766         len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
767         state->c = ctdb_transport_allocate(ctdb, state, CTDB_REQ_CALL, len, 
768                                            struct ctdb_req_call);
769         CTDB_NO_MEMORY_NULL(ctdb, state->c);
770         state->c->hdr.destnode  = header->dmaster;
771
772         /* this limits us to 16k outstanding messages - not unreasonable */
773         state->c->hdr.reqid     = state->reqid;
774         state->c->flags         = call->flags;
775         state->c->db_id         = ctdb_db->db_id;
776         state->c->callid        = call->call_id;
777         state->c->hopcount      = 0;
778         state->c->keylen        = call->key.dsize;
779         state->c->calldatalen   = call->call_data.dsize;
780         memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
781         memcpy(&state->c->data[call->key.dsize], 
782                call->call_data.dptr, call->call_data.dsize);
783         *(state->call)              = *call;
784         state->call->call_data.dptr = &state->c->data[call->key.dsize];
785         state->call->key.dptr       = &state->c->data[0];
786
787         state->state  = CTDB_CALL_WAIT;
788         state->generation = ctdb->vnn_map->generation;
789
790         DLIST_ADD(ctdb->pending_calls, state);
791
792         ctdb_queue_packet(ctdb, &state->c->hdr);
793
794         return state;
795 }
796
797 /*
798   make a remote ctdb call - async recv - called in daemon context
799
800   This is called when the program wants to wait for a ctdb_call to complete and get the 
801   results. This call will block unless the call has already completed.
802 */
803 int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
804 {
805         while (state->state < CTDB_CALL_DONE) {
806                 event_loop_once(state->ctdb_db->ctdb->ev);
807         }
808         if (state->state != CTDB_CALL_DONE) {
809                 ctdb_set_error(state->ctdb_db->ctdb, "%s", state->errmsg);
810                 talloc_free(state);
811                 return -1;
812         }
813
814         if (state->call->reply_data.dsize) {
815                 call->reply_data.dptr = talloc_memdup(call,
816                                                       state->call->reply_data.dptr,
817                                                       state->call->reply_data.dsize);
818                 call->reply_data.dsize = state->call->reply_data.dsize;
819         } else {
820                 call->reply_data.dptr = NULL;
821                 call->reply_data.dsize = 0;
822         }
823         call->status = state->call->status;
824         talloc_free(state);
825         return 0;
826 }
827
828
829 /* 
830    send a keepalive packet to the other node
831 */
832 void ctdb_send_keepalive(struct ctdb_context *ctdb, uint32_t destnode)
833 {
834         struct ctdb_req_keepalive *r;
835         
836         if (ctdb->methods == NULL) {
837                 DEBUG(DEBUG_INFO,(__location__ " Failed to send keepalive. Transport is DOWN\n"));
838                 return;
839         }
840
841         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REQ_KEEPALIVE,
842                                     sizeof(struct ctdb_req_keepalive), 
843                                     struct ctdb_req_keepalive);
844         CTDB_NO_MEMORY_FATAL(ctdb, r);
845         r->hdr.destnode  = destnode;
846         r->hdr.reqid     = 0;
847         
848         CTDB_INCREMENT_STAT(ctdb, keepalive_packets_sent);
849
850         ctdb_queue_packet(ctdb, &r->hdr);
851
852         talloc_free(r);
853 }