src/vol/daemon_com.c

   1 /*
   2  * Copyright 2006-2008, 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
  25 #include <sys/types.h>
  26 #include <stdio.h>
  27 #ifdef AFS_NT40_ENV
  28 #include <winsock2.h>
  29 #include <time.h>
  30 #else
  31 #include <sys/param.h>
  32 #include <sys/socket.h>
  33 #include <netinet/in.h>
  34 #include <netdb.h>
  35 #include <sys/time.h>
  36 #include <unistd.h>
  37 #endif
  38 #include <errno.h>
  39 #include <afs/afs_assert.h>
  40 #include <signal.h>
  41 #include <string.h>
  42
  43
  44 #include <rx/xdr.h>
  45 #include <afs/afsint.h>
  46 #include "nfs.h"
  47 #include <afs/errors.h>
  48 #include "daemon_com.h"
  49 #include "lwp.h"
  50 #include "lock.h"
  51 #include <afs/afssyscalls.h>
  52 #include "ihandle.h"
  53 #include "vnode.h"
  54 #include "volume.h"
  55 #include "partition.h"
  56 #include "common.h"
  57 #include <rx/rx_queue.h>
  58
  59 #ifdef USE_UNIX_SOCKETS
  60 #include <afs/afsutil.h>
  61 #include <sys/un.h>
  62 #endif
  63
  64 int (*V_BreakVolumeCallbacks) (VolumeId);
  65
  66 #define MAXHANDLERS     4       /* Up to 4 clients; must be at least 2, so that
  67                                  * move = dump+restore can run on single server */
  68
  69 #define MAX_BIND_TRIES  5       /* Number of times to retry socket bind */
  70
  71 static int SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res);
  72
  73
  74 /*
  75  * On AIX, connect() and bind() require use of SUN_LEN() macro;
  76  * sizeof(struct sockaddr_un) will not suffice.
  77  */
  78 #if defined(AFS_AIX_ENV) && defined(USE_UNIX_SOCKETS)
  79 #define AFS_SOCKADDR_LEN(sa)  SUN_LEN(sa)
  80 #else
  81 #define AFS_SOCKADDR_LEN(sa)  sizeof(*sa)
  82 #endif
  83
  84
  85 /* daemon com SYNC general interfaces */
  86
  87 /**
  88  * fill in sockaddr structure.
  89  *
  90  * @param[in]  endpoint pointer to sync endpoint object
  91  * @param[out] addr     pointer to sockaddr structure
  92  *
  93  * @post sockaddr structure populated using information from
  94  *       endpoint structure.
  95  */
  96 void
  97 SYNC_getAddr(SYNC_endpoint_t * endpoint, SYNC_sockaddr_t * addr)
  98 {
  99 #ifdef USE_UNIX_SOCKETS
 100     char tbuffer[AFSDIR_PATH_MAX];
 101 #endif /* USE_UNIX_SOCKETS */
 102
 103     memset(addr, 0, sizeof(*addr));
 104
 105 #ifdef USE_UNIX_SOCKETS
 106     strcompose(tbuffer, AFSDIR_PATH_MAX, AFSDIR_SERVER_LOCAL_DIRPATH, "/",
 107                endpoint->un, NULL);
 108     addr->sun_family = AF_UNIX;
 109     strncpy(addr->sun_path, tbuffer, (sizeof(struct sockaddr_un) - sizeof(short)));
 110 #else  /* !USE_UNIX_SOCKETS */
 111 #ifdef STRUCT_SOCKADDR_HAS_SA_LEN
 112     addr->sin_len = sizeof(struct sockaddr_in);
 113 #endif
 114     addr->sin_addr.s_addr = htonl(0x7f000001);
 115     addr->sin_family = AF_INET; /* was localhost->h_addrtype */
 116     addr->sin_port = htons(endpoint->in);       /* XXXX htons not _really_ neccessary */
 117 #endif /* !USE_UNIX_SOCKETS */
 118 }
 119
 120 /**
 121  * get a socket descriptor of the appropriate domain.
 122  *
 123  * @param[in]  endpoint pointer to sync endpoint object
 124  *
 125  * @return socket descriptor
 126  *
 127  * @post socket of domain specified in endpoint structure is created and
 128  *       returned to caller.
 129  */
 130 osi_socket
 131 SYNC_getSock(SYNC_endpoint_t * endpoint)
 132 {
 133     osi_socket sd;
 134     osi_Assert((sd = socket(endpoint->domain, SOCK_STREAM, 0)) >= 0);
 135     return sd;
 136 }
 137
 138 /* daemon com SYNC client interface */
 139
 140 /**
 141  * open a client connection to a sync server
 142  *
 143  * @param[in] state  pointer to sync client handle
 144  *
 145  * @return operation status
 146  *    @retval 1 success
 147  *
 148  * @note at present, this routine aborts rather than returning an error code
 149  */
 150 int
 151 SYNC_connect(SYNC_client_state * state)
 152 {
 153     SYNC_sockaddr_t addr;
 154     /* I can't believe the following is needed for localhost connections!! */
 155     static time_t backoff[] =
 156         { 3, 3, 3, 5, 5, 5, 7, 15, 16, 24, 32, 40, 48, 0 };
 157     time_t *timeout = &backoff[0];
 158
 159     if (state->fd != OSI_NULLSOCKET) {
 160         return 1;
 161     }
 162
 163     SYNC_getAddr(&state->endpoint, &addr);
 164
 165     for (;;) {
 166         state->fd = SYNC_getSock(&state->endpoint);
 167         if (connect(state->fd, (struct sockaddr *)&addr, AFS_SOCKADDR_LEN(&addr)) >= 0)
 168             return 1;
 169         if (!*timeout)
 170             break;
 171         if (!(*timeout & 1))
 172             Log("SYNC_connect: temporary failure on circuit '%s' (will retry)\n",
 173                 state->proto_name);
 174         SYNC_disconnect(state);
 175         sleep(*timeout++);
 176     }
 177     perror("SYNC_connect failed (giving up!)");
 178     return 0;
 179 }
 180
 181 /**
 182  * forcibly disconnect a sync client handle.
 183  *
 184  * @param[in] state  pointer to sync client handle
 185  *
 186  * @retval operation status
 187  *    @retval 0 success
 188  */
 189 int
 190 SYNC_disconnect(SYNC_client_state * state)
 191 {
 192 #ifdef AFS_NT40_ENV
 193     closesocket(state->fd);
 194 #else
 195     close(state->fd);
 196 #endif
 197     state->fd = OSI_NULLSOCKET;
 198     return 0;
 199 }
 200
 201 /**
 202  * gracefully disconnect a sync client handle.
 203  *
 204  * @param[in] state  pointer to sync client handle
 205  *
 206  * @return operation status
 207  *    @retval SYNC_OK success
 208  */
 209 afs_int32
 210 SYNC_closeChannel(SYNC_client_state * state)
 211 {
 212     SYNC_command com;
 213     SYNC_response res;
 214     SYNC_PROTO_BUF_DECL(ores);
 215
 216     if (state->fd == OSI_NULLSOCKET)
 217         return SYNC_OK;
 218
 219     memset(&com, 0, sizeof(com));
 220     memset(&res, 0, sizeof(res));
 221
 222     res.payload.len = SYNC_PROTO_MAX_LEN;
 223     res.payload.buf = ores;
 224
 225     com.hdr.command = SYNC_COM_CHANNEL_CLOSE;
 226     com.hdr.command_len = sizeof(SYNC_command_hdr);
 227     com.hdr.flags |= SYNC_FLAG_CHANNEL_SHUTDOWN;
 228
 229     /* in case the other end dropped, don't do any retries */
 230     state->retry_limit = 0;
 231     state->hard_timeout = 0;
 232
 233     SYNC_ask(state, &com, &res);
 234     SYNC_disconnect(state);
 235
 236     return SYNC_OK;
 237 }
 238
 239 /**
 240  * forcibly break a client connection, and then create a new connection.
 241  *
 242  * @param[in] state  pointer to sync client handle
 243  *
 244  * @post old connection dropped; new connection established
 245  *
 246  * @return @see SYNC_connect()
 247  */
 248 int
 249 SYNC_reconnect(SYNC_client_state * state)
 250 {
 251     SYNC_disconnect(state);
 252     return SYNC_connect(state);
 253 }
 254
 255 /**
 256  * send a command to a sync server and wait for a response.
 257  *
 258  * @param[in]  state  pointer to sync client handle
 259  * @param[in]  com    command object
 260  * @param[out] res    response object
 261  *
 262  * @return operation status
 263  *    @retval SYNC_OK success
 264  *    @retval SYNC_COM_ERROR communications error
 265  *    @retval SYNC_BAD_COMMAND server did not recognize command code
 266  *
 267  * @note this routine merely handles error processing; SYNC_ask_internal()
 268  *       handles the low-level details of communicating with the SYNC server.
 269  *
 270  * @see SYNC_ask_internal
 271  */
 272 afs_int32
 273 SYNC_ask(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
 274 {
 275     int tries;
 276     afs_uint32 now, timeout, code=SYNC_OK;
 277
 278     if (state->fd == OSI_NULLSOCKET) {
 279         SYNC_connect(state);
 280     }
 281
 282     if (state->fd == OSI_NULLSOCKET) {
 283         return SYNC_COM_ERROR;
 284     }
 285
 286 #ifdef AFS_DEMAND_ATTACH_FS
 287     com->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
 288 #endif
 289
 290     now = FT_ApproxTime();
 291     timeout = now + state->hard_timeout;
 292     for (tries = 0;
 293          (tries <= state->retry_limit) && (now <= timeout);
 294          tries++, now = FT_ApproxTime()) {
 295         code = SYNC_ask_internal(state, com, res);
 296         if (code == SYNC_OK) {
 297             break;
 298         } else if (code == SYNC_BAD_COMMAND) {
 299             Log("SYNC_ask: protocol mismatch on circuit '%s'; make sure "
 300                 "fileserver, volserver, salvageserver and salvager are same "
 301                 "version\n", state->proto_name);
 302             break;
 303         } else if ((code == SYNC_COM_ERROR) && (tries < state->retry_limit)) {
 304             Log("SYNC_ask: protocol communications failure on circuit '%s'; "
 305                 "attempting reconnect to server\n", state->proto_name);
 306             SYNC_reconnect(state);
 307             /* try again */
 308         } else {
 309             /*
 310              * unknown (probably protocol-specific) response code, pass it up to
 311              * the caller, and let them deal with it
 312              */
 313             break;
 314         }
 315     }
 316
 317     if (code == SYNC_COM_ERROR) {
 318         Log("SYNC_ask: too many / too latent fatal protocol errors on circuit "
 319             "'%s'; giving up (tries %d timeout %d)\n",
 320             state->proto_name, tries, timeout);
 321     }
 322
 323     return code;
 324 }
 325
 326 /**
 327  * send a command to a sync server and wait for a response.
 328  *
 329  * @param[in]  state  pointer to sync client handle
 330  * @param[in]  com    command object
 331  * @param[out] res    response object
 332  *
 333  * @return operation status
 334  *    @retval SYNC_OK success
 335  *    @retval SYNC_COM_ERROR communications error
 336  *
 337  * @internal
 338  */
 339 static afs_int32
 340 SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
 341 {
 342     int n;
 343     SYNC_PROTO_BUF_DECL(buf);
 344 #ifndef AFS_NT40_ENV
 345     int iovcnt;
 346     struct iovec iov[2];
 347 #endif
 348
 349     if (state->fd == OSI_NULLSOCKET) {
 350         Log("SYNC_ask:  invalid sync file descriptor on circuit '%s'\n",
 351             state->proto_name);
 352         res->hdr.response = SYNC_COM_ERROR;
 353         goto done;
 354     }
 355
 356     if (com->hdr.command_len > SYNC_PROTO_MAX_LEN) {
 357         Log("SYNC_ask:  internal SYNC buffer too small on circuit '%s'; "
 358             "please file a bug\n", state->proto_name);
 359         res->hdr.response = SYNC_COM_ERROR;
 360         goto done;
 361     }
 362
 363     /*
 364      * fill in some common header fields
 365      */
 366     com->hdr.proto_version = state->proto_version;
 367     com->hdr.pkt_seq = ++state->pkt_seq;
 368     com->hdr.com_seq = ++state->com_seq;
 369 #ifdef AFS_NT40_ENV
 370     com->hdr.pid = 0;
 371     com->hdr.tid = 0;
 372 #else
 373     com->hdr.pid = getpid();
 374 #ifdef AFS_PTHREAD_ENV
 375     com->hdr.tid = afs_pointer_to_int(pthread_self());
 376 #else
 377     {
 378         PROCESS handle = LWP_ThreadId();
 379         com->hdr.tid = (handle) ? handle->index : 0;
 380     }
 381 #endif /* !AFS_PTHREAD_ENV */
 382 #endif /* !AFS_NT40_ENV */
 383
 384     memcpy(buf, &com->hdr, sizeof(com->hdr));
 385     if (com->payload.len) {
 386         memcpy(buf + sizeof(com->hdr), com->payload.buf,
 387                com->hdr.command_len - sizeof(com->hdr));
 388     }
 389
 390 #ifdef AFS_NT40_ENV
 391     n = send(state->fd, buf, com->hdr.command_len, 0);
 392     if (n != com->hdr.command_len) {
 393         Log("SYNC_ask:  write failed on circuit '%s'\n", state->proto_name);
 394         res->hdr.response = SYNC_COM_ERROR;
 395         goto done;
 396     }
 397
 398     if (com->hdr.command == SYNC_COM_CHANNEL_CLOSE) {
 399         /* short circuit close channel requests */
 400         res->hdr.response = SYNC_OK;
 401         goto done;
 402     }
 403
 404     n = recv(state->fd, buf, SYNC_PROTO_MAX_LEN, 0);
 405     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
 406         Log("SYNC_ask:  No response on circuit '%s'\n", state->proto_name);
 407         res->hdr.response = SYNC_COM_ERROR;
 408         goto done;
 409     }
 410 #else /* !AFS_NT40_ENV */
 411     n = write(state->fd, buf, com->hdr.command_len);
 412     if (com->hdr.command_len != n) {
 413         Log("SYNC_ask: write failed on circuit '%s'\n", state->proto_name);
 414         res->hdr.response = SYNC_COM_ERROR;
 415         goto done;
 416     }
 417
 418     if (com->hdr.command == SYNC_COM_CHANNEL_CLOSE) {
 419         /* short circuit close channel requests */
 420         res->hdr.response = SYNC_OK;
 421         goto done;
 422     }
 423
 424     /* receive the response */
 425     iov[0].iov_base = (char *)&res->hdr;
 426     iov[0].iov_len = sizeof(res->hdr);
 427     if (res->payload.len) {
 428         iov[1].iov_base = (char *)res->payload.buf;
 429         iov[1].iov_len = res->payload.len;
 430         iovcnt = 2;
 431     } else {
 432         iovcnt = 1;
 433     }
 434     n = readv(state->fd, iov, iovcnt);
 435     if (n == 0 || (n < 0 && errno != EINTR)) {
 436         Log("SYNC_ask: No response on circuit '%s'\n", state->proto_name);
 437         res->hdr.response = SYNC_COM_ERROR;
 438         goto done;
 439     }
 440 #endif /* !AFS_NT40_ENV */
 441
 442     res->recv_len = n;
 443
 444     if (n < sizeof(res->hdr)) {
 445         Log("SYNC_ask:  response too short on circuit '%s'\n",
 446             state->proto_name);
 447         res->hdr.response = SYNC_COM_ERROR;
 448         goto done;
 449     }
 450 #ifdef AFS_NT40_ENV
 451     memcpy(&res->hdr, buf, sizeof(res->hdr));
 452 #endif
 453
 454     if ((n - sizeof(res->hdr)) > res->payload.len) {
 455         Log("SYNC_ask:  response too long on circuit '%s'\n",
 456             state->proto_name);
 457         res->hdr.response = SYNC_COM_ERROR;
 458         goto done;
 459     }
 460 #ifdef AFS_NT40_ENV
 461     memcpy(res->payload.buf, buf + sizeof(res->hdr), n - sizeof(res->hdr));
 462 #endif
 463
 464     if (res->hdr.response_len != n) {
 465         Log("SYNC_ask:  length field in response inconsistent "
 466             "on circuit '%s'\n", state->proto_name);
 467         res->hdr.response = SYNC_COM_ERROR;
 468         goto done;
 469     }
 470     if (res->hdr.response == SYNC_DENIED) {
 471         Log("SYNC_ask: negative response on circuit '%s'\n", state->proto_name);
 472     }
 473
 474   done:
 475     return res->hdr.response;
 476 }
 477
 478
 479 /*
 480  * daemon com SYNC server-side interfaces
 481  */
 482
 483 /**
 484  * receive a command structure off a sync socket.
 485  *
 486  * @param[in]  state  pointer to server-side state object
 487  * @param[in]  fd     file descriptor on which to perform i/o
 488  * @param[out] com    sync command object to be populated
 489  *
 490  * @return operation status
 491  *    @retval SYNC_OK command received
 492  *    @retval SYNC_COM_ERROR there was a socket communications error
 493  */
 494 afs_int32
 495 SYNC_getCom(SYNC_server_state_t * state,
 496             osi_socket fd,
 497             SYNC_command * com)
 498 {
 499     int n;
 500     afs_int32 code = SYNC_OK;
 501 #ifdef AFS_NT40_ENV
 502     SYNC_PROTO_BUF_DECL(buf);
 503 #else
 504     struct iovec iov[2];
 505     int iovcnt;
 506 #endif
 507
 508 #ifdef AFS_NT40_ENV
 509     n = recv(fd, buf, SYNC_PROTO_MAX_LEN, 0);
 510
 511     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
 512         Log("SYNC_getCom:  error receiving command\n");
 513         code = SYNC_COM_ERROR;
 514         goto done;
 515     }
 516 #else /* !AFS_NT40_ENV */
 517     iov[0].iov_base = (char *)&com->hdr;
 518     iov[0].iov_len = sizeof(com->hdr);
 519     if (com->payload.len) {
 520         iov[1].iov_base = (char *)com->payload.buf;
 521         iov[1].iov_len = com->payload.len;
 522         iovcnt = 2;
 523     } else {
 524         iovcnt = 1;
 525     }
 526
 527     n = readv(fd, iov, iovcnt);
 528     if (n == 0 || (n < 0 && errno != EINTR)) {
 529         Log("SYNC_getCom:  error receiving command\n");
 530         code = SYNC_COM_ERROR;
 531         goto done;
 532     }
 533 #endif /* !AFS_NT40_ENV */
 534
 535     com->recv_len = n;
 536
 537     if (n < sizeof(com->hdr)) {
 538         Log("SYNC_getCom:  command too short\n");
 539         code = SYNC_COM_ERROR;
 540         goto done;
 541     }
 542 #ifdef AFS_NT40_ENV
 543     memcpy(&com->hdr, buf, sizeof(com->hdr));
 544 #endif
 545
 546     if ((n - sizeof(com->hdr)) > com->payload.len) {
 547         Log("SYNC_getCom:  command too long\n");
 548         code = SYNC_COM_ERROR;
 549         goto done;
 550     }
 551 #ifdef AFS_NT40_ENV
 552     memcpy(com->payload.buf, buf + sizeof(com->hdr), n - sizeof(com->hdr));
 553 #endif
 554
 555  done:
 556     return code;
 557 }
 558
 559 /**
 560  * write a response structure to a sync socket.
 561  *
 562  * @param[in] state  handle to server-side state object
 563  * @param[in] fd     file descriptor on which to perform i/o
 564  * @param[in] res    handle to response packet
 565  *
 566  * @return operation status
 567  *    @retval SYNC_OK
 568  *    @retval SYNC_COM_ERROR
 569  */
 570 afs_int32
 571 SYNC_putRes(SYNC_server_state_t * state,
 572             osi_socket fd,
 573             SYNC_response * res)
 574 {
 575     int n;
 576     afs_int32 code = SYNC_OK;
 577     SYNC_PROTO_BUF_DECL(buf);
 578
 579     if (res->hdr.response_len > (sizeof(res->hdr) + res->payload.len)) {
 580         Log("SYNC_putRes:  response_len field in response header inconsistent\n");
 581         code = SYNC_COM_ERROR;
 582         goto done;
 583     }
 584
 585     if (res->hdr.response_len > SYNC_PROTO_MAX_LEN) {
 586         Log("SYNC_putRes:  internal SYNC buffer too small; please file a bug\n");
 587         code = SYNC_COM_ERROR;
 588         goto done;
 589     }
 590
 591 #ifdef AFS_DEMAND_ATTACH_FS
 592     res->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
 593 #endif
 594     res->hdr.proto_version = state->proto_version;
 595     res->hdr.pkt_seq = ++state->pkt_seq;
 596     res->hdr.res_seq = ++state->res_seq;
 597
 598     memcpy(buf, &res->hdr, sizeof(res->hdr));
 599     if (res->payload.len) {
 600         memcpy(buf + sizeof(res->hdr), res->payload.buf,
 601                res->hdr.response_len - sizeof(res->hdr));
 602     }
 603
 604 #ifdef AFS_NT40_ENV
 605     n = send(fd, buf, res->hdr.response_len, 0);
 606 #else /* !AFS_NT40_ENV */
 607     n = write(fd, buf, res->hdr.response_len);
 608 #endif /* !AFS_NT40_ENV */
 609
 610     if (res->hdr.response_len != n) {
 611         Log("SYNC_putRes: write failed\n");
 612         res->hdr.response = SYNC_COM_ERROR;
 613         goto done;
 614     }
 615
 616  done:
 617     return code;
 618 }
 619
 620 /* return 0 for legal (null-terminated) string,
 621  * 1 for illegal (unterminated) string */
 622 int
 623 SYNC_verifyProtocolString(char * buf, size_t len)
 624 {
 625     size_t s_len;
 626
 627     s_len = afs_strnlen(buf, len);
 628
 629     return (s_len == len) ? 1 : 0;
 630 }
 631
 632 /**
 633  * clean up old sockets.
 634  *
 635  * @param[in]  state  server state object
 636  *
 637  * @post unix domain sockets are cleaned up
 638  */
 639 void
 640 SYNC_cleanupSock(SYNC_server_state_t * state)
 641 {
 642 #ifdef USE_UNIX_SOCKETS
 643     remove(state->addr.sun_path);
 644 #endif
 645 }
 646
 647 /**
 648  * bind socket and set it to listen state.
 649  *
 650  * @param[in] state  server state object
 651  *
 652  * @return operation status
 653  *    @retval 0 success
 654  *    @retval nonzero failure
 655  *
 656  * @post socket bound and set to listen state
 657  */
 658 int
 659 SYNC_bindSock(SYNC_server_state_t * state)
 660 {
 661     int code;
 662     int on = 1;
 663     int numTries;
 664
 665     /* Reuseaddr needed because system inexplicably leaves crud lying around */
 666     code =
 667         setsockopt(state->fd, SOL_SOCKET, SO_REUSEADDR, (char *)&on,
 668                    sizeof(on));
 669     if (code)
 670         Log("SYNC_bindSock: setsockopt failed with (%d)\n", errno);
 671
 672     for (numTries = 0; numTries < state->bind_retry_limit; numTries++) {
 673         code = bind(state->fd,
 674                     (struct sockaddr *)&state->addr,
 675                     AFS_SOCKADDR_LEN(&state->addr));
 676         if (code == 0)
 677             break;
 678         Log("SYNC_bindSock: bind failed with (%d), will sleep and retry\n",
 679             errno);
 680         sleep(5);
 681     }
 682     listen(state->fd, state->listen_depth);
 683
 684     return code;
 685 }