← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/pbms/src/temp_log_ms.cc

  • Committer: Mark Atwood
  • Date: 2011-12-20 02:32:53 UTC
  • mfrom: (2469.1.1 drizzle-build)
  • Revision ID: me@mark.atwood.name-20111220023253-bvu0kr14kwsdvz7g
merge lp:~brianaker/drizzle/deprecate-pbms

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/pbms/src/temp_log_ms.cc
1
 
/* Copyright (C) 2008 PrimeBase Technologies GmbH, Germany
2
 
 *
3
 
 * PrimeBase Media Stream for MySQL
4
 
 *
5
 
 * This program is free software; you can redistribute it and/or modify
6
 
 * it under the terms of the GNU General Public License as published by
7
 
 * the Free Software Foundation; either version 2 of the License, or
8
 
 * (at your option) any later version.
9
 
 *
10
 
 * This program is distributed in the hope that it will be useful,
11
 
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12
 
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
 
 * GNU General Public License for more details.
14
 
 *
15
 
 * You should have received a copy of the GNU General Public License
16
 
 * along with this program; if not, write to the Free Software
17
 
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
18
 
 *
19
 
 * Original author: Paul McCullagh
20
 
 * Continued development: Barry Leslie
21
 
 *
22
 
 * 2007-07-03
23
 
 *
24
 
 * H&G2JCtL
25
 
 *
26
 
 * Network interface.
27
 
 *
28
 
 */
29
 
 
30
 
#include "cslib/CSConfig.h"
31
 
 
32
 
#include <stddef.h>
33
 
 
34
 
#include "defs_ms.h"
35
 
 
36
 
#include "cslib/CSGlobal.h"
37
 
#include "cslib/CSStrUtil.h"
38
 
#include "cslib/CSStorage.h"
39
 
 
40
 
#include "temp_log_ms.h"
41
 
#include "open_table_ms.h"
42
 
#include "trans_log_ms.h"
43
 
#include "transaction_ms.h"
44
 
#include "parameters_ms.h"
45
 
 
46
 
 
47
 
// Search the transaction log for a MS_ReferenceTxn record for the given BLOB.
48
 
// Just search the log file and not the cache. Seaching the cache may be faster but
49
 
// it would require locks that could block the writers or reader threads and in the worse
50
 
// case it will still require the reading of the log anyway.
51
 
//
52
 
// This search doesn't distinguish between transactions that are still running and
53
 
// transactions that are rolled back.
54
 
class SearchTXNLog : ReadTXNLog {
55
 
        public:
56
 
        SearchTXNLog(uint32_t db_id, MSTrans *log): ReadTXNLog(log), st_db_id(db_id) {}
57
 
        
58
 
        bool    st_found;
59
 
        bool    st_terminated;
60
 
        bool    st_commited;
61
 
        uint32_t st_tid; 
62
 
        uint32_t st_db_id; 
63
 
        uint32_t st_tab_id; 
64
 
        uint64_t st_blob_id;
65
 
        
66
 
        virtual bool rl_CanContinue() { return ((!st_found) || !st_terminated);}
67
 
        virtual void rl_Load(uint64_t log_position, MSTransPtr rec) 
68
 
        {
69
 
                (void) log_position;
70
 
                
71
 
                if ( !st_found && (TRANS_TYPE(rec->tr_type) != MS_ReferenceTxn))
72
 
                        return;
73
 
                
74
 
                if (!st_found) {
75
 
                        if  ((rec->tr_db_id == st_db_id) && (rec->tr_tab_id == st_tab_id) && (rec->tr_blob_id == st_blob_id)) {
76
 
                                st_found = true;
77
 
                                st_tid = rec->tr_id;
78
 
                        } else
79
 
                                return;
80
 
                }
81
 
                st_terminated = TRANS_IS_TERMINATED(rec->tr_type);
82
 
                if (st_terminated)
83
 
                        st_commited = (TRANS_IS_AUTOCOMMIT(rec->tr_type) || (TRANS_TYPE(rec->tr_type) == MS_CommitTxn));
84
 
        }
85
 
        
86
 
