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