src/ubik/remote.c

   1 /*
   2  * Copyright 2000, International Business Machines Corporation 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 #include <afs/param.h>
  11 #include <sys/types.h>
  12 #ifdef AFS_NT40_ENV
  13 #include <winsock2.h>
  14 #else
  15 #include <sys/file.h>
  16 #include <netinet/in.h>
  17 #endif
  18 #include <lock.h>
  19 #include <rx/xdr.h>
  20 #include <rx/rx.h>
  21
  22 #define UBIK_INTERNALS
  23 #include "ubik.h"
  24 #include "ubik_int.h"
  25 int (*ubik_CheckRXSecurityProc)();
  26 char *ubik_CheckRXSecurityRock;
  27
  28 /* routines for handling requests remotely-submitted by the sync site.  These are
  29     only write transactions (we don't propagate read trans), and there is at most one
  30     write transaction extant at any one time.
  31 */
  32
  33 struct ubik_trans *ubik_currentTrans = 0;
  34
  35
  36 ubik_CheckAuth(acall)
  37 register struct rx_call *acall; {
  38     register afs_int32 code;
  39     if (ubik_CheckRXSecurityProc) {
  40         code = (*ubik_CheckRXSecurityProc)(ubik_CheckRXSecurityRock, acall);
  41         return code;
  42     }
  43     else return 0;
  44 }
  45
  46 /* the rest of these guys handle remote execution of write
  47  * transactions: this is the code executed on the other servers when a
  48  * sync site is executing a write transaction.
  49  */
  50 SDISK_Begin(rxcall, atid)
  51     register struct rx_call *rxcall;
  52     struct ubik_tid *atid;
  53 {
  54     register afs_int32 code;
  55
  56     if (code = ubik_CheckAuth(rxcall)) {
  57       return code;
  58     }
  59     DBHOLD(ubik_dbase);
  60     urecovery_CheckTid(atid);
  61     if (ubik_currentTrans) {
  62         /* If the thread is not waiting for lock - ok to end it */
  63         if (ubik_currentTrans->locktype != LOCKWAIT) {
  64            udisk_end(ubik_currentTrans);
  65         }
  66         ubik_currentTrans = (struct ubik_trans *) 0;
  67     }
  68     code = udisk_begin(ubik_dbase, UBIK_WRITETRANS, &ubik_currentTrans);
  69     if (!code && ubik_currentTrans) {
  70         /* label this trans with the right trans id */
  71         ubik_currentTrans->tid.epoch = atid->epoch;
  72         ubik_currentTrans->tid.counter = atid->counter;
  73     }
  74     DBRELE(ubik_dbase);
  75     return code;}
  76
  77
  78 SDISK_Commit(rxcall, atid)
  79     register struct rx_call *rxcall;
  80     struct ubik_tid *atid;
  81 {
  82     register afs_int32 code;
  83     register struct ubik_dbase *dbase;
  84
  85     if (code = ubik_CheckAuth(rxcall)) {
  86       return code;
  87     }
  88
  89     if (!ubik_currentTrans) {
  90       return USYNC;
  91     }
  92     /*
  93      * sanity check to make sure only write trans appear here
  94      */
  95     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
  96       return UBADTYPE;
  97     }
  98
  99     dbase = ubik_currentTrans->dbase;
 100     DBHOLD(dbase);
 101     urecovery_CheckTid(atid);
 102     if (!ubik_currentTrans) {
 103         DBRELE(dbase);
 104         return USYNC;
 105     }
 106
 107     code = udisk_commit(ubik_currentTrans);
 108     if (code == 0) {
 109         /* sync site should now match */
 110         ubik_dbVersion = ubik_dbase->version;
 111     }
 112     DBRELE(dbase);
 113     return code;
 114 }
 115
 116 SDISK_ReleaseLocks(rxcall, atid)
 117     register struct rx_call *rxcall;
 118     struct ubik_tid *atid;
 119 {
 120     register struct ubik_dbase *dbase;
 121     register afs_int32 code;
 122
 123     if (code = ubik_CheckAuth(rxcall)) {
 124       return code;
 125     }
 126
 127     if (!ubik_currentTrans) {
 128       return USYNC;
 129     }
 130     /* sanity check to make sure only write trans appear here */
 131     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 132       return UBADTYPE;
 133     }
 134
 135     dbase = ubik_currentTrans->dbase;
 136     DBHOLD(dbase);
 137     urecovery_CheckTid(atid);
 138     if (!ubik_currentTrans) {
 139         DBRELE(dbase);
 140         return USYNC;
 141     }
 142
 143     /* If the thread is not waiting for lock - ok to end it */
 144     if (ubik_currentTrans->locktype != LOCKWAIT) {
 145        udisk_end(ubik_currentTrans);
 146     }
 147     ubik_currentTrans = (struct ubik_trans *) 0;
 148     DBRELE(dbase);
 149     return 0;
 150 }
 151
 152 SDISK_Abort(rxcall, atid)
 153     register struct rx_call *rxcall;
 154     struct ubik_tid *atid;
 155 {
 156     register afs_int32 code;
 157     register struct ubik_dbase *dbase;
 158
 159     if (code = ubik_CheckAuth(rxcall)) {
 160       return code;
 161     }
 162
 163     if (!ubik_currentTrans) {
 164       return USYNC;
 165     }
 166     /* sanity check to make sure only write trans appear here  */
 167     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 168       return UBADTYPE;
 169     }
 170
 171     dbase = ubik_currentTrans->dbase;
 172     DBHOLD(dbase);
 173     urecovery_CheckTid(atid);
 174     if (!ubik_currentTrans) {
 175         DBRELE(dbase);
 176         return USYNC;
 177     }
 178
 179     code = udisk_abort(ubik_currentTrans);
 180     /* If the thread is not waiting for lock - ok to end it */
 181     if (ubik_currentTrans->locktype != LOCKWAIT) {
 182        udisk_end(ubik_currentTrans);
 183     }
 184     ubik_currentTrans = (struct ubik_trans *) 0;
 185     DBRELE(dbase);
 186     return code;
 187 }
 188
 189 SDISK_Lock(rxcall, atid, afile, apos, alen, atype)
 190     register struct rx_call *rxcall;
 191     struct ubik_tid *atid;
 192     afs_int32 afile, apos, alen, atype;   /* apos and alen are not used */
 193 {
 194     register afs_int32 code;
 195     register struct ubik_dbase *dbase;
 196     struct ubik_trans *ubik_thisTrans;
 197
 198     if (code = ubik_CheckAuth(rxcall)) {
 199       return code;
 200     }
 201     if (!ubik_currentTrans) {
 202       return USYNC;
 203     }
 204     /* sanity check to make sure only write trans appear here */
 205     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 206       return UBADTYPE;
 207     }
 208     if (alen != 1) {
 209       return UBADLOCK;
 210     }
 211     dbase = ubik_currentTrans->dbase;
 212     DBHOLD(dbase);
 213     urecovery_CheckTid(atid);
 214     if (!ubik_currentTrans) {
 215         DBRELE(dbase);
 216         return USYNC;
 217     }
 218
 219     ubik_thisTrans = ubik_currentTrans;
 220     code = ulock_getLock(ubik_currentTrans, atype, 1);
 221
 222     /* While waiting, the transaction may have been ended/
 223      * aborted from under us (urecovery_CheckTid). In that
 224      * case, end the transaction here.
 225      */
 226     if (!code && (ubik_currentTrans != ubik_thisTrans)) {
 227        udisk_end(ubik_thisTrans);
 228        code = USYNC;
 229     }
 230
 231     DBRELE(dbase);
 232     return code;
 233 }
 234
 235 /* Write a vector of data */
 236 SDISK_WriteV(rxcall, atid, io_vector, io_buffer)
 237     register struct rx_call *rxcall;
 238     struct ubik_tid *atid;
 239     iovec_wrt *io_vector;
 240     iovec_buf *io_buffer;
 241 {
 242     afs_int32 code, i, offset;
 243     struct ubik_dbase *dbase;
 244     struct ubik_iovec *iovec;
 245     char              *iobuf;
 246
 247     if (code = ubik_CheckAuth(rxcall)) {
 248       return code;
 249     }
 250     if (!ubik_currentTrans) {
 251       return USYNC;
 252     }
 253     /* sanity check to make sure only write trans appear here */
 254     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 255       return UBADTYPE;
 256     }
 257
 258     dbase = ubik_currentTrans->dbase;
 259     DBHOLD(dbase);
 260     urecovery_CheckTid(atid);
 261     if (!ubik_currentTrans) {
 262         DBRELE(dbase);
 263         return USYNC;
 264     }
 265
 266     iovec = (struct ubik_iovec *)io_vector->iovec_wrt_val;
 267     iobuf = (char *)io_buffer->iovec_buf_val;
 268     for (i=0, offset=0; i<io_vector->iovec_wrt_len; i++) {
 269        /* Sanity check for going off end of buffer */
 270        if ((offset + iovec[i].length) > io_buffer->iovec_buf_len) {
 271           code = UINTERNAL;
 272        } else {
 273           code = udisk_write(ubik_currentTrans, iovec[i].file,     &iobuf[offset],
 274                                                 iovec[i].position, iovec[i].length);
 275        }
 276        if (code) break;
 277
 278        offset += iovec[i].length;
 279     }
 280
 281     DBRELE(dbase);
 282     return code;
 283 }
 284
 285 SDISK_Write(rxcall, atid, afile, apos, adata)
 286     register struct rx_call *rxcall;
 287     struct ubik_tid *atid;
 288     afs_int32 afile, apos;
 289     register bulkdata *adata;
 290 {
 291     register afs_int32 code;
 292     register struct ubik_dbase *dbase;
 293
 294     if (code = ubik_CheckAuth(rxcall)) {
 295       return code;
 296     }
 297     if (!ubik_currentTrans) {
 298       return USYNC;
 299     }
 300     /* sanity check to make sure only write trans appear here */
 301     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 302       return UBADTYPE;
 303     }
 304
 305     dbase = ubik_currentTrans->dbase;
 306     DBHOLD(dbase);
 307     urecovery_CheckTid(atid);
 308     if (!ubik_currentTrans) {
 309         DBRELE(dbase);
 310         return USYNC;
 311     }
 312     code = udisk_write(ubik_currentTrans, afile, adata->bulkdata_val, apos, adata->bulkdata_len);
 313     DBRELE(dbase);
 314     return code;
 315 }
 316
 317 SDISK_Truncate(rxcall, atid, afile, alen)
 318     register struct rx_call *rxcall;
 319     struct ubik_tid *atid;
 320     afs_int32 afile;
 321     afs_int32 alen;
 322 {
 323     register afs_int32 code;
 324     register struct ubik_dbase *dbase;
 325
 326     if (code = ubik_CheckAuth(rxcall)) {
 327       return code;
 328     }
 329     if (!ubik_currentTrans) {
 330       return USYNC;
 331     }
 332     /* sanity check to make sure only write trans appear here */
 333     if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 334       return UBADTYPE;
 335     }
 336
 337     dbase = ubik_currentTrans->dbase;
 338     DBHOLD(dbase);
 339     urecovery_CheckTid(atid);
 340     if (!ubik_currentTrans) {
 341         DBRELE(dbase);
 342         return USYNC;
 343     }
 344     code = udisk_truncate(ubik_currentTrans, afile, alen);
 345     DBRELE(dbase);
 346     return code;
 347 }
 348
 349 SDISK_GetVersion(rxcall, aversion)
 350     register struct rx_call *rxcall;
 351     register struct ubik_version *aversion;
 352 {
 353     register afs_int32 code;
 354
 355     if (code = ubik_CheckAuth(rxcall)) {
 356       return code;
 357     }
 358
 359     /*
 360      * If we are the sync site, recovery shouldn't be running on any
 361      * other site. We shouldn't be getting this RPC as long as we are
 362      * the sync site.  To prevent any unforseen activity, we should
 363      * reject this RPC until we have recognized that we are not the
 364      * sync site anymore, and/or if we have any pending WRITE
 365      * transactions that have to complete. This way we can be assured
 366      * that this RPC would not block any pending transactions that
 367      * should either fail or pass. If we have recognized the fact that
 368      * we are not the sync site any more, all write transactions would
 369      * fail with UNOQUORUM anyway.
 370      */
 371     if (ubeacon_AmSyncSite()) {
 372       return UDEADLOCK;
 373     }
 374
 375     DBHOLD(ubik_dbase);
 376     code = (*ubik_dbase->getlabel) (ubik_dbase, 0, aversion);
 377     DBRELE(ubik_dbase);
 378     if (code) {
 379         /* tell other side there's no dbase */
 380         aversion->epoch = 0;
 381         aversion->counter = 0;
 382     }
 383     return 0;
 384 }
 385
 386 SDISK_GetFile(rxcall, file, version)
 387     register struct rx_call *rxcall;
 388     register afs_int32 file;
 389     struct ubik_version *version;
 390 {
 391     register afs_int32 code;
 392     register struct ubik_dbase *dbase;
 393     register afs_int32 offset;
 394     struct ubik_stat ubikstat;
 395     char tbuffer[256];
 396     afs_int32 tlen;
 397     afs_int32 length;
 398
 399     if (code = ubik_CheckAuth(rxcall)) {
 400       return code;
 401     }
 402 /* temporarily disabled because it causes problems for migration tool.  Hey, it's just
 403  * a sanity check, anyway.
 404     if (ubeacon_AmSyncSite()) {
 405       return UDEADLOCK;
 406     }
 407 */
 408     dbase = ubik_dbase;
 409     DBHOLD(dbase);
 410     code = (*dbase->stat) (dbase, file, &ubikstat);
 411     if (code < 0) {
 412         DBRELE(dbase);
 413         return code;
 414     }
 415     length = ubikstat.size;
 416     tlen = htonl(length);
 417     code = rx_Write(rxcall, &tlen, sizeof(afs_int32));
 418     if (code != sizeof(afs_int32)) {
 419         DBRELE(dbase);
 420         return BULK_ERROR;
 421     }
 422     offset = 0;
 423     while (length > 0) {
 424         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 425         code = (*dbase->read)(dbase, file, tbuffer, offset, tlen);
 426         if (code != tlen) {
 427             DBRELE(dbase);
 428             return UIOERROR;
 429         }
 430         code = rx_Write(rxcall, tbuffer, tlen);
 431         if (code != tlen) {
 432             DBRELE(dbase);
 433             return BULK_ERROR;
 434         }
 435         length -= tlen;
 436         offset += tlen;
 437     }
 438     code = (*dbase->getlabel)(dbase, file, version);    /* return the dbase, too */
 439     DBRELE(dbase);
 440     return code;
 441 }
 442
 443 SDISK_SendFile(rxcall, file, length, avers)
 444     register struct rx_call *rxcall;
 445     afs_int32 file;
 446     afs_int32 length;
 447     struct ubik_version *avers;
 448 {
 449     register afs_int32 code;
 450     register struct ubik_dbase *dbase;
 451     char tbuffer[256];
 452     afs_int32 offset, t;
 453     struct ubik_version tversion;
 454     register int tlen;
 455     struct rx_peer *tpeer;
 456     struct rx_connection *tconn;
 457     afs_uint32  otherHost;
 458
 459     /* send the file back to the requester */
 460
 461     if (code = ubik_CheckAuth(rxcall)) {
 462       goto failed;
 463     }
 464
 465     /* next, we do a sanity check to see if the guy sending us the database is
 466      * the guy we think is the sync site.  It turns out that we might not have
 467      * decided yet that someone's the sync site, but they could have enough
 468      * votes from others to be sync site anyway, and could send us the database
 469      * in advance of getting our votes.  This is fine, what we're really trying
 470      * to check is that some authenticated bogon isn't sending a random database
 471      * into another configuration.  This could happen on a bad configuration
 472      * screwup.  Thus, we only object if we're sure we know who the sync site
 473      * is, and it ain't the guy talking to us.
 474      */
 475     offset = uvote_GetSyncSite();
 476     tconn = rx_ConnectionOf(rxcall);
 477     tpeer = rx_PeerOf(tconn);
 478     otherHost = ubikGetPrimaryInterfaceAddr(rx_HostOf(tpeer));
 479     if (offset && offset != otherHost ) {
 480                         /* we *know* this is the wrong guy */
 481         code = USYNC;
 482         goto failed;
 483     }
 484
 485     dbase = ubik_dbase;
 486     DBHOLD(dbase);
 487
 488     /* abort any active trans that may scribble over the database */
 489     urecovery_AbortAll(dbase);
 490
 491     ubik_print("Ubik: Synchronize database with server %s\n",
 492                afs_inet_ntoa(otherHost));
 493
 494     offset = 0;
 495     (*dbase->truncate) (dbase, file, 0);                /* truncate first */
 496     tversion.epoch = 0; /* start off by labelling in-transit db as invalid */
 497     tversion.counter = 0;
 498     (*dbase->setlabel) (dbase, file, &tversion);        /* setlabel does sync */
 499     while (length > 0) {
 500         tlen = (length > sizeof(tbuffer) ? sizeof(tbuffer) : length);
 501         code = rx_Read(rxcall, tbuffer, tlen);
 502         if (code != tlen) {
 503             DBRELE(dbase);
 504             code = BULK_ERROR;
 505             goto failed;
 506         }
 507         code = (*dbase->write)(dbase, file, tbuffer, offset, tlen);
 508         if (code != tlen) {
 509             DBRELE(dbase);
 510             code = UIOERROR;
 511             goto failed;
 512         }
 513         offset += tlen;
 514         length -= tlen;
 515     }
 516
 517     /* sync data first, then write label and resync (resync done by setlabel call).
 518        This way, good label is only on good database. */
 519     (*ubik_dbase->sync)(dbase, file);
 520     code = (*ubik_dbase->setlabel)(dbase, file, avers);
 521     bcopy(avers, &ubik_dbase->version, sizeof(struct ubik_version));
 522     udisk_Invalidate(dbase, file); /* new dbase, flush disk buffers */
 523     LWP_NoYieldSignal(&dbase->version);
 524     DBRELE(dbase);
 525 failed:
 526     if (code) {
 527        ubik_print("Ubik: Synchronize database with server %s failed (error = %d)\n",
 528                   afs_inet_ntoa(otherHost), code);
 529     } else {
 530        ubik_print("Ubik: Synchronize database completed\n");
 531     }
 532     return code;
 533 }
 534
 535
 536 SDISK_Probe(rxcall)
 537     register struct rx_call *rxcall;
 538 {
 539     return 0;
 540 }
 541
 542 /*
 543 * Update remote machines addresses in my server list
 544 * Send back my addresses to caller of this RPC
 545 * Returns zero on success, else 1.
 546 */
 547 SDISK_UpdateInterfaceAddr(rxcall, inAddr, outAddr)
 548 register struct rx_call *rxcall;
 549 UbikInterfaceAddr       *inAddr, *outAddr;
 550 {
 551     struct ubik_server *ts, *tmp;
 552     afs_uint32 remoteAddr;              /* in net byte order */
 553     int    i, j, found=0, probableMatch=0;
 554
 555     /* copy the output parameters */
 556     for ( i=0; i < UBIK_MAX_INTERFACE_ADDR; i++)
 557         outAddr->hostAddr[i] = ntohl(ubik_host[i]);
 558
 559     remoteAddr = htonl(inAddr->hostAddr[0]);
 560     for(ts = ubik_servers; ts; ts=ts->next)
 561         if ( ts->addr[0] == remoteAddr ) /* both in net byte order */
 562         {
 563             probableMatch = 1;
 564             break;
 565         }
 566
 567     if ( probableMatch )
 568     {
 569         /* verify that all addresses in the incoming RPC are
 570         ** not part of other server entries in my CellServDB
 571         */
 572         for ( i=0; !found && (i<UBIK_MAX_INTERFACE_ADDR)
 573                                 && inAddr->hostAddr[i]; i++)
 574         {
 575             remoteAddr = htonl(inAddr->hostAddr[i]);
 576             for(tmp = ubik_servers; (!found && tmp); tmp=tmp->next)
 577             {
 578                 if ( ts == tmp )        /* this is my server */
 579                         continue;
 580                 for ( j=0; (j<UBIK_MAX_INTERFACE_ADDR) && tmp->addr[j]; j++)
 581                     if ( remoteAddr == tmp->addr[j] )
 582                     {
 583                         found = 1;
 584                         break;
 585                     }
 586             }
 587         }
 588     }           /* if (probableMatch) */
 589
 590     /* inconsistent addresses in CellServDB */
 591     if ( !probableMatch || found )
 592     {
 593         ubik_print("Inconsistent Cell Info from server: ");
 594         for ( i=0; i < UBIK_MAX_INTERFACE_ADDR && inAddr->hostAddr[i]; i++)
 595             printf("%s ", afs_inet_ntoa(htonl(inAddr->hostAddr[i])));
 596         printf("\n");
 597         printServerInfo();
 598         return UBADHOST;
 599     }
 600
 601     /* update our data structures */
 602     for ( i=1; i < UBIK_MAX_INTERFACE_ADDR; i++)
 603             ts->addr[i] = htonl(inAddr->hostAddr[i]);
 604
 605     ubik_print("ubik: A Remote Server has addresses: ");
 606     for ( i=0; i < UBIK_MAX_INTERFACE_ADDR && ts->addr[i]; i++)
 607         printf("%s ", afs_inet_ntoa(ts->addr[i]));
 608     printf("\n");
 609     fflush(stdout); fflush(stderr);
 610
 611     return 0;
 612 }
 613
 614 printServerInfo()
 615 {
 616     struct ubik_server *ts;
 617     int         i,j=1;
 618
 619     ubik_print("Local CellServDB:");
 620     for ( ts=ubik_servers; ts; ts= ts->next, j++)
 621     {
 622         printf("Server %d: ", j);
 623         for ( i=0; (i<UBIK_MAX_INTERFACE_ADDR) && ts->addr[i]; i++)
 624             printf("%s ", afs_inet_ntoa(ts->addr[i]));
 625     }
 626     printf("\n");
 627     fflush(stdout); fflush(stderr);
 628 }
 629
 630 SDISK_SetVersion(rxcall, atid, oldversionp, newversionp)
 631   struct rx_call      *rxcall;
 632   struct ubik_tid     *atid;
 633   struct ubik_version *oldversionp;
 634   struct ubik_version *newversionp;
 635 {
 636   afs_int32               code=0;
 637   struct ubik_dbase   *dbase;
 638
 639   if (code = ubik_CheckAuth(rxcall)) {
 640      return(code);
 641   }
 642
 643   if (!ubik_currentTrans) {
 644      return USYNC;
 645   }
 646   /* sanity check to make sure only write trans appear here */
 647   if (ubik_currentTrans->type != UBIK_WRITETRANS) {
 648      return UBADTYPE;
 649   }
 650
 651   /* Should not get this for the sync site */
 652   if (ubeacon_AmSyncSite()) {
 653      return UDEADLOCK;
 654   }
 655
 656   dbase = ubik_currentTrans->dbase;
 657   DBHOLD(dbase);
 658   urecovery_CheckTid(atid);
 659   if (!ubik_currentTrans) {
 660      DBRELE(dbase);
 661      return USYNC;
 662   }
 663
 664   /* Set the label if its version matches the sync-site's */
 665   if ((oldversionp->epoch   == ubik_dbVersion.epoch) &&
 666       (oldversionp->counter == ubik_dbVersion.counter)) {
 667      code = (*dbase->setlabel) (ubik_dbase, 0, newversionp);
 668      if (!code) {
 669         ubik_dbase->version = *newversionp;
 670         ubik_dbVersion      = *newversionp;
 671      }
 672   } else {
 673      code = USYNC;
 674   }
 675
 676   DBRELE(dbase);
 677   return code;
 678 }