        bool st_FindBlobRef(bool *committed, uint32_t tab_id, uint64_t blob_id)
87
 
        {
88
 
                enter_();
89
 
                st_found = st_terminated = st_commited = false;
90
 
                st_tab_id = tab_id;
91
 
                st_blob_id = blob_id;   
92
 
                
93
 
                rl_ReadLog(rl_log->txn_GetStartPosition(), false);
94
 
                *committed = st_commited;
95
 
                return_(st_found);
96
 
        }
97
 
};
98
 
 
99
 
MSTempLogFile::MSTempLogFile():
100
 
CSReadBufferedFile(),
101
 
myTempLogID(0),
102
 
myTempLog(NULL)
103
 
{
104
 
}
105
 
 
106
 
MSTempLogFile::~MSTempLogFile()
107
 
{
108
 
        close();
109
 
        if (myTempLog)
110
 
                myTempLog->release();
111
 
}
112
 
 
113
 
MSTempLogFile *MSTempLogFile::newTempLogFile(uint32_t id, MSTempLog *temp_log, CSFile *file)
114
 
{
115
 
        MSTempLogFile *f;
116
 
        enter_();
117
 
        
118
 
        push_(temp_log);
119
 
        push_(file);
120
 
        
121
 
        if (!(f = new MSTempLogFile())) 
122
 
                CSException::throwOSError(CS_CONTEXT, ENOMEM);
123
 
 
124
 
        f->myTempLogID = id;
125
 
        
126
 
        pop_(file);
127
 
        f->setFile(file);
128
 
        
129
 
        pop_(temp_log);
130
 
        f->myTempLog = temp_log;
131
 
        return_(f);
132
 
}
133
 
 
134
 
MSTempLog::MSTempLog(uint32_t id, MSDatabase *db, off64_t file_size):
135
 
CSRefObject(),
136
 
myLogID(id),
137
 
myTempLogSize(file_size),
138
 
myTemplogRecSize(0),
139
 
myTempLogHeadSize(0),
140
 
iLogDatabase(db),
141
 
iDeleteLog(false)
142
 
{
143
 
}
144
 
 
145
 
MSTempLog::~MSTempLog()
146
 
{
147
 
        enter_();
148
 
        if (iDeleteLog) {
149
 
                CSPath *path;
150
 
 
151
 
                path = getLogPath();
152
 
                push_(path);
153
 
                path->removeFile();
154
 
                release_(path);
155
 
        }
156
 
        exit_();
157
 
}
158
 
 
159
 
void MSTempLog::deleteLog()
160
 
{
161
 
        iDeleteLog = true;
162
 
}
163
 
 
164
 
CSPath *MSTempLog::getLogPath()
165
 
{
166
 
        char file_name[120];
167
 
 
168
 
        cs_strcpy(120, file_name, "bs-logs");
169
 
        cs_add_dir_char(120, file_name);
170
 
        cs_strcat(120, file_name, "temp-");
171
 
        cs_strcat(120, file_name, myLogID);
172
 
        cs_strcat(120, file_name, ".bs");
173
 
        return CSPath::newPath(RETAIN(iLogDatabase->myDatabasePath), file_name);
174
 
}
175
 
 
176
 
MSTempLogFile *MSTempLog::openTempLog()
177
 
