← Back to branch summary

~ubuntu-branches/ubuntu/saucy/drizzle/saucy-proposed

« back to all changes in this revision

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

  • Committer: Package Import Robot
  • Author(s): Clint Byrum
  • Date: 2012-06-19 10:46:49 UTC
  • mfrom: (1.1.6)
  • mto: This revision was merged to the branch mainline in revision 29.
  • Revision ID: package-import@ubuntu.com-20120619104649-e2l0ggd4oz3um0f4
Tags: upstream-7.1.36-stable
Import upstream version 7.1.36-stable

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/pbms/src/compactor_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-10
23
 
 *
24
 
 * H&G2JCtL
25
 
 *
26
 
 * Network interface.
27
 
 *
28
 
 */
29
 
 
30
 
#include "cslib/CSConfig.h"
31
 
 
32
 
#include "defs_ms.h"
33
 
 
34
 
#include "cslib/CSGlobal.h"
35
 
#include "cslib/CSStrUtil.h"
36
 
#include "cslib/CSStorage.h"
37
 
 
38
 
#include "compactor_ms.h"
39
 
#include "open_table_ms.h"
40
 
#include "repository_ms.h"
41
 
#include "parameters_ms.h"
42
 
 
43
 
/*
44
 
 * ---------------------------------------------------------------
45
 
 * COMPACTOR THREAD
46
 
 */
47
 
 
48
 
MSCompactorThread::MSCompactorThread(time_t wait_time, MSDatabase *db):
49
 
CSDaemon(wait_time, NULL),
50
 
iCompactorDatabase(db)
51
 
{
52
 
}
53
 
 
54
 
void MSCompactorThread::close()
55
 
{
56
 
}
57
 
 
58
 
bool MSCompactorThread::doWork()
59
 
