ctdb/common/ctdb_call.c

   1 /*
   2    ctdb_call protocol code
   3
   4    Copyright (C) Andrew Tridgell  2006
   5
   6    This library is free software; you can redistribute it and/or
   7    modify it under the terms of the GNU Lesser General Public
   8    License as published by the Free Software Foundation; either
   9    version 2 of the License, or (at your option) any later version.
  10
  11    This library 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 GNU
  14    Lesser General Public License for more details.
  15
  16    You should have received a copy of the GNU Lesser General Public
  17    License along with this library; if not, write to the Free Software
  18    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19 */
  20
  21 #include "includes.h"
  22 #include "lib/events/events.h"
  23 #include "system/network.h"
  24 #include "system/filesys.h"
  25 #include "ctdb_private.h"
  26
  27
  28 /*
  29   local version of ctdb_call
  30 */
  31 static int ctdb_call_local(struct ctdb_context *ctdb, TDB_DATA key, int call_id,
  32                            TDB_DATA *call_data, TDB_DATA *reply_data)
  33 {
  34         struct ctdb_call *c;
  35         struct ctdb_registered_call *fn;
  36         TDB_DATA data;
  37
  38         c = talloc(ctdb, struct ctdb_call);
  39         CTDB_NO_MEMORY(ctdb, c);
  40
  41         data = tdb_fetch(ctdb->ltdb, key);
  42
  43         c->key = key;
  44         c->call_data = call_data;
  45         c->record_data.dptr = talloc_memdup(c, data.dptr, data.dsize);
  46         CTDB_NO_MEMORY(ctdb, c->record_data.dptr);
  47         if (data.dptr) free(data.dptr);
  48         c->new_data = NULL;
  49         c->reply_data = NULL;
  50
  51         for (fn=ctdb->calls;fn;fn=fn->next) {
  52                 if (fn->id == call_id) break;
  53         }
  54         if (fn == NULL) {
  55                 ctdb_set_error(ctdb, "Unknown call id %u\n", call_id);
  56                 return -1;
  57         }
  58
  59         if (fn->fn(c) != 0) {
  60                 free(c->record_data.dptr);
  61                 ctdb_set_error(ctdb, "ctdb_call %u failed\n", call_id);
  62                 return -1;
  63         }
  64
  65         if (c->new_data) {
  66                 if (tdb_store(ctdb->ltdb, key, *c->new_data, TDB_REPLACE) != 0) {
  67                         ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
  68                         return -1;
  69                 }
  70         }
  71
  72         if (reply_data) {
  73                 if (c->reply_data) {
  74                         *reply_data = *c->reply_data;
  75                         talloc_steal(ctdb, reply_data->dptr);
  76                 } else {
  77                         reply_data->dptr = NULL;
  78                         reply_data->dsize = 0;
  79                 }
  80         }
  81
  82         talloc_free(c);
  83
  84         return -1;
  85 }
  86
  87 /*
  88   make a remote ctdb call
  89 */
  90 int ctdb_call(struct ctdb_context *ctdb, TDB_DATA key, int call_id,
  91               TDB_DATA *call_data, TDB_DATA *reply_data)
  92 {
  93         uint32_t dest;
  94         struct ctdb_req_call *c;
  95         uint32_t len;
  96         struct ctdb_node *node;
  97
  98         dest = ctdb_hash(&key) % ctdb->num_nodes;
  99         if (dest == ctdb->vnn) {
 100                 return ctdb_call_local(ctdb, key, call_id, call_data, reply_data);
 101         }
 102
 103         len = sizeof(*c) + key.dsize + (call_data?call_data->dsize:0);
 104         c = talloc_size(ctdb, len);
 105         CTDB_NO_MEMORY(ctdb, c);
 106
 107         c->hdr.operation = CTDB_OP_CALL;
 108         c->hdr.destnode  = dest;
 109         c->hdr.srcnode   = ctdb->vnn;
 110         /* this limits us to 16k outstanding messages - not unreasonable */
 111         c->hdr.reqid     = idr_get_new(ctdb->idr, c, 0xFFFF);
 112         c->callid        = call_id;
 113         c->keylen        = key.dsize;
 114         c->calldatalen   = call_data?call_data->dsize:0;
 115         memcpy(&c->data[0], key.dptr, key.dsize);
 116         if (call_data) {
 117                 memcpy(&c->data[key.dsize], call_data->dptr, call_data->dsize);
 118         }
 119
 120         node = ctdb->nodes[dest];
 121
 122         if (ctdb->methods->queue_pkt(node, (uint8_t *)c, len) != 0) {
 123                 talloc_free(c);
 124                 return -1;
 125         }
 126
 127         /*
 128         event_add_timed(ctdb->ev, c, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
 129                         ctdb_call_timeout, c);
 130         */
 131         return -1;
 132 }
 133