{
178
 
        CSPath                  *path;
179
 
        MSTempLogFile   *fh;
180
 
 
181
 
        enter_();
182
 
        path = getLogPath();
183
 
        retain();
184
 
        fh = MSTempLogFile::newTempLogFile(myLogID, this, CSFile::newFile(path));
185
 
        push_(fh);
186
 
        if (myTempLogSize)
187
 
                fh->open(CSFile::DEFAULT);
188
 
        else
189
 
                fh->open(CSFile::CREATE);
190
 
        if (!myTempLogHeadSize) {
191
 
                MSTempLogHeadRec        head;
192
 
 
193
 
                lock_(iLogDatabase->myTempLogArray);
194
 
                /* Check again after locking: */
195
 
                if (!myTempLogHeadSize) {
196
 
                        size_t rem;
197
 
 
198
 
                        if (fh->read(&head, 0, offsetof(MSTempLogHeadRec, th_reserved_4), 0) < offsetof(MSTempLogHeadRec, th_reserved_4)) {
199
 
                                CS_SET_DISK_4(head.th_magic_4, MS_TEMP_LOG_MAGIC);
200
 
                                CS_SET_DISK_2(head.th_version_2, MS_TEMP_LOG_VERSION);
201
 
                                CS_SET_DISK_2(head.th_head_size_2, MS_TEMP_LOG_HEAD_SIZE);
202
 
                                CS_SET_DISK_2(head.th_rec_size_2, sizeof(MSTempLogItemRec));
203
 
                                CS_SET_DISK_4(head.th_reserved_4, 0);
204
 
                                fh->write(&head, 0, sizeof(MSTempLogHeadRec));
205
 
                                fh->flush();
206
 
                        }
207
 
                        
208
 
                        /* Check the file header: */
209
 
                        if (CS_GET_DISK_4(head.th_magic_4) != MS_TEMP_LOG_MAGIC)
210
 
                                CSException::throwFileError(CS_CONTEXT, fh->getPathString(), CS_ERR_BAD_HEADER_MAGIC);
211
 
                        if (CS_GET_DISK_2(head.th_version_2) > MS_TEMP_LOG_VERSION)
212
 
                                CSException::throwFileError(CS_CONTEXT, fh->getPathString(), CS_ERR_VERSION_TOO_NEW);
213
 
 
214
 
                        /* Load the header details: */
215
 
                        myTempLogHeadSize = CS_GET_DISK_2(head.th_head_size_2);
216
 
                        myTemplogRecSize = CS_GET_DISK_2(head.th_rec_size_2);
217
 
 
218
 
                        /* File size, cannot be less than header size, adjust to correct offset: */
219
 
                        if (myTempLogSize < myTempLogHeadSize)
220
 
                                myTempLogSize = myTempLogHeadSize;
221
 
                        if ((rem = (myTempLogSize - myTempLogHeadSize) % myTemplogRecSize))
222
 
                                myTempLogSize += myTemplogRecSize - rem;
223
 
                }
224
 
                unlock_(iLogDatabase->myTempLogArray);
225
 
        }
226
 
        pop_(fh);
227
 
        return_(fh);
228
 
}
229
 
 
230
 
time_t MSTempLog::adjustWaitTime(time_t then, time_t now)
231
 
{
232
 
        time_t wait;
233
 
 
234
 
        if (now < (time_t)(then + PBMSParameters::getTempBlobTimeout())) {
235
 
                wait = ((then + PBMSParameters::getTempBlobTimeout() - now) * 1000);
236
 
                if (wait < 2000)
237
 
                        wait = 2000;
238
 
                else if (wait > 120 * 1000)
239
 
                        wait = 120 * 1000;
240
 
        }
241
 
        else
242
 
                wait = 1;
243
 
                        
244
 
        return wait;
245
 
}
246
 
 
247
 
/*
248
 
 * ---------------------------------------------------------------
249
 
 * TEMP LOG THREAD
250
 
 */
251
 
 
252
 
MSTempLogThread::MSTempLogThread(time_t wait_time, MSDatabase *db):
253
 
CSDaemon(wait_time, NULL),
254
 
iTempLogDatabase(db),
255
 
iTempLogFile(NULL),
256
 
iLogRecSize(0),
257
 
iLogOffset(0)
258
 
{
259
 
}
260
 
 
261
 
 
262
 
void MSTempLogThread::close()
263
 