{
60
 
        bool                            complete;
61
 
        MSRepository            *src_repo, *dst_repo;
62
 
        MSRepoFile                      *src_file, *dst_file;
63
 
        uint32_t                        src_repo_id;
64
 
        MSBlobHeadRec           blob;
65
 
        off64_t                         src_offset;
66
 
        uint16_t                        head_size;
67
 
        uint64_t                        blob_size, blob_data_size;
68
 
        CSStringBuffer          *head;
69
 
        MSRepoPointersRec       ptr;
70
 
        uint32_t                                table_ref_count;
71
 
        uint32_t                                blob_ref_count;
72
 
        int                                     ref_count;
73
 
        size_t                          ref_size;
74
 
        CSMutex                         *mylock;
75
 
        uint32_t                        tab_id;
76
 
        uint64_t                        blob_id;
77
 
        MSOpenTable                     *otab;
78
 
        uint32_t                        repo_id;
79
 
        uint64_t                        repo_offset;
80
 
        uint64_t                        repo_blob_size;
81
 
        uint16_t                        repo_head_size;
82
 
        uint16_t                        tab_index;
83
 
        uint8_t                         status;
84
 
 
85
 
        enter_();
86
 
        retry:
87
 
        
88
 
#ifdef MS_COMPACTOR_POLLS
89
 
        if (!(src_repo = iCompactorDatabase->getRepoFullOfTrash(NULL)))
90
 
                return_(true);
91
 
#else
92
 
        myWaitTime = MS_DEFAULT_COMPACTOR_WAIT * 1000;  // Time in milli-seconds
93
 
        if (!(src_repo = iCompactorDatabase->getRepoFullOfTrash(&myWaitTime)))
94
 
                return_(true);
95
 
#endif
96
 
        frompool_(src_repo);
97
 
        src_file = src_repo->openRepoFile();
98
 
        push_(src_file);
99
 
 
100
 
        dst_repo = iCompactorDatabase->lockRepo(src_repo->myRepoFileSize - src_repo->myGarbageCount);
101
 
        frompool_(dst_repo);
102
 
        dst_file = dst_repo->openRepoFile();
103
 
        push_(dst_file);
104
 
 
105
 
        new_(head, CSStringBuffer(100));
106
 
        push_(head);
107
 
 
108
 
        complete = false;
109
 
        src_repo_id = src_repo->myRepoID;
110
 
        src_offset = src_repo->myRepoHeadSize;
111
 
        //printf("\nCompacting repo %"PRId32"\n\n", src_repo_id);
112
 
        // For testing:
113
 
        {
114
 
                int blockit = 0;
115
 
                if (blockit) {
116
 
                        release_(head);
117
 
                        release_(dst_file);
118
 
                        backtopool_(dst_repo);
119
 
                        release_(src_file);
120
 
                        backtopool_(src_repo);
121
 
 
122
 
                        myWaitTime = 5 * 1000;  // Time in milli-seconds
123
 
                        return_(true);
124
 
                }
125
 
        }
126
 
        while (src_offset < src_repo->myRepoFileSize) {                 
127
 
                retry_loop:
128
 
                suspended();
129
 
 
130
 
                if (myMustQuit)
131
 
                        goto quit;
132
 
                retry_read:
133
 
                
134
 
                // A lock is required here because references and dereferences to the
135
 
                // BLOBs can result in the repository record being updated while 
136
 
                // it is being copied.
137
 
                mylock = &src_repo->myRepoLock[src_offset % CS_REPO_REC_LOCK_COUNT];
138
 
                lock_(mylock);
139
 
                if (src_file->read(&blob, src_offset, src_repo->myRepoBlobHeadSize, 0) < src_repo->myRepoBlobHeadSize) {
140
 
                        unlock_(mylock);
141
 
                        break;
142
 
                }
143
 
                ref_size = CS_GET_DISK_1(blob.rb_ref_size_1);
144
 
                ref_count = CS_GET_DISK_2(blob.rb_ref_count_2);
145
 
                head_size = CS_GET_DISK_2(blob.rb_head_size_2);
146
 
                blob_size = CS_GET_DISK_6(blob.rb_blob_repo_size_6);
147
 
                blob_data_size = CS_GET_DISK_6(blob.rb_blob_data_size_6);
148
 
                status = CS_GET_DISK_1(blob.rb_status_1);
149
 
                if ((blob_data_size == 0) || ref_count <= 0 || ref_size == 0 ||
150
 
                        head_size < src_repo->myRepoBlobHeadSize + ref_count * ref_size ||
151
 
                        !VALID_BLOB_STATUS(status)) {
152
 
                        /* Can't be true. Assume this is garbage! */
153
 
                        unlock_(mylock);
154
 
                        src_offset++;
155
 
                        goto retry_read;
156
 
                }
157
 
                if (IN_USE_BLOB_STATUS(status)) {
158
 
                        head->setLength(head_size);
159
 
                        if (src_file->read(head->getBuffer(0), src_offset, head_size, 0) != head_size) {
160
 
                                unlock_(mylock);
161
 
                                break;
162
 
                        }
163
 
 
164
 
                        table_ref_count = 0;
165
 
                        blob_ref_count = 0;
166
 
                        
167
 
                        ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
168
 
                        for (int count = 0; count < ref_count; count++) {
169
 
                                switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
170
 
                                        case MS_BLOB_FREE_REF:
171
 
                                                break;
172
 
                                        case MS_BLOB_TABLE_REF:
173
 
                                                /* Check the reference: */
174
 
                                                tab_id = CS_GET_DISK_4(ptr.rp_tab_ref->tr_table_id_4);
175
 
                                                blob_id = CS_GET_DISK_6(ptr.rp_tab_ref->tr_blob_id_6);
176
 
 
177
 
                                                otab = MSTableList::getOpenTableByID(iCompactorDatabase->myDatabaseID, tab_id);
178
 
                                                if (otab) {
179
 
                                                        frompool_(otab);
180
 
                                                        /* Ignore the return value (it will fail because auth_code is wrong!)!! */
181
 
                                                        uint32_t auth_code = 0;
182
 
                                                        otab->getDBTable()->readBlobHandle(otab, blob_id, &auth_code, &repo_id, &repo_offset, &repo_blob_size, &repo_head_size, false);
183
 
                                                        backtopool_(otab);
184
 
                                                        if (repo_id == src_repo_id &&
185
 
                                                                repo_offset == src_offset &&
186
 
                                                                repo_blob_size == blob_data_size &&
187
 
                                                                repo_head_size == head_size)
188
 
                                                                table_ref_count++;
189
 
                                                        else
190
 
                                                                /* Remove the reference: */
191
 
                                                                CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
192
 
                                                }
193
 
                                                else
194
 
                                                        CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
195
 
                                                break;
196
 
                                        case MS_BLOB_DELETE_REF:
197
 
                                                /* These are temporary references from the TempLog file. */
198
 
                                                /* We try to prevent this from happening, but it can! */
199
 
                                                uint32_t                        temp_log_id;
200
 
                                                uint32_t                        temp_log_offset;
201
 
                                                MSTempLogFile   *temp_log;
202
 
 
203
 
                                                temp_log_id = CS_GET_DISK_4(ptr.rp_temp_ref->tp_log_id_4);
204
 
                                                temp_log_offset = CS_GET_DISK_4(ptr.rp_temp_ref->tp_offset_4);
205
 
                                                if ((temp_log = iCompactorDatabase->openTempLogFile(temp_log_id, NULL, NULL))) {
206
 
                                                        MSTempLogItemRec        log_item;
207
 
                                                        uint32_t                                then;
208
 
                                                        time_t                          now;
209
 
 
210
 
                                                        push_(temp_log);
211
 
                                                        if (temp_log->read(&log_item, temp_log_offset, sizeof(MSTempLogItemRec), 0) == sizeof(MSTempLogItemRec)) {
212
 
                                                                then = CS_GET_DISK_4(log_item.ti_time_4);
213
 
                                                                now = time(NULL);
214
 
                                                                if (now < (time_t)(then + PBMSParameters::getTempBlobTimeout())) {
215
 
                                                                        /* Wait for the BLOB to expire before we continue: */                                                                   
216
 
                                                                        release_(temp_log);
217
 
                                                                        unlock_(mylock);
218
 
 
219
 
                                                                        /* Go to sleep until the problem has gone away! */
220
 
                                                                        lock_(this);
221
 
                                                                        suspendedWait(MSTempLog::adjustWaitTime(then, now));
222
 
                                                                        unlock_(this);
223
 
                                                                        goto retry_loop;
224
 
                                                                }
225
 
                                                        }
226
 
                                                        release_(temp_log);
227
 
                                                }
228
 
 
229
 
                                                /* Remove the temp reference: */
230
 
                                                CS_SET_DISK_2(ptr.rp_ref->rr_type_2, MS_BLOB_FREE_REF);
231
 
                                                break;
232
 
                                        default:
233
 
                                                tab_index = CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2);
234
 
                                                if (tab_index > ref_count || !tab_index) {
235
 
                                                        /* Can't be true. Assume this is garbage! */
236
 
                                                        unlock_(mylock);
237
 
                                                        src_offset++;
238
 
                                                        goto retry_read;
239
 
                                                }
240
 
                                                blob_ref_count++;
241
 
                                                break;
242
 
                                }
243
 
                                ptr.rp_chars += ref_size;
244
 
                        }
245
 
 
246
 
                        if (table_ref_count && blob_ref_count) {
247
 
                                /* Check the blob references again to make sure that they
248
 
                                 * refer to valid table references.
249
 
                                 */
250
 
                                MSRepoTableRefPtr       tab_ref;
251
 
 
252
 
                                blob_ref_count = 0;
253
 
                                ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
254
 
                                for (int count = 0; count < ref_count; count++) {
255
 
                                        switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
256
 
                                                case MS_BLOB_FREE_REF:
257
 
                                                case MS_BLOB_TABLE_REF:
258
 
                                                case MS_BLOB_DELETE_REF:
259
 
                                                        break;
260
 
                                                default: // If it isn't one of the above we assume it is an blob ref. (er_table_2 can never have a value equal to one of the above REF type flags.)
261
 
                                                                // It was already verified above that the index was with in range.
262
 
                                                        tab_ref = (MSRepoTableRefPtr) (head->getBuffer(0) + src_repo->myRepoBlobHeadSize + (CS_GET_DISK_2(ptr.rp_blob_ref->er_table_2)-1) * ref_size);
263
 
                                                        if (CS_GET_DISK_2(tab_ref->rr_type_2) == MS_BLOB_TABLE_REF)
264
 
                                                                blob_ref_count++;
265
 
                                                        break;
266
 
                                        }
267
 
                                        ptr.rp_chars += ref_size;
268
 
                                }
269
 
                        }
