direct/ctdbd_test.c

   1 /*
   2    test of messaging
   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 "system/network.h"
  23 #include "../include/ctdb.h"
  24 #include "../include/ctdb_private.h"
  25
  26 #define CTDB_SOCKET "/tmp/ctdb.socket.127.0.0.1"
  27
  28
  29 /*
  30   connect to the unix domain socket
  31 */
  32 static int ux_socket_connect(const char *name)
  33 {
  34         struct sockaddr_un addr;
  35         int fd;
  36
  37         memset(&addr, 0, sizeof(addr));
  38         addr.sun_family = AF_UNIX;
  39         strncpy(addr.sun_path, name, sizeof(addr.sun_path));
  40
  41         fd = socket(AF_UNIX, SOCK_STREAM, 0);
  42         if (fd == -1) {
  43                 return -1;
  44         }
  45
  46         if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
  47                 close(fd);
  48                 return -1;
  49         }
  50
  51         return fd;
  52 }
  53
  54 void register_pid_with_daemon(int fd, int pid)
  55 {
  56         struct ctdb_req_register r;
  57
  58         bzero(&r, sizeof(r));
  59         r.hdr.length       = sizeof(r);
  60         r.hdr.ctdb_magic   = CTDB_MAGIC;
  61         r.hdr.ctdb_version = CTDB_VERSION;
  62         r.hdr.operation    = CTDB_REQ_REGISTER;
  63         r.srvid            = pid;
  64
  65         /* XXX must deal with partial writes here */
  66         write(fd, &r, sizeof(r));
  67 }
  68
  69 /* send a command to the cluster to wait until all nodes are connected
  70    and the cluster is fully operational
  71  */
  72 int wait_for_cluster(int fd)
  73 {
  74         struct ctdb_req_connect_wait req;
  75         struct ctdb_reply_connect_wait rep;
  76         int cnt, tot;
  77
  78         /* send a connect wait command to the local node */
  79         bzero(&req, sizeof(req));
  80         req.hdr.length       = sizeof(req);
  81         req.hdr.ctdb_magic   = CTDB_MAGIC;
  82         req.hdr.ctdb_version = CTDB_VERSION;
  83         req.hdr.operation    = CTDB_REQ_CONNECT_WAIT;
  84
  85         /* XXX must deal with partial writes here */
  86         write(fd, &req, sizeof(req));
  87
  88
  89         /* read the 4 bytes of length for the pdu */
  90         cnt=0;
  91         tot=4;
  92         while(cnt!=tot){
  93                 int numread;
  94                 numread=read(fd, ((char *)&rep)+cnt, tot-cnt);
  95                 if(numread>0){
  96                         cnt+=numread;
  97                 }
  98         }
  99         /* read the rest of the pdu */
 100         tot=rep.hdr.length;
 101         while(cnt!=tot){
 102                 int numread;
 103                 numread=read(fd, ((char *)&rep)+cnt, tot-cnt);
 104                 if(numread>0){
 105                         cnt+=numread;
 106                 }
 107         }
 108
 109         return rep.vnn;
 110 }
 111
 112
 113 int send_a_message(int fd, int ourvnn, int vnn, int pid, TDB_DATA data)
 114 {
 115         struct ctdb_req_message r;
 116         int len, cnt;
 117
 118         len = offsetof(struct ctdb_req_message, data) + data.dsize;
 119         r.hdr.length     = len;
 120         r.hdr.ctdb_magic = CTDB_MAGIC;
 121         r.hdr.ctdb_version = CTDB_VERSION;
 122         r.hdr.operation  = CTDB_REQ_MESSAGE;
 123         r.hdr.destnode   = vnn;
 124         r.hdr.srcnode    = ourvnn;
 125         r.hdr.reqid      = 0;
 126         r.srvid          = pid;
 127         r.datalen        = data.dsize;
 128
 129         /* write header */
 130         cnt=write(fd, &r, offsetof(struct ctdb_req_message, data));
 131         /* write data */
 132         if(data.dsize){
 133             cnt+=write(fd, data.dptr, data.dsize);
 134         }
 135 }
 136
 137 void wait_for_a_message(int fd)
 138 {
 139         int cnt, tot;
 140         uint32_t len;
 141         struct ctdb_req_message *msg;
 142
 143         /* read the 4 bytes of length for the pdu */
 144         cnt=0;
 145         tot=4;
 146         while(cnt!=tot){
 147                 int numread;
 148                 numread=read(fd, ((char *)&len)+cnt, tot-cnt);
 149                 if(numread>0){
 150                         cnt+=numread;
 151                 }
 152         }
 153         msg=malloc(len);
 154         msg->hdr.length=len;
 155         /* read the rest of the pdu */
 156         tot=msg->hdr.length;
 157         while(cnt!=tot){
 158                 int numread;
 159                 numread=read(fd, (char *)msg+cnt, tot-cnt);
 160                 if(numread>0){
 161                         cnt+=numread;
 162                 }
 163         }
 164         printf("got a message : %s\n",&msg->data[0]);
 165 }
 166
 167 int main(int argc, const char *argv[])
 168 {
 169         int fd, pid, vnn, dstvnn, dstpid;
 170         TDB_DATA message;
 171
 172         /* open the socket to talk to the local ctdb daemon */
 173         fd=ux_socket_connect(CTDB_SOCKET);
 174         if (fd==-1) {
 175                 printf("failed to open domain socket\n");
 176                 exit(10);
 177         }
 178
 179
 180         /* register our local server id with the daemon so that it knows
 181            where to send messages addressed to our local pid.
 182          */
 183         pid=getpid();
 184         register_pid_with_daemon(fd, pid);
 185
 186
 187         /* do a connect wait to ensure that all nodes in the cluster are up
 188            and operational.
 189            this also tells us the vnn of the local cluster.
 190            If someone wants to send us a emssage they should send it to
 191            this vnn and our pid
 192          */
 193         vnn=wait_for_cluster(fd);
 194         printf("our address is vnn:%d pid:%d  if someone wants to send us a message!\n",vnn,pid);
 195
 196
 197         /* send a message to ourself */
 198         dstvnn=vnn;
 199         dstpid=pid;
 200         message.dptr="Test message";
 201         message.dsize=strlen(message.dptr)+1;
 202         send_a_message(fd, vnn, dstvnn, dstpid, message);
 203
 204
 205         /* wait for the message to come back.
 206            i.e. the one we just sent to ourself
 207          */
 208         wait_for_a_message(fd);
 209
 210         return 0;
 211 }