{
264
 
        if (iTempLogFile) {
265
 
                iTempLogFile->release();
266
 
                iTempLogFile = NULL;
267
 
        }
268
 
}
269
 
 
270
 
bool MSTempLogThread::try_ReleaseBLOBReference(CSThread *self, CSStringBuffer *buffer, uint32_t tab_id, int type, uint64_t blob_id, uint32_t auth_code)
271
 
{
272
 
        volatile bool rtc = true;
273
 
        try_(a) {
274
 
                /* Release the BLOB reference. */
275
 
                MSOpenTable *otab;
276
 
 
277
 
                if (type == MS_TL_REPO_REF) {
278
 
                        MSRepoFile      *repo_file;
279
 
 
280
 
                        if ((repo_file = iTempLogDatabase->getRepoFileFromPool(tab_id, true))) {
281
 
                                frompool_(repo_file);
282
 
                                repo_file->checkBlob(buffer, blob_id, auth_code, iTempLogFile->myTempLogID, iLogOffset);
283
 
                                backtopool_(repo_file);
284
 
                        }
285
 
                }
286
 
                else {
287
 
                        if ((otab = MSTableList::getOpenTableByID(iTempLogDatabase->myDatabaseID, tab_id))) {
288
 
                                frompool_(otab);
289
 
                                if (type == MS_TL_BLOB_REF) {
290
 
                                        otab->checkBlob(buffer, blob_id, auth_code, iTempLogFile->myTempLogID, iLogOffset);
291
 
                                        backtopool_(otab);
292
 
                                }
293
 
                                else {
294
 
                                        ASSERT(type == MS_TL_TABLE_REF);
295
 
                                        if ((type == MS_TL_TABLE_REF) && otab->deleteReferences(iTempLogFile->myTempLogID, iLogOffset, &myMustQuit)) {
296
 
                                                /* Delete the file now... */
297
 
                                                MSTable                 *tab;
298
 
                                                CSPath                  *from_path;
299
 
                                                MSOpenTablePool *tab_pool;
300
 
 
301
 
                                                tab = otab->getDBTable();
302
 
                                                from_path = otab->getDBTable()->getTableFile();
303
 
 
304
 
                                                pop_(otab);
305
 
 
306
 
                                                push_(from_path);
307
 
                                                tab->retain();
308
 
                                                push_(tab);
309
 
 
310
 
                                                tab_pool = MSTableList::lockTablePoolForDeletion(otab); // This returns otab to the pool.
311
 
                                                frompool_(tab_pool);
312
 
 
313
 
                                                from_path->removeFile();
314
 
                                                tab->myDatabase->removeTable(tab);
315
 
 
316
 
                                                backtopool_(tab_pool); // The will unlock and close the table pool freeing all tables in it.
317
 
                                                pop_(tab);                              // Returning the pool will have released this. (YUK!)
318
 
                                                release_(from_path);
319
 
                                        }
320
 
                                        else 
321
 
                                                backtopool_(otab);
322
 
                                }
323
 
                        }
324
 
                }
325
 
                
326
 
                rtc = false;
327
 
        }
328
 
        
329
 
        catch_(a);
330
 
        cont_(a);
331
 
        return rtc;
332
 
}
333
 
 
334
 
bool MSTempLogThread::doWork()
335
 