270
 
 
271
 
                        if (blob_ref_count) {
272
 
                                off64_t dst_offset;
273
 
 
274
 
                                dst_offset = dst_repo->myRepoFileSize;
275
 
 
276
 
                                /* Write the header. */
277
 
                                dst_file->write(head->getBuffer(0), dst_offset, head_size);
278
 
 
279
 
                                /* We have an engine reference, copy the BLOB over: */
280
 
                                CSFile::transfer(RETAIN(dst_file), dst_offset + head_size, RETAIN(src_file), src_offset + head_size, blob_size, iCompactBuffer, MS_COMPACTOR_BUFFER_SIZE);
281
 
 
282
 
#ifdef HAVE_ALIAS_SUPPORT
283
 
                                /* If the BLOB has an alias update the alias index. */
284
 
                                if (CS_GET_DISK_2(blob.rb_alias_offset_2)) {
285
 
                                        iCompactorDatabase->moveBlobAlias( src_repo_id, src_offset, CS_GET_DISK_4(blob.rb_alias_hash_4), dst_repo->myRepoID, dst_offset);
286
 
                                }
287
 
#endif                          
288
 
                                /* Update the references: */
289
 
                                ptr.rp_chars = head->getBuffer(0) + src_repo->myRepoBlobHeadSize;
290
 
                                for (int count = 0; count < ref_count; count++) {
291
 
                                        switch (CS_GET_DISK_2(ptr.rp_ref->rr_type_2)) {
292
 
                                                case MS_BLOB_FREE_REF:
293
 
                                                        break;
294
 
                                                case MS_BLOB_TABLE_REF:
295
 
                                                        tab_id = CS_GET_DISK_4(ptr.rp_tab_ref->tr_table_id_4);
296
 
                                                        blob_id = CS_GET_DISK_6(ptr.rp_tab_ref->tr_blob_id_6);
297
 
 
298
 
                                                        if ((otab = MSTableList::getOpenTableByID(iCompactorDatabase->myDatabaseID, tab_id))) {
299
 
                                                                frompool_(otab);
300
 
                                                                otab->getDBTable()->updateBlobHandle(otab, blob_id, dst_repo->myRepoID, dst_offset, 0);
301
 
                                                                backtopool_(otab);
302
 
                                                        }
303
 
                                                        break;
304
 
                                                case MS_BLOB_DELETE_REF:
305
 
                                                        break;
306
 
                                                default:
307
 
                                                        break;
308
 
                                        }
309
 
                                        ptr.rp_chars += ref_size;
310
 
                                }
311
 
 
312
 
                                dst_repo->myRepoFileSize += head_size + blob_size;
313
 
                        }
314
 
                }
315
 
                
316
 
                unlock_(mylock);
317
 
                src_offset += head_size + blob_size;
318
 
        }
319
 
 
320
 
        src_repo->mustBeDeleted = true;
321
 
        complete = true;
322
 
 
323
 
        quit:
324
 
        release_(head);
325
 
        release_(dst_file);
326
 
        backtopool_(dst_repo);
327
 
        release_(src_file);
328
 
        backtopool_(src_repo);
329
 
 
330
 
        if (complete)
331
 
                iCompactorDatabase->removeRepo(src_repo_id, &myMustQuit);
332
 
 
333
 
        if (!myMustQuit)
334
 
                goto retry;
335
 
        return_(true);
336
 
}
337
 
 
338
 
void *MSCompactorThread::completeWork()
339
 
{
340
 
        close();
341
 
        return NULL;
342
 
}
343