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         c->record_data.dsize = data.dsize;
  47         CTDB_NO_MEMORY(ctdb, c->record_data.dptr);
  48         if (data.dptr) free(data.dptr);
  49         c->new_data = NULL;
  50         c->reply_data = NULL;
  51
  52         for (fn=ctdb->calls;fn;fn=fn->next) {
  53                 if (fn->id == call_id) break;
  54         }
  55         if (fn == NULL) {
  56                 ctdb_set_error(ctdb, "Unknown call id %u\n", call_id);
  57                 return -1;
  58         }
  59
  60         if (fn->fn(c) != 0) {
  61                 free(c->record_data.dptr);
  62                 ctdb_set_error(ctdb, "ctdb_call %u failed\n", call_id);
  63                 return -1;
  64         }
  65
  66         if (c->new_data) {
  67                 if (tdb_store(ctdb->ltdb, key, *c->new_data, TDB_REPLACE) != 0) {
  68                         ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
  69                         return -1;
  70                 }
  71         }
  72
  73         if (reply_data) {
  74                 if (c->reply_data) {
  75                         *reply_data = *c->reply_data;
  76                         talloc_steal(ctdb, reply_data->dptr);
  77                 } else {
  78                         reply_data->dptr = NULL;
  79                         reply_data->dsize = 0;
  80                 }
  81         }
  82
  83         talloc_free(c);
  84
  85         return 0;
  86 }
  87
  88 /*
  89   make a remote ctdb call
  90 */
  91 int ctdb_call(struct ctdb_context *ctdb, TDB_DATA key, int call_id,
  92               TDB_DATA *call_data, TDB_DATA *reply_data)
  93 {
  94         uint32_t dest;
  95         struct ctdb_req_call *c;
  96         uint32_t len;
  97         struct ctdb_node *node;
  98
  99         dest = ctdb_hash(&key) % ctdb->num_nodes;
 100         if (dest == ctdb->vnn) {
 101                 return ctdb_call_local(ctdb, key, call_id, call_data, reply_data);
 102         }
 103
 104         len = sizeof(*c) + key.dsize + (call_data?call_data->dsize:0);
 105         c = talloc_size(ctdb, len);
 106         CTDB_NO_MEMORY(ctdb, c);
 107
 108         c->hdr.operation = CTDB_OP_CALL;
 109         c->hdr.destnode  = dest;
 110         c->hdr.srcnode   = ctdb->vnn;
 111         /* this limits us to 16k outstanding messages - not unreasonable */
 112         c->hdr.reqid     = idr_get_new(ctdb->idr, c, 0xFFFF);
 113         c->callid        = call_id;
 114         c->keylen        = key.dsize;
 115         c->calldatalen   = call_data?call_data->dsize:0;
 116         memcpy(&c->data[0], key.dptr, key.dsize);
 117         if (call_data) {
 118                 memcpy(&c->data[key.dsize], call_data->dptr, call_data->dsize);
 119         }
 120
 121         node = ctdb->nodes[dest];
 122
 123         if (ctdb->methods->queue_pkt(node, (uint8_t *)c, len) != 0) {
 124                 talloc_free(c);
 125                 return -1;
 126         }
 127
 128         /*
 129         event_add_timed(ctdb->ev, c, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
 130                         ctdb_call_timeout, c);
 131         */
 132         return -1;
 133 }
 134