{
336
 
        size_t                          tfer;
337
 
        MSTempLogItemRec        log_item;
338
 
        CSStringBuffer          *buffer;
339
 
        SearchTXNLog            txn_log(iTempLogDatabase->myDatabaseID, MSTransactionManager::tm_Log);
340
 
 
341
 
        enter_();
342
 
        new_(buffer, CSStringBuffer(20));
343
 
        push_(buffer);
344
 
        while (!myMustQuit) {
345
 
                if (!iTempLogFile) {
346
 
                        size_t head_size;
347
 
                        if (!(iTempLogFile = iTempLogDatabase->openTempLogFile(0, &iLogRecSize, &head_size))) {
348
 
                                release_(buffer);
349
 
                                return_(true);
350
 
                        }
351
 
                        iLogOffset = head_size;
352
 
                }
353
 
 
354
 
                tfer = iTempLogFile->read(&log_item, iLogOffset, sizeof(MSTempLogItemRec), 0);
355
 
                if (tfer == 0) {
356
 
                        /* No more data to be read: */
357
 
 
358
 
                        /* Check to see if there is a log after this: */
359
 
                        if (iTempLogDatabase->getTempLogCount() <= 1) {
360
 
                                /* The next log does not yet exist. We wait for
361
 
                                 * it to be created before we delete and
362
 
                                 * close the current log.
363
 
                                 */
364
 
                                myWaitTime = PBMSParameters::getTempBlobTimeout() * 1000;
365
 
                                break;
366
 
                        }
367
 
 
368
 
                        iTempLogFile->myTempLog->deleteLog();
369
 
                        iTempLogDatabase->removeTempLog(iTempLogFile->myTempLogID);
370
 
                        close();
371
 
                }
372
 
                else if (tfer == sizeof(MSTempLogItemRec)) {
373
 
                        /* We have a record: */
374
 
                        int             type;
375
 
                        uint32_t tab_id;
376
 
                        uint64_t blob_id= 0;
377
 
                        uint32_t auth_code;
378
 
                        uint32_t then;
379
 
                        time_t  now;
380
 
 
381
 
                        /*
382
 
                         * Items in the temp log are never updated.
383
 
                         * If a temp operation is canceled then the object 
384
 
                         * records this itself and when the temp operation 
385
 
                         * is attempted it will recognize by the templog
386
 
                         * id and offset that it is no longer a valid 
387
 
                         * operation.
388
 
                         */
389
 
                        tab_id = CS_GET_DISK_4(log_item.ti_table_id_4);
390
 
                                
391
 
                        type = CS_GET_DISK_1(log_item.ti_type_1);
392
 
                        blob_id = CS_GET_DISK_6(log_item.ti_blob_id_6);
393
 
                        auth_code = CS_GET_DISK_4(log_item.ti_auth_code_4);
394
 
                        then = CS_GET_DISK_4(log_item.ti_time_4);
395
 
 
396
 
                        now = time(NULL);
397
 
                        if (now < (time_t)(then + PBMSParameters::getTempBlobTimeout())) {
398
 
                                /* Time has not yet exired, adjust wait time: */
399
 
                                myWaitTime = MSTempLog::adjustWaitTime(then, now);
400
 
                                break;
401
 
                        }
402
 
                
403
 
                        if (try_ReleaseBLOBReference(self, buffer, tab_id, type, blob_id, auth_code)) {
404
 
                                int err = self->myException.getErrorCode();
405
 
                                
406
 
                                if (err == MS_ERR_TABLE_LOCKED) {
407
 
                                        throw_();
408
 
                                }
409
 
                                else if (err == MS_ERR_REMOVING_REPO) {
410
 
                                        /* Wait for the compactor to finish: */
411
 
                                        myWaitTime = 2 * 1000;
412
 
                                        release_(buffer);
413
 
                                        return_(true);
414
 
                                }
415
 
                                else if ((err == MS_ERR_UNKNOWN_TABLE) || (err == MS_ERR_DATABASE_DELETED))
416
 
                                        ;
417
 
                                else
418
 
                                        self->myException.log(NULL);
419
 
                        }
420
 
 
421
 
                }
422
 
                else {
423
 
                        // Only part of the data read, don't wait very long to try again:
424
 
                        myWaitTime = 2 * 1000;
425
 
                        break;
426
 
                }
427
 
                iLogOffset += iLogRecSize;
428
 
        }
429
 
 
430
 
        release_(buffer);
431
 
        return_(true);
432
 
}
433
 
 
434
 
void *MSTempLogThread::completeWork()
435
 
{
436
 
        close();
437
 
        return NULL;
438
 
}
439