DEVEL15-kill-ubik-pthread-env-20080718
[openafs.git] / 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 #ifdef AFS_PTHREAD_ENV
119     pthread_t dumperPid, watcherPid;
120     pthread_attr_t dumperPid_tattr;
121     pthread_attr_t watcherPid_tattr;
122 #else
123     PROCESS dumperPid, watcherPid;
124 #endif
125     int readSize;
126     afs_int32 code = 0;
127
128     if (callPermitted(call) == 0)
129         ERROR(BUDB_NOTPERMITTED);
130
131     ObtainWriteLock(&dumpSyncPtr->ds_lock);
132
133     /* If asking for zero bytes, then this is a call to reset the timeToLive
134      * timer. Reset it if there is a dump in progress. 
135      */
136     if (maxLength == 0) {
137         charListPtr->charListT_val = NULL;
138         charListPtr->charListT_len = 0;
139
140         *done = ((dumpSyncPtr->statusFlags == 0) ? 1 : 0);
141
142         /* reset the clock on dump timeout */
143         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
144         goto error_exit;
145     }
146
147     if (dumpSyncPtr->statusFlags == 0) {
148         if (!firstcall)
149             ERROR(BUDB_DUMPFAILED);
150
151         LogDebug(5, "Setup dump\n");
152
153         /* no dump in progress - setup and retake lock */
154         memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
155 /*      ObtainWriteLock(&dumpSyncPtr->ds_lock); */
156
157         /* mark dump in progress */
158         dumpSyncPtr->statusFlags = 1;
159
160         code = pipe(dumpSyncPtr->pipeFid);
161         if (code)
162             ERROR(errno);
163
164 #ifdef AFS_PTHREAD_ENV
165         /* Initialize the condition variables and the mutexes we use
166          * to signal and synchronize the reader and writer threads.
167          */
168         assert(pthread_cond_init(&dumpSyncPtr->ds_readerStatus_cond, (const pthread_condattr_t *)0) == 0);
169         assert(pthread_cond_init(&dumpSyncPtr->ds_writerStatus_cond, (const pthread_condattr_t *)0) == 0);
170         assert(pthread_mutex_init(&dumpSyncPtr->ds_readerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
171         assert(pthread_mutex_init(&dumpSyncPtr->ds_writerStatus_mutex, (const pthread_mutexattr_t *)0) == 0);
172
173         /* Initialize the thread attributes and launch the thread */
174
175         assert(pthread_attr_init(&dumperPid_tattr) == 0);
176         assert(pthread_attr_setdetachstate(&dumperPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
177         assert(pthread_create(&dumperPid, &dumperPid_tattr, (void *)setupDbDump, NULL) == 0);
178
179 #else
180         code =
181             LWP_CreateProcess(setupDbDump, 16384, 1,
182                               (void *)dumpSyncPtr->pipeFid[1],
183                               "Database Dumper", &dumperPid);
184         if (code)
185             goto error_exit;
186 #endif
187
188         dumpSyncPtr->dumperPid = dumperPid;
189         dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
190
191 #ifdef AFS_PTHREAD_ENV
192         /* Initialize the thread attributes and launch the thread */
193
194         assert(pthread_attr_init(&watcherPid_tattr) == 0);
195         assert(pthread_attr_setdetachstate(&watcherPid_tattr, PTHREAD_CREATE_DETACHED) == 0);
196         assert(pthread_create(&watcherPid, &watcherPid_tattr, (void *)dumpWatcher, NULL) == 0);
197 #else
198         /* now create the watcher thread */
199         code =
200             LWP_CreateProcess(dumpWatcher, 16384, 1, 0,
201                               "Database Dump Watchdog", &watcherPid);
202 #endif
203     } else if (firstcall)
204         ERROR(BUDB_LOCKED);
205
206     /* now read the database and feed it to the rpc connection */
207
208     /* wait for data */
209     while (dumpSyncPtr->ds_bytes == 0) {
210         /* if no more data */
211         if ((dumpSyncPtr->ds_writerStatus == DS_DONE)
212             || (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR)) {
213             break;
214         }
215
216         if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
217             LogDebug(6, "wakup writer\n");
218             dumpSyncPtr->ds_writerStatus = 0;
219 #ifdef AFS_PTHREAD_ENV
220             assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
221 #else
222             code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
223             if (code)
224                 LogError(code, "BUDB_DumpDB: signal delivery failed\n");
225 #endif
226         }
227         LogDebug(6, "wait for writer\n");
228         dumpSyncPtr->ds_readerStatus = DS_WAITING;
229         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
230 #ifdef AFS_PTHREAD_ENV
231         assert(pthread_mutex_lock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
232         assert(pthread_cond_wait(&dumpSyncPtr->ds_readerStatus_cond, &dumpSyncPtr->ds_readerStatus_mutex) == 0);
233         assert(pthread_mutex_unlock(&dumpSyncPtr->ds_readerStatus_mutex) == 0);
234 #else
235         LWP_WaitProcess(&dumpSyncPtr->ds_readerStatus);
236 #endif
237         ObtainWriteLock(&dumpSyncPtr->ds_lock);
238     }
239
240     charListPtr->charListT_val = (char *)malloc(maxLength);
241     readSize =
242         read(dumpSyncPtr->pipeFid[0], charListPtr->charListT_val, maxLength);
243
244     /* reset the clock on dump timeout */
245     dumpSyncPtr->timeToLive = time(0) + DUMP_TTL_INC;
246
247     LogDebug(4, "read of len %d returned %d\n", maxLength, readSize);
248
249     charListPtr->charListT_len = readSize;
250
251     if (readSize == 0) {        /* last chunk */
252         *done = 1;
253         close(dumpSyncPtr->pipeFid[0]);
254         dumpSyncPtr->statusFlags = 0;
255     } else
256         *done = 0;
257
258     dumpSyncPtr->ds_bytes -= readSize;
259     if (dumpSyncPtr->ds_writerStatus == DS_WAITING) {
260         dumpSyncPtr->ds_writerStatus = 0;
261 #ifdef AFS_PTHREAD_ENV
262         assert(pthread_cond_broadcast(&dumpSyncPtr->ds_writerStatus_cond) == 0);
263 #else
264         code = LWP_SignalProcess(&dumpSyncPtr->ds_writerStatus);
265         if (code)
266             LogError(code, "BUDB_DumpDB: signal delivery failed\n");
267 #endif
268     }
269
270   error_exit:
271     if (!code && (dumpSyncPtr->ds_writerStatus == DS_DONE_ERROR))
272         code = -1;
273     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
274     return (code);
275 }
276
277 afs_int32
278 SBUDB_RestoreDbHeader(call, header)
279      struct rx_call *call;
280      struct DbHeader *header;
281 {
282     afs_int32 code;
283
284     code = RestoreDbHeader(call, header);
285     osi_auditU(call, BUDB_RstDBHEvent, code, AUD_END);
286     return code;
287 }
288
289 afs_int32
290 RestoreDbHeader(call, header)
291      struct rx_call *call;
292      struct DbHeader *header;
293 {
294     struct ubik_trans *ut = 0;
295     afs_int32 code = 0;
296
297     extern struct memoryDB db;
298
299     if (callPermitted(call) == 0)
300         ERROR(BUDB_NOTPERMITTED);
301
302     code = InitRPC(&ut, LOCKWRITE, 1);
303     if (code)
304         goto error_exit;
305
306     if (header->dbversion != ntohl(db.h.version))
307         ERROR(BUDB_VERSIONMISMATCH);
308
309     /* merge rather than replace the header information */
310     if (db.h.lastDumpId < htonl(header->lastDumpId))
311         db.h.lastDumpId = htonl(header->lastDumpId);
312
313     if (db.h.lastTapeId < htonl(header->lastTapeId))
314         db.h.lastTapeId = htonl(header->lastTapeId);
315
316     if (db.h.lastInstanceId < htonl(header->lastInstanceId))
317         db.h.lastInstanceId = htonl(header->lastInstanceId);
318
319     code = dbwrite(ut, 0, (char *)&db.h, sizeof(db.h));
320     if (code)
321         code = BUDB_IO;
322
323   error_exit:
324     if (ut)
325         ubik_EndTrans(ut);
326     return (code);
327 }
328
329 /* dumpWatcher
330  *      monitors the state of a database dump. If the dump calls do not 
331  *      reset the time to live value, the dump times out. In that case,
332  *      we kill the database traversal thread and clean up all the other 
333  *      state. Most importantly, the database is unlocked so that other
334  *      transactions can proceed.
335  */
336
337 void *
338 dumpWatcher(void *unused)
339 {
340     afs_int32 code;
341
342     while (1) {                 /*w */
343
344         /* printf("dumpWatcher\n"); */
345         ObtainWriteLock(&dumpSyncPtr->ds_lock);
346
347         if (dumpSyncPtr->statusFlags == 0) {
348             /* dump has finished */
349             goto exit;
350         }
351
352         /* check time to live */
353         if (time(0) > dumpSyncPtr->timeToLive) {        /*i */
354             /* dump has exceeded the allocated time - terminate it */
355             LogError(0, "Database dump timeout exceeded: %s",
356                      ctime(&dumpSyncPtr->timeToLive));
357             LogError(0, "Terminating database dump\n");
358
359             close(dumpSyncPtr->pipeFid[0]);
360             close(dumpSyncPtr->pipeFid[1]);
361 #ifdef AFS_PTHREAD_ENV
362             assert(pthread_cancel(dumpSyncPtr->dumperPid) == 0);
363 #else
364             code = LWP_DestroyProcess(dumpSyncPtr->dumperPid);
365             if (code)
366                 LogError(code, "dumpWatcher: failed to kill dump thread\n");
367 #endif
368
369             if (dumpSyncPtr->ut) {
370                 code = ubik_AbortTrans(dumpSyncPtr->ut);
371                 if (code)
372                     LogError(code, "Aborting dump transaction\n");
373             }
374
375             memset(dumpSyncPtr, 0, sizeof(*dumpSyncPtr));
376             goto exit;
377         }
378         /*i */
379         ReleaseWriteLock(&dumpSyncPtr->ds_lock);
380 #ifdef AFS_PTHREAD_ENV
381         sleep(5);
382 #else
383         IOMGR_Sleep(5);
384 #endif
385     }                           /*w */
386
387   exit:
388     ReleaseWriteLock(&dumpSyncPtr->ds_lock);
389     /* printf("dumpWatcher exit\n"); */
390     return (0);
391 }