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