ctdb/common/sock_client.c

   1 /*
   2    A client based on unix domain socket
   3
   4    Copyright (C) Amitay Isaacs  2017
   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 #include "replace.h"
  21 #include "system/filesys.h"
  22 #include "system/network.h"
  23
  24 #include <talloc.h>
  25 #include <tevent.h>
  26
  27 #include "lib/util/debug.h"
  28 #include "lib/util/time.h"
  29 #include "lib/util/tevent_unix.h"
  30
  31 #include "common/logging.h"
  32 #include "common/reqid.h"
  33 #include "common/comm.h"
  34 #include "common/sock_client.h"
  35
  36 struct sock_client_context {
  37         struct sock_client_proto_funcs *funcs;
  38         void *private_data;
  39
  40         void (*disconnect_callback)(void *private_data);
  41         void *disconnect_data;
  42
  43         int fd;
  44         struct comm_context *comm;
  45         struct reqid_context *idr;
  46 };
  47
  48 /*
  49  * connect to a unix domain socket
  50  */
  51
  52 static int socket_connect(const char *sockpath)
  53 {
  54         struct sockaddr_un addr;
  55         size_t len;
  56         int fd, ret;
  57
  58         memset(&addr, 0, sizeof(addr));
  59         addr.sun_family = AF_UNIX;
  60
  61         len = strlcpy(addr.sun_path, sockpath, sizeof(addr.sun_path));
  62         if (len >= sizeof(addr.sun_path)) {
  63                 D_ERR("socket path too long: %s\n", sockpath);
  64                 return -1;
  65         }
  66
  67         fd = socket(AF_UNIX, SOCK_STREAM, 0);
  68         if (fd == -1) {
  69                 D_ERR("socket create failed - %s\n", sockpath);
  70                 return -1;
  71         }
  72
  73         ret = connect(fd, (struct sockaddr *)&addr, sizeof(addr));
  74         if (ret != 0) {
  75                 D_ERR("socket connect failed - %s\n", sockpath);
  76                 close(fd);
  77                 return -1;
  78         }
  79
  80         return fd;
  81 }
  82
  83 /*
  84  * Socket client
  85  */
  86
  87 static int sock_client_context_destructor(struct sock_client_context *sockc);
  88 static void sock_client_read_handler(uint8_t *buf, size_t buflen,
  89                                      void *private_data);
  90 static void sock_client_dead_handler(void *private_data);
  91
  92 static void sock_client_msg_reply(struct sock_client_context *sockc,
  93                                   uint8_t *buf, size_t buflen);
  94
  95 int sock_client_setup(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  96                       const char *sockpath,
  97                       struct sock_client_proto_funcs *funcs,
  98                       void *private_data,
  99                       struct sock_client_context **result)
 100 {
 101         struct sock_client_context *sockc;
 102         int ret;
 103
 104         if (sockpath == NULL) {
 105                 return EINVAL;
 106         }
 107
 108         if (funcs == NULL || funcs->request_push == NULL ||
 109             funcs->reply_pull == NULL || funcs->reply_reqid == NULL) {
 110                 return EINVAL;
 111         }
 112
 113         sockc = talloc_zero(mem_ctx, struct sock_client_context);
 114         if (sockc == NULL) {
 115                 return ENOMEM;
 116         }
 117
 118         sockc->funcs = funcs;
 119         sockc->private_data = private_data;
 120
 121         sockc->fd = socket_connect(sockpath);
 122         if (sockc->fd == -1) {
 123                 talloc_free(sockc);
 124                 return EIO;
 125         }
 126
 127         ret = comm_setup(sockc, ev, sockc->fd,
 128                          sock_client_read_handler, sockc,
 129                          sock_client_dead_handler, sockc,
 130                          &sockc->comm);
 131         if (ret != 0) {
 132                 D_ERR("comm_setup() failed, ret=%d\n", ret);
 133                 close(sockc->fd);
 134                 talloc_free(sockc);
 135                 return ret;
 136         }
 137
 138         ret = reqid_init(sockc, INT_MAX-200, &sockc->idr);
 139         if (ret != 0) {
 140                 D_ERR("reqid_init() failed, ret=%d\n", ret);
 141                 close(sockc->fd);
 142                 talloc_free(sockc);
 143                 return ret;
 144         }
 145
 146         talloc_set_destructor(sockc, sock_client_context_destructor);
 147
 148         *result = sockc;
 149         return 0;
 150 }
 151
 152 static int sock_client_context_destructor(struct sock_client_context *sockc)
 153 {
 154         TALLOC_FREE(sockc->comm);
 155         if (sockc->fd != -1) {
 156                 close(sockc->fd);
 157                 sockc->fd = -1;
 158         }
 159         return 0;
 160 }
 161
 162
 163 static void sock_client_read_handler(uint8_t *buf, size_t buflen,
 164                                      void *private_data)
 165 {
 166         struct sock_client_context *sockc = talloc_get_type_abort(
 167                 private_data, struct sock_client_context);
 168
 169         sock_client_msg_reply(sockc, buf, buflen);
 170 }
 171
 172 static void sock_client_dead_handler(void *private_data)
 173 {
 174         struct sock_client_context *sockc = talloc_get_type_abort(
 175                 private_data, struct sock_client_context);
 176
 177         if (sockc->disconnect_callback != NULL) {
 178                 sockc->disconnect_callback(sockc->disconnect_data);
 179                 talloc_free(sockc);
 180                 return;
 181         }
 182
 183         D_NOTICE("connection to daemon closed, exiting\n");
 184         exit(1);
 185 }
 186
 187 void sock_client_set_disconnect_callback(struct sock_client_context *sockc,
 188                                          sock_client_callback_func_t callback,
 189                                          void *private_data)
 190 {
 191         sockc->disconnect_callback = callback;
 192         sockc->disconnect_data = private_data;
 193 }
 194
 195
 196 struct sock_client_msg_state {
 197         struct sock_client_context *sockc;
 198         uint32_t reqid;
 199         struct tevent_req *req;
 200         void *reply;
 201 };
 202
 203 static int sock_client_msg_state_destructor(
 204                                 struct sock_client_msg_state *state);
 205 static void sock_client_msg_done(struct tevent_req *subreq);
 206
 207 struct tevent_req *sock_client_msg_send(TALLOC_CTX *mem_ctx,
 208                                         struct tevent_context *ev,
 209                                         struct sock_client_context *sockc,
 210                                         struct timeval timeout,
 211                                         void *request)
 212 {
 213         struct tevent_req *req, *subreq;
 214         struct sock_client_msg_state *state;
 215         uint8_t *buf;
 216         size_t buflen;
 217         int ret;
 218
 219         req = tevent_req_create(mem_ctx, &state, struct sock_client_msg_state);
 220         if (req == NULL) {
 221                 return NULL;
 222         }
 223
 224         state->sockc = sockc;
 225
 226         state->reqid = reqid_new(sockc->idr, state);
 227         if (state->reqid == REQID_INVALID) {
 228                 talloc_free(req);
 229                 return NULL;
 230         }
 231
 232         state->req = req;
 233
 234         talloc_set_destructor(state, sock_client_msg_state_destructor);
 235
 236         ret = sockc->funcs->request_push(request, state->reqid, state,
 237                                          &buf, &buflen, sockc->private_data);
 238         if (ret != 0) {
 239                 tevent_req_error(req, ret);
 240                 return tevent_req_post(req, ev);
 241         }
 242
 243         subreq = comm_write_send(state, ev, sockc->comm, buf, buflen);
 244         if (tevent_req_nomem(subreq, req)) {
 245                 return tevent_req_post(req, ev);
 246         }
 247         tevent_req_set_callback(subreq, sock_client_msg_done, req);
 248
 249         if (! timeval_is_zero(&timeout)) {
 250                 if (!tevent_req_set_endtime(req, ev, timeout)) {
 251                         return tevent_req_post(req, ev);
 252                 }
 253         }
 254
 255         return req;
 256 }
 257
 258 static int sock_client_msg_state_destructor(
 259                                 struct sock_client_msg_state *state)
 260 {
 261         reqid_remove(state->sockc->idr, state->reqid);
 262         return 0;
 263 }
 264
 265 static void sock_client_msg_done(struct tevent_req *subreq)
 266 {
 267         struct tevent_req *req = tevent_req_callback_data(
 268                 subreq, struct tevent_req);
 269         int ret;
 270         bool status;
 271
 272         status = comm_write_recv(subreq, &ret);
 273         TALLOC_FREE(subreq);
 274         if (! status) {
 275                 tevent_req_error(req, ret);
 276                 return;
 277         }
 278
 279         /* wait for the reply or timeout */
 280 }
 281
 282 static void sock_client_msg_reply(struct sock_client_context *sockc,
 283                                   uint8_t *buf, size_t buflen)
 284 {
 285         struct sock_client_msg_state *state;
 286         uint32_t reqid;
 287         int ret;
 288
 289         ret = sockc->funcs->reply_reqid(buf, buflen, &reqid,
 290                                         sockc->private_data);
 291         if (ret != 0) {
 292                 D_WARNING("Invalid packet received, ret=%d\n", ret);
 293                 return;
 294         }
 295
 296         state = reqid_find(sockc->idr, reqid, struct sock_client_msg_state);
 297         if (state == NULL) {
 298                 return;
 299         }
 300
 301         if (reqid != state->reqid) {
 302                 return;
 303         }
 304
 305         ret = sockc->funcs->reply_pull(buf, buflen, state, &state->reply,
 306                                        sockc->private_data);
 307         if (ret != 0) {
 308                 tevent_req_error(state->req, ret);
 309                 return;
 310         }
 311
 312         tevent_req_done(state->req);
 313 }
 314
 315 bool sock_client_msg_recv(struct tevent_req *req, int *perr,
 316                           TALLOC_CTX *mem_ctx, void *reply)
 317 {
 318         struct sock_client_msg_state *state = tevent_req_data(
 319                 req, struct sock_client_msg_state);
 320         int ret;
 321
 322         if (tevent_req_is_unix_error(req, &ret)) {
 323                 if (perr != NULL) {
 324                         *perr = ret;
 325                 }
 326                 return false;
 327         }
 328
 329         if (reply != NULL) {
 330                 *(void **)reply = talloc_steal(mem_ctx, state->reply);
 331         }
 332
 333         return true;
 334 }