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