src/budb/dbs_dump.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 <afsconfig.h>
  11 #include <afs/param.h>
  12
  13 RCSID
  14     ("$Header$");
  15
  16 #ifdef AFS_NT40_ENV
  17 #include <winsock2.h>
  18 #include <io.h>
  19 #include <fcntl.h>
  20 #else
  21 #include <netinet/in.h>
  22 #include <sys/time.h>
  23 #include <sys/resource.h>
  24 #include <sys/file.h>
  25 #endif
  26 #include <time.h>
  27 #include <sys/types.h>
  28 #include <afs/stds.h>
  29 #include <stdio.h>
  30 #include <lock.h>
  31 #include <ubik.h>
  32 #include <lwp.h>
  33 #include <rx/xdr.h>
  34 #include <rx/rx.h>
  35 #include <rx/rxkad.h>
  36 #include <string.h>
  37 #include <des.h>
  38 #include <afs/cellconfig.h>
  39 #include <errno.h>
  40 #include "budb.h"
  41 #include "budb_errs.h"
  42 #include "database.h"
  43 #include "error_macros.h"
  44 #include "globals.h"
  45 #include "afs/audit.h"
  46
  47 void *dumpWatcher(void *);
  48
  49 /* dump ubik database - interface routines */
  50
  51 /* badEntry
  52  *      no checking for now.
  53  */
  54
  55 afs_int32
  56 badEntry(dbAddr)
  57      afs_uint32 dbAddr;
  58 {
  59     /* return entry ok */
  60     return (0);
  61 }
  62
  63 /* setupDbDump
  64  *      decode the arguments passed via LWP and dump the database.
  65  */
  66
  67 void *
  68 setupDbDump(void *param)
  69 {
  70     int writeFid = (int)param;
  71     afs_int32 code = 0;
  72
  73     code = InitRPC(&dumpSyncPtr->ut, LOCKREAD, 1);
  74     if (code)
  75         goto error_exit;
  76
  77     code = writeDatabase(dumpSyncPtr->ut, writeFid);
  78     if (code)
  79         LogError(code, "writeDatabase failed\n");
  80
  81     code = close(writeFid);
  82     if (code)
  83         LogError(code, "pipe writer close failed\n");
  84
  85     LogDebug(5, "writeDatabase complete\n");
  86
  87   error_exit:
  88     if (dumpSyncPtr->ut)
  89         ubik_EndTrans(dumpSyncPtr->ut);
  90     return (void *)(code);
  91 }
  92
  93
  94 afs_int32 DumpDB(), RestoreDbHeader();
  95 afs_int32
  96 SBUDB_DumpDB(call, firstcall, maxLength, charListPtr, done)
  97      struct rx_call *call;
  98      int firstcall;
  99      afs_int32 maxLength;
 100      charListT *charListPtr;
 101      afs_int32 *done;
 102 {
 103     afs_int32 code;
 104
 105     code = DumpDB(call, firstcall, maxLength, charListPtr, done);
 106     osi_auditU(call, BUDB_DmpDBEvent, code, AUD_END);
 107     return code;
 108 }
 109
 110 afs_int32
 111 DumpDB(call, firstcall, maxLength, charListPtr, done)
 112      struct rx_call *call;
 113      int firstcall;             /* 1 - init.  0 - no init */
 114      afs_int32 maxLength;
 115      charListT *charListPtr;
 116      afs_int32 *done;
 117 {
 118 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 119     pthread_t dumperPid, watcherPid;
 120     pthread_attr_t dumperPid_tattr;
 121     pthread_attr_t watcherPid_tattr;
 122
 123 #else
 124     PROCESS dumperPid, watcherPid;
 125 #endif
 126     int readSize;
 127     afs_int32 code = 0;
 128
 129     if (callPermitted(call) == 0)
 130         ERROR(BUDB_NOTPERMITTED);
 131
 132     ObtainWriteLock(&dumpSyncPtr->ds_lock);
 133
 134     /* If asking for zero bytes, then this is a call to reset the timeToLive
 135      * timer. Reset it if there is a dump in progress.
 136      */
 137     if (maxLength == 0) {
 138         charListPtr->charListT_val = NULL;
 139         charListPtr->charListT_len = 0;
 140
 141         *done = ((dumpSyncPtr->statusFlags == 0) ? 1 : 0);
 142
 143         /* reset the clock on dump timeout */
 144         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 145         goto error_exit;
 146     }
 147
 148     if (dumpSyncPtr->statusFlags == 0) {
 149         if (!firstcall)
 150             ERROR(BUDB_DUMPFAILED);
 151
 152         LogDebug(5, "Setup dump\n");
 153
 154         /* no dump in progress - setup and retake lock */
 155         memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
 156 /*      ObtainWriteLock(&dumpSyncPtr->ds_lock); */
 157
 158         /* mark dump in progress */
 159         dumpSyncPtr->statusFlags = 1;
 160
 161         code = pipe(dumpSyncPtr->pipeFid);
 162         if (code)
 163             ERROR(errno);
 164
 165 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 166         /* Initialize the condition variables and the mutexes we use
 167          * to signal and synchronize the reader and writer threads.
 168          */
 169         assert(pthread_cond_init(&dumpSyncPtr->ds_readerStatus_cond, (const pthread_condattr_t *)0) == 0);
 170         assert(pthread_cond_init(&dumpSyncPtr->ds_writerStatus_cond, (const pthread_condattr_t *)0) == 0);
 171         assert(pthread_mutex_init(&dumpSyncPtr->ds_readerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
 172         assert(pthread_mutex_init(&dumpSyncPtr->ds_writerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
 173
 174         /* Initialize the thread attributes and launch the thread */
 175
 176         assert(pthread_attr_init(&dumperPid_tattr) == 0);
 177         assert(pthread_attr_setdetachstate(&dumperPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
 178         assert(pthread_create(&dumperPid, &dumperPid_tattr, (void *)setupDbDump, NULL) == 0);
 179
 180 #else
 181         code =
 182             LWP_CreateProcess(setupDbDump, 16384, 1,
 183                               (void *)dumpSyncPtr->pipeFid[1],
 184                               "Database Dumper", &dumperPid);
 185         if (code)
 186             goto error_exit;
 187 #endif
 188
 189         dumpSyncPtr->dumperPid = dumperPid;
 190         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 191
 192 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 193         /* Initialize the thread attributes and launch the thread */
 194
 195         assert(pthread_attr_init(&watcherPid_tattr) == 0);
 196         assert(pthread_attr_setdetachstate(&watcherPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
 197         assert(pthread_create(&watcherPid, &watcherPid_tattr, (void *)dumpWatcher, NULL) == 0);
 198 #else
 199         /* now create the watcher thread */
 200         code =
 201             LWP_CreateProcess(dumpWatcher, 16384, 1, 0,
 202                               "Database Dump Watchdog", &watcherPid);
 203 #endif
 204     } else if (firstcall)
 205         ERROR(BUDB_LOCKED);
 206
 207     /* now read the database and feed it to the rpc connection */
 208
 209     /* wait for data */
 210     while (dumpSyncPtr->ds_bytes == 0) {
 211         /* if no more data */
 212         if ((dumpSyncPtr->ds_writerStatus == DS_DONE)
 213             || (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR)) {
 214             break;
 215         }
 216
 217         if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
 218             LogDebug(6, "wakup writer\n");
 219             dumpSyncPtr->ds_writerStatus = 0;
 220 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 221             assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
 222 #else
 223             code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
 224             if (code)
 225                 LogError(code, "BUDB_DumpDB: signal delivery failed\n");
 226 #endif
 227         }
 228         LogDebug(6, "wait for writer\n");
 229         dumpSyncPtr->ds_readerStatus = DS_WAITING;
 230         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 231 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 232         assert(pthread_mutex_lock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
 233         assert(pthread_cond_wait(&dumpSyncPtr->ds_readerStatus_cond, &dumpSyncPtr->ds_readerStatus_mutex) == 0);
 234         assert(pthread_mutex_unlock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
 235 #else
 236         LWP_WaitProcess(&dumpSyncPtr->ds_readerStatus);
 237 #endif
 238         ObtainWriteLock(&dumpSyncPtr->ds_lock);
 239     }
 240
 241     charListPtr->charListT_val = (char *)malloc(maxLength);
 242     readSize =
 243         read(dumpSyncPtr->pipeFid[0], charListPtr->charListT_val, maxLength);
 244
 245     /* reset the clock on dump timeout */
 246     dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
 247
 248     LogDebug(4, "read of len %d returned %d\n", maxLength, readSize);
 249
 250     charListPtr->charListT_len = readSize;
 251
 252     if (readSize == 0) {        /* last chunk */
 253         *done = 1;
 254         close(dumpSyncPtr->pipeFid[0]);
 255         dumpSyncPtr->statusFlags = 0;
 256     } else
 257         *done = 0;
 258
 259     dumpSyncPtr->ds_bytes -= readSize;
 260     if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
 261         dumpSyncPtr->ds_writerStatus = 0;
 262 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 263         assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
 264 #else
 265         code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
 266         if (code)
 267             LogError(code, "BUDB_DumpDB: signal delivery failed\n");
 268 #endif
 269     }
 270
 271   error_exit:
 272     if (!code && (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR))
 273         code = -1;
 274     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 275     return (code);
 276 }
 277
 278 afs_int32
 279 SBUDB_RestoreDbHeader(call, header)
 280      struct rx_call *call;
 281      struct DbHeader *header;
 282 {
 283     afs_int32 code;
 284
 285     code = RestoreDbHeader(call, header);
 286     osi_auditU(call, BUDB_RstDBHEvent, code, AUD_END);
 287     return code;
 288 }
 289
 290 afs_int32
 291 RestoreDbHeader(call, header)
 292      struct rx_call *call;
 293      struct DbHeader *header;
 294 {
 295     struct ubik_trans *ut = 0;
 296     afs_int32 code = 0;
 297
 298     extern struct memoryDB db;
 299
 300     if (callPermitted(call) == 0)
 301         ERROR(BUDB_NOTPERMITTED);
 302
 303     code = InitRPC(&ut, LOCKWRITE, 1);
 304     if (code)
 305         goto error_exit;
 306
 307     if (header->dbversion != ntohl(db.h.version))
 308         ERROR(BUDB_VERSIONMISMATCH);
 309
 310     /* merge rather than replace the header information */
 311     if (db.h.lastDumpId < htonl(header->lastDumpId))
 312         db.h.lastDumpId = htonl(header->lastDumpId);
 313
 314     if (db.h.lastTapeId < htonl(header->lastTapeId))
 315         db.h.lastTapeId = htonl(header->lastTapeId);
 316
 317     if (db.h.lastInstanceId < htonl(header->lastInstanceId))
 318         db.h.lastInstanceId = htonl(header->lastInstanceId);
 319
 320     code = dbwrite(ut, 0, (char *)&db.h, sizeof(db.h));
 321     if (code)
 322         code = BUDB_IO;
 323
 324   error_exit:
 325     if (ut)
 326         ubik_EndTrans(ut);
 327     return (code);
 328 }
 329
 330 /* dumpWatcher
 331  *      monitors the state of a database dump. If the dump calls do not
 332  *      reset the time to live value, the dump times out. In that case,
 333  *      we kill the database traversal thread and clean up all the other
 334  *      state. Most importantly, the database is unlocked so that other
 335  *      transactions can proceed.
 336  */
 337
 338 void *
 339 dumpWatcher(void *unused)
 340 {
 341     afs_int32 code;
 342
 343     while (1) {                 /*w */
 344
 345         /* printf("dumpWatcher\n"); */
 346         ObtainWriteLock(&dumpSyncPtr->ds_lock);
 347
 348         if (dumpSyncPtr->statusFlags == 0) {
 349             /* dump has finished */
 350             goto exit;
 351         }
 352
 353         /* check time to live */
 354         if (time(0) > dumpSyncPtr->timeToLive) {        /*i */
 355             /* dump has exceeded the allocated time - terminate it */
 356             LogError(0, "Database dump timeout exceeded: %s",
 357                      ctime(&dumpSyncPtr->timeToLive));
 358             LogError(0, "Terminating database dump\n");
 359
 360             close(dumpSyncPtr->pipeFid[0]);
 361             close(dumpSyncPtr->pipeFid[1]);
 362 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 363             assert(pthread_cancel(dumpSyncPtr->dumperPid) == 0);
 364 #else
 365             code = LWP_DestroyProcess(dumpSyncPtr->dumperPid);
 366             if (code)
 367                 LogError(code, "dumpWatcher: failed to kill dump thread\n");
 368 #endif
 369
 370             if (dumpSyncPtr->ut) {
 371                 code = ubik_AbortTrans(dumpSyncPtr->ut);
 372                 if (code)
 373                     LogError(code, "Aborting dump transaction\n");
 374             }
 375
 376             memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
 377             goto exit;
 378         }
 379         /*i */
 380         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 381 #if defined(AFS_PTHREAD_ENV) && defined(UBIK_PTHREAD_ENV)
 382         sleep(5);
 383 #else
 384         IOMGR_Sleep(5);
 385 #endif
 386     }                           /*w */
 387
 388   exit:
 389     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
 390     /* printf("dumpWatcher exit\n"); */
 391     return (0);
 392 }