src/vol/daemon_com.c

   1 /*
   2  * Copyright 2006, Sine Nomine Associates and others.
   3  * All Rights Reserved.
   4  *
   5  * This software has been released under the terms of the IBM Public
   6  * License.  For details, see the LICENSE file in the top-level source
   7  * directory or online at http://www.openafs.org/dl/license10.html
   8  */
   9
  10 /*
  11  * localhost interprocess communication for servers
  12  *
  13  * currently handled by a localhost socket
  14  * (yes, this needs to be replaced someday)
  15  */
  16
  17 #ifndef _WIN32
  18 #define FD_SETSIZE 65536
  19 #endif
  20
  21 #include <afsconfig.h>
  22 #include <afs/param.h>
  23
  24 RCSID
  25     ("$Header$");
  26
  27 #include <sys/types.h>
  28 #include <stdio.h>
  29 #ifdef AFS_NT40_ENV
  30 #include <winsock2.h>
  31 #include <time.h>
  32 #else
  33 #include <sys/param.h>
  34 #include <sys/socket.h>
  35 #include <netinet/in.h>
  36 #include <netdb.h>
  37 #include <sys/time.h>
  38 #endif
  39 #include <errno.h>
  40 #include <assert.h>
  41 #include <signal.h>
  42
  43 #ifdef HAVE_STRING_H
  44 #include <string.h>
  45 #else
  46 #ifdef HAVE_STRINGS_H
  47 #include <strings.h>
  48 #endif
  49 #endif
  50
  51
  52 #include <rx/xdr.h>
  53 #include <afs/afsint.h>
  54 #include "nfs.h"
  55 #include <afs/errors.h>
  56 #include "daemon_com.h"
  57 #include "lwp.h"
  58 #include "lock.h"
  59 #include <afs/afssyscalls.h>
  60 #include "ihandle.h"
  61 #include "vnode.h"
  62 #include "volume.h"
  63 #include "partition.h"
  64 #include <rx/rx_queue.h>
  65
  66 #ifdef USE_UNIX_SOCKETS
  67 #include <afs/afsutil.h>
  68 #include <sys/un.h>
  69 #endif
  70
  71 /*@printflike@*/ extern void Log(const char *format, ...);
  72
  73 #ifdef osi_Assert
  74 #undef osi_Assert
  75 #endif
  76 #define osi_Assert(e) (void)(e)
  77
  78 int (*V_BreakVolumeCallbacks) ();
  79
  80 #define MAXHANDLERS     4       /* Up to 4 clients; must be at least 2, so that
  81                                  * move = dump+restore can run on single server */
  82
  83 #define MAX_BIND_TRIES  5       /* Number of times to retry socket bind */
  84
  85 #ifdef USE_UNIX_SOCKETS
  86 static getport(SYNC_client_state * state, struct sockaddr_un *addr);
  87 #else  /* USE_UNIX_SOCKETS */
  88 static getport(SYNC_client_state * state, struct sockaddr_in *addr);
  89 #endif /* USE_UNIX_SOCKETS */
  90
  91 static int SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res);
  92
  93 /* daemon com SYNC client interface */
  94
  95 int
  96 SYNC_connect(SYNC_client_state * state)
  97 {
  98 #ifdef USE_UNIX_SOCKETS
  99     struct sockaddr_un addr;
 100 #else /* USE_UNIX_SOCKETS */
 101     struct sockaddr_in addr;
 102 #endif /* USE_UNIX_SOCKETS */
 103     /* I can't believe the following is needed for localhost connections!! */
 104     static time_t backoff[] =
 105         { 3, 3, 3, 5, 5, 5, 7, 15, 16, 24, 32, 40, 48, 0 };
 106     time_t *timeout = &backoff[0];
 107
 108     if (state->fd >= 0) {
 109         return 1;
 110     }
 111
 112     for (;;) {
 113         state->fd = getport(state, &addr);
 114         if (connect(state->fd, (struct sockaddr *)&addr, sizeof(addr)) >= 0)
 115             return 1;
 116         if (!*timeout)
 117             break;
 118         if (!(*timeout & 1))
 119             Log("SYNC_connect temporary failure (will retry)\n");
 120         SYNC_disconnect(state);
 121         sleep(*timeout++);
 122     }
 123     perror("SYNC_connect failed (giving up!)");
 124     return 0;
 125 }
 126
 127 int
 128 SYNC_disconnect(SYNC_client_state * state)
 129 {
 130 #ifdef AFS_NT40_ENV
 131     closesocket(state->fd);
 132 #else
 133     close(state->fd);
 134 #endif
 135     state->fd = -1;
 136     return 0;
 137 }
 138
 139 afs_int32
 140 SYNC_closeChannel(SYNC_client_state * state)
 141 {
 142     afs_int32 code;
 143     SYNC_command com;
 144     SYNC_response res;
 145     SYNC_PROTO_BUF_DECL(ores);
 146
 147     if (state->fd == -1)
 148         return SYNC_OK;
 149
 150     memset(&com, 0, sizeof(com));
 151     memset(&res, 0, sizeof(res));
 152
 153     res.payload.len = SYNC_PROTO_MAX_LEN;
 154     res.payload.buf = ores;
 155
 156     com.hdr.command = SYNC_COM_CHANNEL_CLOSE;
 157     com.hdr.command_len = sizeof(SYNC_command_hdr);
 158
 159     /* in case the other end dropped, don't do any retries */
 160     state->retry_limit = 0;
 161     state->hard_timeout = 0;
 162
 163     code = SYNC_ask(state, &com, &res);
 164
 165     if (code == SYNC_OK) {
 166         if (res.hdr.response != SYNC_OK) {
 167             Log("SYNC_closeChannel:  channel shutdown request denied; closing socket anyway\n");
 168         } else if (!(res.hdr.flags & SYNC_FLAG_CHANNEL_SHUTDOWN)) {
 169             Log("SYNC_closeChannel:  channel shutdown request mishandled by server\n");
 170         }
 171     } else {
 172         Log("SYNC_closeChannel: channel communications problem");
 173     }
 174
 175     SYNC_disconnect(state);
 176
 177     return code;
 178 }
 179
 180 int
 181 SYNC_reconnect(SYNC_client_state * state)
 182 {
 183     SYNC_disconnect(state);
 184     return SYNC_connect(state);
 185 }
 186
 187 /* private function to fill in the sockaddr struct for us */
 188 #ifdef USE_UNIX_SOCKETS
 189 static int
 190 getport(SYNC_client_state * state, struct sockaddr_un *addr)
 191 {
 192     int sd;
 193     char tbuffer[AFSDIR_PATH_MAX];
 194
 195     strcompose(tbuffer, AFSDIR_PATH_MAX, AFSDIR_SERVER_LOCAL_DIRPATH, "/",
 196                "fssync.sock", NULL);
 197     memset(addr, 0, sizeof(*addr));
 198     addr->sun_family = AF_UNIX;
 199     strncpy(addr->sun_path, tbuffer, (sizeof(struct sockaddr_un) - sizeof(short)));
 200     assert((sd = socket(AF_UNIX, SOCK_STREAM, 0)) >= 0);
 201     return sd;
 202 }
 203 #else  /* USE_UNIX_SOCKETS */
 204 static int
 205 getport(SYNC_client_state * state, struct sockaddr_in *addr)
 206 {
 207     int sd;
 208     memset(addr, 0, sizeof(*addr));
 209     assert((sd = socket(AF_INET, SOCK_STREAM, 0)) >= 0);
 210 #ifdef STRUCT_SOCKADDR_HAS_SA_LEN
 211     addr->sin_len = sizeof(struct sockaddr_in);
 212 #endif
 213     addr->sin_addr.s_addr = htonl(0x7f000001);
 214     addr->sin_family = AF_INET; /* was localhost->h_addrtype */
 215     addr->sin_port = htons(state->port);        /* XXXX htons not _really_ neccessary */
 216     return sd;
 217 }
 218 #endif /* USE_UNIX_SOCKETS */
 219
 220 afs_int32
 221 SYNC_ask(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
 222 {
 223     int tries;
 224     afs_uint32 now, timeout, code=SYNC_OK;
 225
 226     if (state->fatal_error) {
 227         return SYNC_COM_ERROR;
 228     }
 229
 230     if (state->fd == -1) {
 231         SYNC_connect(state);
 232     }
 233
 234     if (state->fd == -1) {
 235         state->fatal_error = 1;
 236         return SYNC_COM_ERROR;
 237     }
 238
 239 #ifdef AFS_DEMAND_ATTACH_FS
 240     com->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
 241 #endif
 242
 243     now = FT_ApproxTime();
 244     timeout = now + state->hard_timeout;
 245     for (tries = 0;
 246          (tries <= state->retry_limit) && (now <= timeout);
 247          tries++, now = FT_ApproxTime()) {
 248         code = SYNC_ask_internal(state, com, res);
 249         if (code == SYNC_OK) {
 250             break;
 251         } else if (code == SYNC_BAD_COMMAND) {
 252             Log("SYNC_ask: protocol mismatch; make sure fileserver, volserver, salvageserver and salvager are same version\n");
 253             break;
 254         } else if (code == SYNC_COM_ERROR) {
 255             Log("SYNC_ask: protocol communications failure; attempting reconnect to server\n");
 256             SYNC_reconnect(state);
 257             /* try again */
 258         } else {
 259             /* unknown (probably protocol-specific) response code, pass it up to the caller, and let them deal with it */
 260             break;
 261         }
 262     }
 263
 264     if (code == SYNC_COM_ERROR) {
 265         Log("SYNC_ask: fatal protocol error; disabling sync protocol to server running on port %d until next server restart\n",
 266             state->port);
 267         state->fatal_error = 1;
 268     }
 269
 270     return code;
 271 }
 272
 273 static afs_int32
 274 SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
 275 {
 276     int n;
 277     SYNC_PROTO_BUF_DECL(buf);
 278 #ifndef AFS_NT40_ENV
 279     int iovcnt;
 280     struct iovec iov[2];
 281 #endif
 282
 283     if (state->fd == -1) {
 284         Log("SYNC_ask:  invalid sync file descriptor\n");
 285         res->hdr.response = SYNC_COM_ERROR;
 286         goto done;
 287     }
 288
 289     if (com->hdr.command_len > SYNC_PROTO_MAX_LEN) {
 290         Log("SYNC_ask:  internal SYNC buffer too small; please file a bug\n");
 291         res->hdr.response = SYNC_COM_ERROR;
 292         goto done;
 293     }
 294
 295     com->hdr.proto_version = state->proto_version;
 296
 297     memcpy(buf, &com->hdr, sizeof(com->hdr));
 298     if (com->payload.len) {
 299         memcpy(buf + sizeof(com->hdr), com->payload.buf,
 300                com->hdr.command_len - sizeof(com->hdr));
 301     }
 302
 303 #ifdef AFS_NT40_ENV
 304     n = send(state->fd, buf, com->hdr.command_len, 0);
 305     if (n != com->hdr.command_len) {
 306         Log("SYNC_ask:  write failed\n");
 307         res->hdr.response = SYNC_COM_ERROR;
 308         goto done;
 309     }
 310
 311     n = recv(state->fd, buf, SYNC_PROTO_MAX_LEN, 0);
 312     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
 313         Log("SYNC_ask:  No response\n");
 314         res->hdr.response = SYNC_COM_ERROR;
 315         goto done;
 316     }
 317 #else /* !AFS_NT40_ENV */
 318     n = write(state->fd, buf, com->hdr.command_len);
 319     if (com->hdr.command_len != n) {
 320         Log("SYNC_ask: write failed\n");
 321         res->hdr.response = SYNC_COM_ERROR;
 322         goto done;
 323     }
 324
 325     /* receive the response */
 326     iov[0].iov_base = (char *)&res->hdr;
 327     iov[0].iov_len = sizeof(res->hdr);
 328     if (res->payload.len) {
 329         iov[1].iov_base = (char *)res->payload.buf;
 330         iov[1].iov_len = res->payload.len;
 331         iovcnt = 2;
 332     } else {
 333         iovcnt = 1;
 334     }
 335     n = readv(state->fd, iov, iovcnt);
 336     if (n == 0 || (n < 0 && errno != EINTR)) {
 337         Log("SYNC_ask: No response\n");
 338         res->hdr.response = SYNC_COM_ERROR;
 339         goto done;
 340     }
 341 #endif /* !AFS_NT40_ENV */
 342
 343     res->recv_len = n;
 344
 345     if (n < sizeof(res->hdr)) {
 346         Log("SYNC_ask:  response too short\n");
 347         res->hdr.response = SYNC_COM_ERROR;
 348         goto done;
 349     }
 350 #ifdef AFS_NT40_ENV
 351     memcpy(&res->hdr, buf, sizeof(res->hdr));
 352 #endif
 353
 354     if ((n - sizeof(res->hdr)) > res->payload.len) {
 355         Log("SYNC_ask:  response too long\n");
 356         res->hdr.response = SYNC_COM_ERROR;
 357         goto done;
 358     }
 359 #ifdef AFS_NT40_ENV
 360     memcpy(res->payload.buf, buf + sizeof(res->hdr), n - sizeof(res->hdr));
 361 #endif
 362
 363     if (res->hdr.response_len != n) {
 364         Log("SYNC_ask:  length field in response inconsistent\n");
 365         res->hdr.response = SYNC_COM_ERROR;
 366         goto done;
 367     }
 368     if (res->hdr.response == SYNC_DENIED) {
 369         Log("SYNC_ask: negative response\n");
 370     }
 371
 372   done:
 373     return res->hdr.response;
 374 }
 375
 376
 377 /*
 378  * daemon com SYNC server-side interfaces
 379  */
 380
 381 /* get a command */
 382 afs_int32
 383 SYNC_getCom(int fd, SYNC_command * com)
 384 {
 385     int n;
 386     afs_int32 code = SYNC_OK;
 387 #ifdef AFS_NT40_ENV
 388     SYNC_PROTO_BUF_DECL(buf);
 389 #else
 390     struct iovec iov[2];
 391     int iovcnt;
 392 #endif
 393
 394 #ifdef AFS_NT40_ENV
 395     n = recv(fd, buf, SYNC_PROTO_MAX_LEN, 0);
 396
 397     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
 398         Log("SYNC_getCom:  error receiving command\n");
 399         code = SYNC_COM_ERROR;
 400         goto done;
 401     }
 402 #else /* !AFS_NT40_ENV */
 403     iov[0].iov_base = (char *)&com->hdr;
 404     iov[0].iov_len = sizeof(com->hdr);
 405     if (com->payload.len) {
 406         iov[1].iov_base = (char *)com->payload.buf;
 407         iov[1].iov_len = com->payload.len;
 408         iovcnt = 2;
 409     } else {
 410         iovcnt = 1;
 411     }
 412
 413     n = readv(fd, iov, iovcnt);
 414     if (n == 0 || (n < 0 && errno != EINTR)) {
 415         Log("SYNC_getCom:  error receiving command\n");
 416         code = SYNC_COM_ERROR;
 417         goto done;
 418     }
 419 #endif /* !AFS_NT40_ENV */
 420
 421     com->recv_len = n;
 422
 423     if (n < sizeof(com->hdr)) {
 424         Log("SYNC_getCom:  command too short\n");
 425         code = SYNC_COM_ERROR;
 426         goto done;
 427     }
 428 #ifdef AFS_NT40_ENV
 429     memcpy(&com->hdr, buf, sizeof(com->hdr));
 430 #endif
 431
 432     if ((n - sizeof(com->hdr)) > com->payload.len) {
 433         Log("SYNC_getCom:  command too long\n");
 434         code = SYNC_COM_ERROR;
 435         goto done;
 436     }
 437 #ifdef AFS_NT40_ENV
 438     memcpy(com->payload.buf, buf + sizeof(com->hdr), n - sizeof(com->hdr));
 439 #endif
 440
 441  done:
 442     return code;
 443 }
 444
 445 /* put a response */
 446 afs_int32
 447 SYNC_putRes(int fd, SYNC_response * res)
 448 {
 449     int n;
 450     afs_int32 code = SYNC_OK;
 451     SYNC_PROTO_BUF_DECL(buf);
 452
 453     if (res->hdr.response_len > (sizeof(res->hdr) + res->payload.len)) {
 454         Log("SYNC_putRes:  response_len field in response header inconsistent\n");
 455         code = SYNC_COM_ERROR;
 456         goto done;
 457     }
 458
 459     if (res->hdr.response_len > SYNC_PROTO_MAX_LEN) {
 460         Log("SYNC_putRes:  internal SYNC buffer too small; please file a bug\n");
 461         code = SYNC_COM_ERROR;
 462         goto done;
 463     }
 464
 465 #ifdef AFS_DEMAND_ATTACH_FS
 466     res->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
 467 #endif
 468
 469     memcpy(buf, &res->hdr, sizeof(res->hdr));
 470     if (res->payload.len) {
 471         memcpy(buf + sizeof(res->hdr), res->payload.buf,
 472                res->hdr.response_len - sizeof(res->hdr));
 473     }
 474
 475 #ifdef AFS_NT40_ENV
 476     n = send(fd, buf, res->hdr.response_len, 0);
 477 #else /* !AFS_NT40_ENV */
 478     n = write(fd, buf, res->hdr.response_len);
 479 #endif /* !AFS_NT40_ENV */
 480
 481     if (res->hdr.response_len != n) {
 482         Log("SYNC_putRes: write failed\n");
 483         res->hdr.response = SYNC_COM_ERROR;
 484         goto done;
 485     }
 486
 487  done:
 488     return code;
 489 }
 490
 491 /* return 0 for legal (null-terminated) string,
 492  * 1 for illegal (unterminated) string */
 493 int
 494 SYNC_verifyProtocolString(char * buf, size_t len)
 495 {
 496     int ret = 0;
 497     size_t s_len;
 498
 499     s_len = afs_strnlen(buf, len);
 500
 501     return (s_len == len) ? 1 : 0;
 502 }